diff --git a/.github/workflows/benchmark-tmpl.yml b/.github/workflows/benchmark-tmpl.yml index 46a80c874..d49d1a608 100644 --- a/.github/workflows/benchmark-tmpl.yml +++ b/.github/workflows/benchmark-tmpl.yml @@ -129,6 +129,11 @@ on: required: false type: string default: '3600' + require-power: + description: "Fail fixed-sequence result processing when GPU power is invalid" + required: false + type: boolean + default: false agentx-fast: description: "Use one warmup request per AgentX lane and a 20-minute profile" required: false @@ -179,6 +184,7 @@ env: KV_P2P_TRANSFER: ${{ inputs.kv-p2p-transfer }} TOTAL_CPU_DRAM_GB: ${{ inputs.total-cpu-dram-gb }} DURATION: ${{ inputs.duration }} + REQUIRE_POWER: ${{ inputs.require-power && '1' || '0' }} AIPERF_EXPERIMENTAL_FAST: ${{ inputs.agentx-fast && '1' || '0' }} EVAL_LIMIT: ${{ inputs.eval-limit }} SWEBENCH_GEN_MODE: ${{ inputs.swebench-gen-mode }} @@ -359,6 +365,18 @@ jobs: path: gpu_metrics.csv if-no-files-found: ignore + - name: Upload power audit bundle + if: ${{ always() && !inputs.eval-only && inputs.scenario-type != 'agentic-coding' }} + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: power_audit_${{ env.RESULT_FILENAME }} + path: | + ${{ env.RESULT_FILENAME }}.json + agg_${{ env.RESULT_FILENAME }}.json + gpu_metrics.csv + power_validation_${{ env.RESULT_FILENAME }}.json + if-no-files-found: ignore + - name: Upload eval results (if any) if: ${{ always() && (env.RUN_EVAL == 'true' || inputs.eval-only) }} uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 diff --git a/.github/workflows/e2e-tests.yml b/.github/workflows/e2e-tests.yml index 06604b2f7..004f7db45 100644 --- a/.github/workflows/e2e-tests.yml +++ b/.github/workflows/e2e-tests.yml @@ -25,6 +25,11 @@ on: required: false type: string default: "" + require-power: + description: "Fail single-node fixed-sequence jobs when GPU power is invalid" + required: false + type: boolean + default: false agentx-fast: description: "AgentX fast feedback: one warmup request per lane and a 20-minute profile." required: false @@ -95,6 +100,11 @@ on: required: false type: string default: "" + require-power: + description: "Fail single-node fixed-sequence jobs when GPU power is invalid" + required: false + type: boolean + default: false agentx-fast: description: "AgentX fast feedback: one warmup request per lane and a 20-minute profile." required: false @@ -531,6 +541,7 @@ jobs: spec-decoding: ${{ matrix.config.spec-decoding }} disagg: ${{ matrix.config.disagg }} run-eval: false + require-power: ${{ inputs.require-power }} ref: ${{ inputs.ref }} test-sweep-evals: diff --git a/.github/workflows/test-process-result.yml b/.github/workflows/test-process-result.yml index 833e60252..d5579238b 100644 --- a/.github/workflows/test-process-result.yml +++ b/.github/workflows/test-process-result.yml @@ -3,7 +3,14 @@ name: Test Process Result on: pull_request: paths: + - '.github/workflows/benchmark-tmpl.yml' + - '.github/workflows/e2e-tests.yml' + - '.github/workflows/test-process-result.yml' + - 'benchmarks/benchmark_lib.sh' + - 'utils/aggregate_power.py' + - 'utils/bench_serving/benchmark_serving.py' - 'utils/process_result.py' + - 'utils/test_aggregate_power.py' - 'utils/test_process_result.py' permissions: @@ -11,7 +18,6 @@ permissions: jobs: test: - if: github.event.pull_request.draft != true runs-on: ubuntu-latest permissions: contents: read @@ -33,4 +39,4 @@ jobs: - name: Run pytest run: | cd utils - pytest test_process_result.py -v + python -m pytest test_aggregate_power.py test_process_result.py -v diff --git a/AGENTS.md b/AGENTS.md index 9e93f49f6..9544cc087 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -42,7 +42,7 @@ Run `ls` for details. Key paths: - `utils/matrix_logic/` - `generate_sweep_configs.py`, `validation.py` Pydantic schemas, tests. - `utils/bench_serving/` - `benchmark_serving.py` and backends. - `utils/evals/` - lm-eval task configs, thresholds, `validate_scores.py` (see `EVALS.md`). -- `utils/` - `process_result.py`, `process_changelog.py` (incl. `trim_conc`), `collect_*.py`, `compare_results.py`. +- `utils/` - `process_result.py`, `process_changelog.py` (incl. `trim_conc`), `aggregate_power.py` (single-node GPU energy validation + aggregation), `collect_*.py`, `compare_results.py`. - `experimental/` - non-core experiments. ## Terminology @@ -117,7 +117,7 @@ gh api -X POST \ -f 'inputs[duration-override]=' ``` -Inputs: top-level `ref` (required) is the workflow ref to dispatch from, almost always `main`. `inputs[ref]` is the repo ref under test (defaults to the dispatch ref's `github.sha`). `inputs[generate-cli-command]` (required) is passed verbatim to `generate_sweep_configs.py` - test locally first. `inputs[test-name]` is the display name in the Actions UI. `inputs[duration-override]` overrides per-config duration (seconds); empty = use matrix value. +Inputs: top-level `ref` (required) is the workflow ref to dispatch from, almost always `main`. `inputs[ref]` is the repo ref under test (defaults to the dispatch ref's `github.sha`). `inputs[generate-cli-command]` (required) is passed verbatim to `generate_sweep_configs.py` - test locally first. `inputs[test-name]` is the display name in the Actions UI. `inputs[duration-override]` overrides per-config duration (seconds); empty = use matrix value. `inputs[require-power]` fails single-node fixed-sequence jobs when GPU power telemetry is invalid (default: power is best-effort and never fails a run). For an AgentX preflight, add `-F 'inputs[agentx-fast]=true'` to the dispatch command. This reduces deterministic warmup to one request per lane and profiling to 20 minutes for every AgentX job. Use it to get faster feedback while debugging, then run the official sweep without `agentx-fast`; canonical AgentX warmup remains 10 requests per lane with a 1-hour profile. @@ -200,6 +200,8 @@ cat ./results/agg_bmk.json | jq '[.[] | select(.infmax_model_prefix == "gptoss") `tput_per_gpu` (total throughput per GPU, tok/s), `output_tput_per_gpu` (output token throughput), `mean_ttft` / `p99_ttft` (time to first token), `mean_tpot` (time per output token), `mean_e2el` (end-to-end latency). +Single-node fixed-sequence results also carry GPU power metrics when telemetry is valid: `power_valid` (1/0), `avg_power_w` (average board power per GPU), `avg_total_gpu_power_w` (all observed GPUs), `total_gpu_energy_j` (integrated over the formal benchmark window), and `joules_per_successful_query` / `joules_per_input_token` / `joules_per_output_token` / `joules_per_total_token`. Invalid telemetry records `power_valid: 0` and no energy metrics; it fails the job only under `REQUIRE_POWER=1` (see the `require-power` dispatch input). + ### Artifacts -`results_bmk` → `agg_bmk.json` (aggregated). `results_all` → all results aggregated (may not exist). `eval_results_all` → `agg_eval_all.json` (may not exist). `run-stats` → `run_stats.json` (which nodes ran and succeeded). +`results_bmk` → `agg_bmk.json` (aggregated). `results_all` → all results aggregated (may not exist). `eval_results_all` → `agg_eval_all.json` (may not exist). `run-stats` → `run_stats.json` (which nodes ran and succeeded). `power_audit_` → `power_validation_.json` (canonical power validity verdict + reason codes; uploaded even when invalid). diff --git a/benchmarks/benchmark_lib.sh b/benchmarks/benchmark_lib.sh index c9816fc9d..ba6d23b10 100644 --- a/benchmarks/benchmark_lib.sh +++ b/benchmarks/benchmark_lib.sh @@ -109,7 +109,9 @@ unset _benchmark_caller # -------------------------------- GPU_MONITOR_PID="" -GPU_METRICS_CSV="/workspace/gpu_metrics.csv" +GPU_MONITOR_VENDOR="" +GPU_METRICS_CSV="${GPU_METRICS_CSV:-gpu_metrics.csv}" +NVIDIA_GPU_MONITOR_QUERY="timestamp,index,power.draw,temperature.gpu,clocks.current.sm,clocks.current.memory,utilization.gpu,utilization.memory" export GPU_METRICS_CSV # Start background GPU monitoring that logs metrics every second to CSV. @@ -131,11 +133,13 @@ start_gpu_monitor() { export GPU_METRICS_CSV if command -v nvidia-smi &>/dev/null; then - nvidia-smi --query-gpu=timestamp,index,power.draw,temperature.gpu,clocks.current.sm,clocks.current.memory,utilization.gpu,utilization.memory \ + GPU_MONITOR_VENDOR="nvidia" + nvidia-smi --query-gpu="$NVIDIA_GPU_MONITOR_QUERY" \ --format=csv -l "$interval" > "$output" 2>/dev/null & GPU_MONITOR_PID=$! echo "[GPU Monitor] Started NVIDIA (PID=$GPU_MONITOR_PID, interval=${interval}s, output=$output)" elif command -v amd-smi &>/dev/null; then + GPU_MONITOR_VENDOR="amd" # Use amd-smi native watch mode (-w) which includes timestamps automatically. # Pipe through awk to: skip preamble lines, keep first CSV header, skip repeated headers. amd-smi metric -p -c -t -u -w "$interval" --csv 2>/dev/null \ @@ -143,6 +147,7 @@ start_gpu_monitor() { GPU_MONITOR_PID=$! echo "[GPU Monitor] Started AMD (PID=$GPU_MONITOR_PID, interval=${interval}s, output=$output)" else + GPU_MONITOR_VENDOR="" echo "[GPU Monitor] No GPU monitoring tool found (nvidia-smi or amd-smi), skipping" return 0 fi @@ -153,6 +158,33 @@ stop_gpu_monitor() { if [[ -n "$GPU_MONITOR_PID" ]] && kill -0 "$GPU_MONITOR_PID" 2>/dev/null; then kill "$GPU_MONITOR_PID" 2>/dev/null wait "$GPU_MONITOR_PID" 2>/dev/null || true + # benchmark_end_time_unix is recorded shortly before the benchmark + # process exits. For the NVIDIA PR1 canary, append one post-exit sample + # so boundary interpolation is deterministic even when the run ends + # between 1 Hz monitor ticks. AMD one-shot CSV schemas vary by amd-smi + # version, so strict AMD lifecycle validation remains follow-up work. + case "$GPU_MONITOR_VENDOR" in + nvidia) + local append_final_nvidia_sample=true + local repaired_metrics="${GPU_METRICS_CSV}.repair.$$" + if [[ -s "$GPU_METRICS_CSV" ]] && + ! tail -c 1 "$GPU_METRICS_CSV" | grep -q '^$'; then + if sed '$d' "$GPU_METRICS_CSV" > "$repaired_metrics" && + mv "$repaired_metrics" "$GPU_METRICS_CSV"; then + echo "[GPU Monitor] Dropped truncated trailing sample" + else + rm -f "$repaired_metrics" + append_final_nvidia_sample=false + echo "[GPU Monitor] Warning: could not repair truncated trailing sample" >&2 + fi + fi + if [[ "$append_final_nvidia_sample" == true ]]; then + nvidia-smi --query-gpu="$NVIDIA_GPU_MONITOR_QUERY" \ + --format=csv,noheader >> "$GPU_METRICS_CSV" 2>/dev/null || + echo "[GPU Monitor] Warning: final NVIDIA sample failed" >&2 + fi + ;; + esac echo "[GPU Monitor] Stopped (PID=$GPU_MONITOR_PID)" if [[ -f "$GPU_METRICS_CSV" ]]; then local lines @@ -161,6 +193,7 @@ stop_gpu_monitor() { fi fi GPU_MONITOR_PID="" + GPU_MONITOR_VENDOR="" } # Block until the GPUs have released a prior job's memory before starting a run. diff --git a/utils/aggregate_power.py b/utils/aggregate_power.py index 3c204085a..a269fc509 100644 --- a/utils/aggregate_power.py +++ b/utils/aggregate_power.py @@ -1,44 +1,87 @@ -"""Aggregate measured GPU power from a vendor SMI CSV into the agg result JSON. - -Reads a GPU-metrics CSV produced by `start_gpu_monitor` (nvidia-smi or amd-smi), -filters samples to the benchmark load window using start/end Unix timestamps -written by benchmark_serving.py, and patches two keys into the aggregated -result JSON consumed by InferenceX-app's ETL: - - - avg_power_w: mean per-GPU power draw (W) during the load window - - joules_per_output_token: (avg_power_w * num_gpus * duration_s) / total_output_tokens - -The ETL (`packages/db/src/etl/benchmark-mapper.ts`) auto-captures any numeric -field in the agg JSON into the `metrics` JSONB column, so no schema migration -is required. - -Vendor schema detection is regex-based: any timestamp-like column + any column -whose name contains "power" (excluding "limit"/"cap"/"max") is picked up. -NVIDIA emits "power.draw [W]"; AMD's amd-smi varies by version. Both are -handled. - -This script is best-effort. Missing or malformed CSV exits 0 without patching -so a monitoring hiccup never breaks the benchmark upload. +"""Validate and aggregate single-node GPU-board power into benchmark results. + +The formal benchmark window comes from ``benchmark_serving.py``. Power is +integrated independently for every observed GPU with trapezoidal integration +and linear interpolation at the window boundaries. Unqualified energy metrics +therefore always describe the whole deployment, never a prefill/decode role. + +Ordinary benchmark runs are best-effort: invalid telemetry is recorded in the +aggregate and a validation sidecar, but does not fail the benchmark. Power +studies can set ``REQUIRE_POWER=1`` to fail after those audit artifacts exist. +The aggregate carries numeric ``power_valid`` (1/0) for metric ingestion; the +sidecar is the canonical source for boolean validity and reason codes. """ from __future__ import annotations import argparse +import bisect import csv import json +import math +import os import re import sys +from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from statistics import mean - _POWER_COL_RE = re.compile(r"power", re.IGNORECASE) _POWER_EXCLUDE_RE = re.compile(r"limit|cap|max|min", re.IGNORECASE) _TIMESTAMP_COL_RE = re.compile(r"time", re.IGNORECASE) _GPU_INDEX_COL_RE = re.compile(r"^(index|gpu|gpu_id|gpu_index|card|device)$", re.IGNORECASE) _NUMBER_RE = re.compile(r"-?\d+(?:\.\d+)?") +_INTEGRATION_METHOD = "per_device_trapezoidal_with_linear_boundary_interpolation" +_DEFAULT_MAX_SAMPLE_GAP_S = 3.0 +_POWER_METRIC_KEYS = { + "avg_power_w", + "avg_total_gpu_power_w", + "total_gpu_energy_j", + "joules_per_successful_query", + "joules_per_input_token", + "joules_per_output_token", + "joules_per_total_token", +} + + +@dataclass(frozen=True) +class PowerIntegration: + """Auditable result of integrating one single-node telemetry stream.""" + + power_valid: bool + invalid_reasons: tuple[str, ...] + expected_num_gpus: int | None + observed_gpu_ids: tuple[str, ...] + per_gpu_sample_counts: dict[str, int] + per_gpu_max_sample_gap_s: dict[str, float] + per_gpu_energy_j: dict[str, float] + device_issues: dict[str, list[str]] + avg_power_w: float | None = None + avg_total_gpu_power_w: float | None = None + total_gpu_energy_j: float | None = None + + @property + def observed_num_gpus(self) -> int: + return len(self.observed_gpu_ids) + + +@dataclass(frozen=True) +class BenchmarkData: + """Raw benchmark fields required for energy normalization.""" + + start_unix: float + end_unix: float + reported_duration_s: float + completed: int + total_input_tokens: int + total_output_tokens: int + + @property + def integration_duration_s(self) -> float: + return self.end_unix - self.start_unix + def _parse_timestamp(value: str) -> float | None: """Best-effort timestamp parse to Unix epoch seconds (local wall clock). @@ -113,10 +156,10 @@ def aggregate_power( start_unix: float, end_unix: float, ) -> tuple[float, int] | None: - """Return (per_gpu_avg_power_w, num_gpus) for samples in [start, end]. + """Legacy arithmetic mean retained for callers of the pre-PR1 helper. - Returns None if the CSV is missing, empty, has no detectable power column, - or no rows fall in the window. + Published PR1 metrics use :func:`integrate_power`; this compatibility API + must not be used for energy calculations. """ if not csv_path.is_file() or csv_path.stat().st_size == 0: return None @@ -190,32 +233,335 @@ def aggregate_power( return mean(per_sample_mean_per_gpu), num_gpus -def _load_bench_window( - bench_result_path: Path, -) -> tuple[float, float, float, int, int] | None: - """Read (start_unix, end_unix, duration_s, total_output_tokens, total_input_tokens) - from the raw bench JSON. Returns None if any required field is missing. +def _append_reason(reasons: list[str], reason: str) -> None: + """Append a validation reason once while preserving discovery order.""" + if reason not in reasons: + reasons.append(reason) + + +def _gpu_sort_key(gpu_id: str) -> tuple[int, int | str]: + """Sort numeric GPU identities before arbitrary string identities.""" + return (0, int(gpu_id)) if gpu_id.isdigit() else (1, gpu_id) + + +def _empty_integration( + *, + expected_num_gpus: int | None, + reasons: list[str], +) -> PowerIntegration: + """Build an invalid integration result when no device data is available.""" + return PowerIntegration( + power_valid=False, + invalid_reasons=tuple(reasons), + expected_num_gpus=expected_num_gpus, + observed_gpu_ids=(), + per_gpu_sample_counts={}, + per_gpu_max_sample_gap_s={}, + per_gpu_energy_j={}, + device_issues={}, + ) - total_input_tokens defaults to 0 if absent (older bench JSONs may not have it); - this only degrades joules_per_total_token to equal joules_per_output_token in - that case, never breaks the rest of the aggregation. + +def _interpolate_power(samples: list[tuple[float, float]], timestamp: float) -> float: + """Linearly interpolate power at a timestamp bracketed by ``samples``.""" + times = [sample_time for sample_time, _ in samples] + right_index = bisect.bisect_left(times, timestamp) + if right_index < len(samples) and math.isclose( + samples[right_index][0], timestamp, rel_tol=0.0, abs_tol=1e-9 + ): + return samples[right_index][1] + left_index = right_index - 1 + if left_index < 0 or right_index >= len(samples): + raise ValueError("timestamp is not bracketed") + left_time, left_power = samples[left_index] + right_time, right_power = samples[right_index] + fraction = (timestamp - left_time) / (right_time - left_time) + return left_power + fraction * (right_power - left_power) + + +def _integrate_device( + samples: list[tuple[float, float]], + *, + start_unix: float, + end_unix: float, +) -> float: + """Integrate one device over ``[start_unix, end_unix]``.""" + start_power = _interpolate_power(samples, start_unix) + end_power = _interpolate_power(samples, end_unix) + clipped = [(start_unix, start_power)] + clipped.extend( + (timestamp, power) + for timestamp, power in samples + if start_unix < timestamp < end_unix + ) + clipped.append((end_unix, end_power)) + + energy_j = 0.0 + for (left_time, left_power), (right_time, right_power) in zip( + clipped, clipped[1:] + ): + energy_j += (right_time - left_time) * (left_power + right_power) / 2.0 + return energy_j + + +def integrate_power( + csv_path: Path, + *, + start_unix: float, + end_unix: float, + expected_num_gpus: int | None = None, + max_sample_gap_s: float = _DEFAULT_MAX_SAMPLE_GAP_S, +) -> PowerIntegration: + """Validate and integrate per-device GPU power over the formal window. + + A valid stream must contain parseable timestamps and power values, expose + stable GPU identities, match the expected topology when supplied, bracket + both window boundaries for every device, and have no in-window sampling gap + larger than ``max_sample_gap_s``. """ + reasons: list[str] = [] + if ( + not math.isfinite(start_unix) + or not math.isfinite(end_unix) + or end_unix <= start_unix + ): + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=["invalid_benchmark_window"], + ) + if expected_num_gpus is not None and expected_num_gpus <= 0: + _append_reason(reasons, "invalid_expected_gpu_count") + if not math.isfinite(max_sample_gap_s) or max_sample_gap_s <= 0: + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=["invalid_max_sample_gap"], + ) + + if not csv_path.is_file(): + _append_reason(reasons, "telemetry_file_missing") + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ) + try: + if csv_path.stat().st_size == 0: + _append_reason(reasons, "telemetry_file_empty") + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ) + except OSError: + _append_reason(reasons, "telemetry_file_unreadable") + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ) + + # Values are first grouped by GPU and exact timestamp. Some SMI versions + # expose timestamps at lower resolution than their sampling cadence, so + # duplicate-timestamp readings are averaged rather than treated as corrupt. + raw_samples: dict[str, dict[float, list[float]]] = {} + saw_missing_gpu_identity = False + try: + with csv_path.open("r", newline="", encoding="utf-8", errors="replace") as f: + reader = csv.DictReader(f, skipinitialspace=True) + header = [column.strip() for column in (reader.fieldnames or [])] + reader.fieldnames = header + timestamp_col, power_col, gpu_col = _detect_columns(header) + missing_required_columns = False + if not timestamp_col: + _append_reason(reasons, "timestamp_column_missing") + missing_required_columns = True + if not power_col: + _append_reason(reasons, "power_column_missing") + missing_required_columns = True + if not gpu_col: + _append_reason(reasons, "gpu_identity_column_missing") + missing_required_columns = True + if missing_required_columns: + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ) + + for row in reader: + timestamp = _parse_timestamp((row.get(timestamp_col) or "").strip()) + if timestamp is None or not math.isfinite(timestamp): + _append_reason(reasons, "invalid_timestamp_sample") + continue + # Note (wenyao): corrupt warmup/eval rows are skipped only when + # their parseable timestamp proves they cannot affect the + # window. An unparseable timestamp has unknown position, so it + # was rejected above instead of skipped. + if ( + timestamp < start_unix - max_sample_gap_s + or timestamp > end_unix + max_sample_gap_s + ): + continue + + power = _parse_power((row.get(power_col) or "").strip()) + gpu_id = (row.get(gpu_col) or "").strip() + if power is None: + _append_reason(reasons, "invalid_power_sample") + continue + if not math.isfinite(power) or power < 0: + _append_reason(reasons, "invalid_power_sample") + continue + if not gpu_id: + saw_missing_gpu_identity = True + continue + values = raw_samples.setdefault(gpu_id, {}).setdefault(timestamp, []) + values.append(power) + except (OSError, csv.Error): + _append_reason(reasons, "telemetry_file_unreadable") + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ) + + if saw_missing_gpu_identity: + _append_reason(reasons, "gpu_identity_missing") + if not raw_samples: + _append_reason(reasons, "no_usable_power_samples") + return _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ) + + observed_gpu_ids = tuple(sorted(raw_samples, key=_gpu_sort_key)) + if ( + expected_num_gpus is not None + and expected_num_gpus > 0 + and len(observed_gpu_ids) != expected_num_gpus + ): + _append_reason(reasons, "expected_gpu_count_mismatch") + + per_gpu_sample_counts: dict[str, int] = {} + per_gpu_max_sample_gap_s: dict[str, float] = {} + per_gpu_energy_j: dict[str, float] = {} + device_issues: dict[str, list[str]] = {} + + for gpu_id in observed_gpu_ids: + timestamp_values = raw_samples[gpu_id] + samples = sorted( + (timestamp, mean(values)) for timestamp, values in timestamp_values.items() + ) + per_gpu_sample_counts[gpu_id] = len(samples) + issues: list[str] = [] + + if len(samples) < 2: + issues.append("insufficient_power_samples") + _append_reason(reasons, "insufficient_power_samples") + elif samples[0][0] > start_unix or samples[-1][0] < end_unix: + issues.append("benchmark_window_not_bracketed") + _append_reason(reasons, "benchmark_window_not_bracketed") + else: + times = [timestamp for timestamp, _ in samples] + left_index = bisect.bisect_right(times, start_unix) - 1 + right_index = bisect.bisect_left(times, end_unix) + relevant = samples[left_index : right_index + 1] + gaps = [ + right[0] - left[0] for left, right in zip(relevant, relevant[1:]) + ] + max_gap = max(gaps, default=0.0) + per_gpu_max_sample_gap_s[gpu_id] = max_gap + if max_gap > max_sample_gap_s: + issues.append("sampling_gap_exceeded") + _append_reason(reasons, "sampling_gap_exceeded") + per_gpu_energy_j[gpu_id] = _integrate_device( + samples, + start_unix=start_unix, + end_unix=end_unix, + ) + + if issues: + device_issues[gpu_id] = issues + + duration_s = end_unix - start_unix + total_gpu_energy_j: float | None = None + avg_total_gpu_power_w: float | None = None + avg_power_w: float | None = None + if len(per_gpu_energy_j) == len(observed_gpu_ids): + total_gpu_energy_j = sum(per_gpu_energy_j.values()) + avg_total_gpu_power_w = total_gpu_energy_j / duration_s + avg_power_w = avg_total_gpu_power_w / len(observed_gpu_ids) + + return PowerIntegration( + power_valid=not reasons, + invalid_reasons=tuple(reasons), + expected_num_gpus=expected_num_gpus, + observed_gpu_ids=observed_gpu_ids, + per_gpu_sample_counts=per_gpu_sample_counts, + per_gpu_max_sample_gap_s=per_gpu_max_sample_gap_s, + per_gpu_energy_j=per_gpu_energy_j, + device_issues=device_issues, + avg_power_w=avg_power_w, + avg_total_gpu_power_w=avg_total_gpu_power_w, + total_gpu_energy_j=total_gpu_energy_j, + ) + + +def _load_benchmark_data( + bench_result_path: Path, +) -> tuple[BenchmarkData | None, list[str]]: + """Load the strict energy-normalization contract from raw benchmark JSON.""" try: bench = json.loads(bench_result_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): - return None + return None, ["invalid_benchmark_result"] + start = bench.get("benchmark_start_time_unix") end = bench.get("benchmark_end_time_unix") duration = bench.get("duration") - total_output = bench.get("total_output_tokens") - total_input = bench.get("total_input_tokens", 0) - if not all(isinstance(v, (int, float)) for v in (start, end, duration)): - return None - if not isinstance(total_output, int) or total_output <= 0: - return None - if not isinstance(total_input, int) or total_input < 0: + numeric_window = all( + isinstance(value, (int, float)) and not isinstance(value, bool) + for value in (start, end, duration) + ) + if not numeric_window: + return None, ["invalid_benchmark_window"] + + start = float(start) + end = float(end) + duration = float(duration) + if ( + not all(math.isfinite(value) for value in (start, end, duration)) + or end <= start + or duration <= 0 + ): + return None, ["invalid_benchmark_window"] + + reasons: list[str] = [] + integration_duration = end - start + duration_tolerance = max(0.5, integration_duration * 0.01) + if abs(duration - integration_duration) > duration_tolerance: + _append_reason(reasons, "benchmark_duration_mismatch") + + completed = bench.get("completed") + if not isinstance(completed, int) or isinstance(completed, bool) or completed <= 0: + _append_reason(reasons, "invalid_successful_query_count") + completed = 0 + + total_input = bench.get("total_input_tokens") + if not isinstance(total_input, int) or isinstance(total_input, bool) or total_input <= 0: + _append_reason(reasons, "invalid_input_token_count") total_input = 0 - return float(start), float(end), float(duration), int(total_output), int(total_input) + + total_output = bench.get("total_output_tokens") + if not isinstance(total_output, int) or isinstance(total_output, bool) or total_output <= 0: + _append_reason(reasons, "invalid_output_token_count") + total_output = 0 + + return ( + BenchmarkData( + start_unix=start, + end_unix=end, + reported_duration_s=duration, + completed=completed, + total_input_tokens=total_input, + total_output_tokens=total_output, + ), + reasons, + ) def patch_agg_result( @@ -234,57 +580,195 @@ def patch_agg_result( tmp_path.replace(agg_path) -def run(csv_path: Path, bench_result: Path, agg_result: Path) -> int: - window = _load_bench_window(bench_result) - if window is None: - print( - f"[aggregate_power] No bench window in {bench_result} — skipping power aggregation", - file=sys.stderr, +def _write_json_atomic(path: Path, data: dict) -> None: + """Atomically replace a JSON artifact with formatted UTF-8 content.""" + path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = path.with_suffix(path.suffix + ".tmp") + tmp_path.write_text(json.dumps(data, indent=2), encoding="utf-8") + tmp_path.replace(path) + + +def _derived_metrics( + integration: PowerIntegration, + benchmark: BenchmarkData, +) -> dict[str, float]: + """Return whole-deployment energy metrics for a valid measurement.""" + if ( + integration.avg_power_w is None + or integration.avg_total_gpu_power_w is None + or integration.total_gpu_energy_j is None + ): + raise ValueError("valid power integration has incomplete power metrics") + avg_power_w = integration.avg_power_w + avg_total_gpu_power_w = integration.avg_total_gpu_power_w + energy = integration.total_gpu_energy_j + total_tokens = benchmark.total_input_tokens + benchmark.total_output_tokens + return { + "avg_power_w": avg_power_w, + "avg_total_gpu_power_w": avg_total_gpu_power_w, + "total_gpu_energy_j": energy, + "joules_per_successful_query": energy / benchmark.completed, + "joules_per_input_token": energy / benchmark.total_input_tokens, + "joules_per_output_token": energy / benchmark.total_output_tokens, + "joules_per_total_token": energy / total_tokens, + } + + +def _patch_power_result( + agg_path: Path, + *, + power_valid: bool, + metrics: dict[str, float], +) -> None: + """Replace aggregate power fields with one validated metric set.""" + data = json.loads(agg_path.read_text(encoding="utf-8")) + for key in _POWER_METRIC_KEYS: + data.pop(key, None) + # Keep the canonical aggregate numeric-only for InferenceX-app's metric + # auto-capture. Detailed reason codes live in the validation sidecar. + data["power_valid"] = int(power_valid) + data.pop("power_invalid_reasons", None) + if power_valid: + for key, value in metrics.items(): + if value is None or not math.isfinite(value): + raise ValueError(f"non-finite power metric: {key}") + precision = 3 if key.endswith(("_w", "_j")) else 6 + data[key] = round(value, precision) + _write_json_atomic(agg_path, data) + + +def _validation_payload( + *, + csv_path: Path, + bench_result: Path, + benchmark: BenchmarkData | None, + integration: PowerIntegration, + power_valid: bool, + reasons: list[str], + metrics: dict[str, float], +) -> dict: + """Build the complete auditable power-validation sidecar payload.""" + benchmark_window = None + if benchmark is not None: + benchmark_window = { + "start_time_unix": benchmark.start_unix, + "end_time_unix": benchmark.end_unix, + "reported_duration_s": benchmark.reported_duration_s, + "integration_duration_s": benchmark.integration_duration_s, + "completed": benchmark.completed, + "total_input_tokens": benchmark.total_input_tokens, + "total_output_tokens": benchmark.total_output_tokens, + } + return { + "schema_version": 1, + "power_valid": power_valid, + "reasons": reasons, + "telemetry_source": str(csv_path), + "benchmark_result": str(bench_result), + "benchmark_window": benchmark_window, + "integration_method": _INTEGRATION_METHOD, + "expected_gpu_count": integration.expected_num_gpus, + "observed_gpu_count": integration.observed_num_gpus, + "observed_gpu_ids": list(integration.observed_gpu_ids), + "per_gpu_sample_counts": integration.per_gpu_sample_counts, + "per_gpu_max_sample_gap_s": integration.per_gpu_max_sample_gap_s, + "per_gpu_energy_j": integration.per_gpu_energy_j, + "device_issues": integration.device_issues, + "metrics": { + key: round(value, 6) + for key, value in metrics.items() + if value is not None and math.isfinite(value) + }, + } + + +def run( + csv_path: Path, + bench_result: Path, + agg_result: Path, + *, + expected_num_gpus: int | None = None, + validation_result: Path | None = None, + require_power: bool = False, +) -> int: + """Aggregate power, always preserving validity, and optionally fail closed.""" + validation_result = validation_result or bench_result.with_name( + f"power_validation_{bench_result.stem}.json" + ) + benchmark, benchmark_reasons = _load_benchmark_data(bench_result) + if benchmark is None: + integration = _empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=benchmark_reasons.copy(), ) - return 0 - start, end, duration, total_output, total_input = window - - result = aggregate_power(csv_path, start, end) - if result is None: - print( - f"[aggregate_power] No usable power samples in {csv_path} for " - f"window [{start}, {end}] — skipping", - file=sys.stderr, + else: + integration = integrate_power( + csv_path, + start_unix=benchmark.start_unix, + end_unix=benchmark.end_unix, + expected_num_gpus=expected_num_gpus, ) - return 0 - avg_power_w, num_gpus = result - - # Joules consumed by the system during the bench window, divided by either - # output tokens (for generation-cost metrics) or all tokens (for whole- - # workload efficiency). - total_system_energy_j = avg_power_w * num_gpus * duration - joules_per_output_token = total_system_energy_j / total_output - total_tokens = total_output + total_input - joules_per_total_token = ( - total_system_energy_j / total_tokens if total_tokens > 0 else joules_per_output_token - ) + + reasons = benchmark_reasons.copy() + for reason in integration.invalid_reasons: + _append_reason(reasons, reason) + + metrics: dict[str, float] = {} + power_valid = not reasons + if power_valid and benchmark is not None: + metrics = _derived_metrics(integration, benchmark) if not agg_result.is_file(): + _append_reason(reasons, "aggregate_result_missing") + power_valid = False + metrics = {} + else: + try: + _patch_power_result( + agg_result, + power_valid=power_valid, + metrics=metrics, + ) + except (OSError, json.JSONDecodeError, ValueError) as exc: + _append_reason(reasons, "aggregate_result_unwritable") + power_valid = False + metrics = {} + print(f"[aggregate_power] Failed to patch {agg_result}: {exc}", file=sys.stderr) + + try: + _write_json_atomic( + validation_result, + _validation_payload( + csv_path=csv_path, + bench_result=bench_result, + benchmark=benchmark, + integration=integration, + power_valid=power_valid, + reasons=reasons, + metrics=metrics, + ), + ) + except OSError as exc: print( - f"[aggregate_power] Agg result {agg_result} missing — cannot patch", + f"[aggregate_power] Failed to write validation artifact " + f"{validation_result}: {exc}", file=sys.stderr, ) - return 0 + return 1 if require_power else 0 - try: - patch_agg_result( - agg_result, avg_power_w, joules_per_output_token, joules_per_total_token + if not power_valid: + print( + f"[aggregate_power] Power validation failed: {', '.join(reasons)} " + f"(details: {validation_result})", + file=sys.stderr, ) - except (OSError, json.JSONDecodeError) as exc: - print(f"[aggregate_power] Failed to patch {agg_result}: {exc}", file=sys.stderr) - return 0 + return 1 if require_power else 0 print( - f"[aggregate_power] avg_power_w={avg_power_w:.2f} (per GPU, n={num_gpus}) " - f"joules_per_output_token={joules_per_output_token:.4f} " - f"joules_per_total_token={joules_per_total_token:.4f} " - f"duration={duration:.1f}s output_tokens={total_output} input_tokens={total_input} " - f"-> {agg_result}" + f"[aggregate_power] avg_power_w={metrics['avg_power_w']:.2f} " + f"avg_total_gpu_power_w={metrics['avg_total_gpu_power_w']:.2f} " + f"total_gpu_energy_j={metrics['total_gpu_energy_j']:.2f} " + f"(n={integration.observed_num_gpus}) -> {agg_result}" ) return 0 @@ -309,8 +793,32 @@ def main() -> int: required=True, help="Path to the agg_.json output of process_result.py (will be patched in place)", ) + parser.add_argument( + "--expected-num-gpus", + type=int, + required=True, + help="Expected single-node GPU count from TP * PP * PCP", + ) + parser.add_argument( + "--validation-result", + type=Path, + help="Path for the power validation sidecar", + ) + parser.add_argument( + "--require-power", + action="store_true", + default=os.environ.get("REQUIRE_POWER", "").lower() in {"1", "true", "yes"}, + help="Fail when power telemetry is invalid (also enabled by REQUIRE_POWER=1)", + ) args = parser.parse_args() - return run(args.csv, args.bench_result, args.agg_result) + return run( + args.csv, + args.bench_result, + args.agg_result, + expected_num_gpus=args.expected_num_gpus, + validation_result=args.validation_result, + require_power=args.require_power, + ) if __name__ == "__main__": diff --git a/utils/process_result.py b/utils/process_result.py index bf3bcdaba..7e999c78d 100644 --- a/utils/process_result.py +++ b/utils/process_result.py @@ -1,6 +1,6 @@ -import sys import json import os +import sys from pathlib import Path @@ -40,6 +40,57 @@ def get_optional_component_metadata(env_var): return metadata +def record_power_internal_error( + *, + csv_path, + bench_result, + agg_result, + validation_result, + expected_num_gpus, + error, +): + """Preserve an auditable invalid result when aggregation fails unexpectedly.""" + reasons = ["aggregation_internal_error"] + try: + from aggregate_power import ( + _POWER_METRIC_KEYS, + _empty_integration, + _validation_payload, + _write_json_atomic, + ) + + agg_data = json.loads(agg_result.read_text(encoding="utf-8")) + for key in _POWER_METRIC_KEYS: + agg_data.pop(key, None) + agg_data["power_valid"] = 0 + agg_data.pop("power_invalid_reasons", None) + _write_json_atomic(agg_result, agg_data) + + validation_data = _validation_payload( + csv_path=csv_path, + bench_result=bench_result, + benchmark=None, + integration=_empty_integration( + expected_num_gpus=expected_num_gpus, + reasons=reasons, + ), + power_valid=False, + reasons=reasons, + metrics={}, + ) + validation_data["internal_error"] = { + "type": type(error).__name__, + "message": str(error)[:500], + } + _write_json_atomic(validation_result, validation_data) + except (OSError, json.JSONDecodeError, ImportError, AttributeError) as fallback_error: + print( + f"[process_result] failed to preserve power validation fallback: " + f"{fallback_error}", + file=sys.stderr, + ) + + # Base required env vars base_env = get_required_env_vars([ 'RUNNER_TYPE', 'FRAMEWORK', 'PRECISION', 'SPEC_DECODING', @@ -204,16 +255,23 @@ def get_optional_component_metadata(env_var): data[key.replace('_ms', '').replace( 'tpot', 'intvty')] = 1000.0 / float(value) -print(json.dumps(data, indent=2)) - agg_path = Path(f'agg_{result_filename}.json') with open(agg_path, 'w') as f: json.dump(data, f, indent=2) -# Best-effort: patch measured power into the agg JSON. Never fails the run. -try: - from aggregate_power import run as _aggregate_power_run - +# Single-node measured power is best-effort by default. Power studies can set +# REQUIRE_POWER=1 to fail closed after the validation sidecar has been written. +_require_power = os.environ.get('REQUIRE_POWER', '').lower() in {'1', 'true', 'yes'} +_power_status = 0 +if is_multinode: + if _require_power: + print( + '[process_result] Power validation failed: PR1 supports only ' + 'single-node non-disaggregated telemetry', + file=sys.stderr, + ) + _power_status = 1 +else: _csv_candidates = [ os.environ.get('GPU_METRICS_CSV'), 'gpu_metrics.csv', @@ -221,13 +279,36 @@ def get_optional_component_metadata(env_var): ] _csv_path = next( (Path(p) for p in _csv_candidates if p and Path(p).is_file()), - None, + Path(next(p for p in _csv_candidates if p)), ) - if _csv_path is not None: - _aggregate_power_run( + _bench_path = Path(f'{result_filename}.json') + _validation_path = Path(f'power_validation_{result_filename}.json') + try: + from aggregate_power import run as _aggregate_power_run + + _power_status = _aggregate_power_run( csv_path=_csv_path, - bench_result=Path(f'{result_filename}.json'), + bench_result=_bench_path, agg_result=agg_path, + expected_num_gpus=num_gpus, + validation_result=_validation_path, + require_power=_require_power, ) -except Exception as exc: # noqa: BLE001 — never block on telemetry - print(f'[process_result] power aggregation skipped: {exc}', file=sys.stderr) + except Exception as exc: # noqa: BLE001 — preserve ordinary benchmark behavior + print(f'[process_result] power aggregation failed: {exc}', file=sys.stderr) + record_power_internal_error( + csv_path=_csv_path, + bench_result=_bench_path, + agg_result=agg_path, + validation_result=_validation_path, + expected_num_gpus=num_gpus, + error=exc, + ) + if _require_power: + _power_status = 1 + +with open(agg_path) as f: + print(json.dumps(json.load(f), indent=2)) + +if _power_status: + raise SystemExit(_power_status) diff --git a/utils/test_aggregate_power.py b/utils/test_aggregate_power.py index bf81ee7b1..7f4affc2c 100644 --- a/utils/test_aggregate_power.py +++ b/utils/test_aggregate_power.py @@ -3,13 +3,13 @@ Covers: - NVIDIA CSV (nvidia-smi --query-gpu format with "X W" power cells) - AMD CSV (amd-smi --csv with ISO/epoch timestamps and bare numeric power) - - Window filtering (samples outside [start, end] are excluded) - - Multi-GPU per-sample aggregation (sum across GPUs at each timestamp, - then mean over samples — yields per-GPU mean) + - Backward-compatible arithmetic aggregation + - Per-device trapezoidal integration over the formal benchmark window + - Expected/observed GPU validation, window coverage, and sampling gaps - Missing / empty / malformed CSV: returns None, no exception - - End-to-end run(): patches agg JSON with avg_power_w + joules_per_output_token - + joules_per_total_token - - Missing bench window keys: skips gracefully without patching + - Whole-deployment power/energy/J-query/J-token metric semantics + - Best-effort invalid results and REQUIRE_POWER strict failures + - Atomic aggregate and validation artifacts """ from __future__ import annotations @@ -27,6 +27,8 @@ _parse_power, _parse_timestamp, aggregate_power, + integrate_power, + main, patch_agg_result, run, ) @@ -292,6 +294,7 @@ def _write_bench_result( duration: float, total_output: int, total_input: int = 0, + completed: int = 1, ) -> None: path.write_text( json.dumps( @@ -299,6 +302,7 @@ def _write_bench_result( "benchmark_start_time_unix": start, "benchmark_end_time_unix": end, "duration": duration, + "completed": completed, "total_output_tokens": total_output, "total_input_tokens": total_input, } @@ -307,20 +311,269 @@ def _write_bench_result( ) -def test_run_patches_agg_with_power_and_joules(tmp_path: Path): +def _write_constant_window_samples( + path: Path, + *, + start: float, + end: float, + watts_per_gpu: float, + num_gpus: int, +) -> None: + """Write samples that bracket the formal benchmark window.""" + duration = int(end - start) + timestamps = [start - 1.0] + timestamps.extend(start + offset for offset in range(duration + 1)) + timestamps.append(end + 1.0) + _write_nvidia_csv( + path, + [ + (timestamp, gpu, watts_per_gpu) + for timestamp in timestamps + for gpu in range(num_gpus) + ], + ) + + +# --------------------------------------------------------------------------- # +# Validated per-device integration contract +# --------------------------------------------------------------------------- # + + +def test_integrate_power_uses_per_device_trapezoidal_integration(tmp_path: Path): + """Irregular samples are integrated in time per GPU, not averaged by row.""" + csv = tmp_path / "gpu_metrics.csv" base = 1_700_000_000.0 + _write_nvidia_csv( + csv, + [ + # Both devices are linear in time. Boundary values at t=0 and + # t=10 must be interpolated from samples outside the window. + (base - 1, 0, 100.0), + (base + 1, 0, 140.0), + (base + 3.5, 0, 190.0), + (base + 6, 0, 240.0), + (base + 8.5, 0, 290.0), + (base + 11, 0, 340.0), + (base - 1, 1, 200.0), + (base + 1, 1, 240.0), + (base + 3.5, 1, 290.0), + (base + 6, 1, 340.0), + (base + 8.5, 1, 390.0), + (base + 11, 1, 440.0), + ], + ) + + result = integrate_power( + csv, + start_unix=base, + end_unix=base + 10, + expected_num_gpus=2, + ) + + assert result.power_valid is True + assert result.invalid_reasons == () + # GPU 0: average 220 W over 10s = 2200 J. + # GPU 1: average 320 W over 10s = 3200 J. + assert result.per_gpu_energy_j == pytest.approx({"0": 2200.0, "1": 3200.0}) + assert result.total_gpu_energy_j == pytest.approx(5400.0) + assert result.avg_total_gpu_power_w == pytest.approx(540.0) + assert result.avg_power_w == pytest.approx(270.0) + + +def test_integrate_power_rejects_expected_gpu_count_mismatch(tmp_path: Path): + csv = tmp_path / "gpu_metrics.csv" + base = 1_700_000_000.0 + _write_constant_window_samples( + csv, + start=base, + end=base + 10, + watts_per_gpu=500.0, + num_gpus=2, + ) + + result = integrate_power( + csv, + start_unix=base, + end_unix=base + 10, + expected_num_gpus=4, + ) + + assert result.power_valid is False + assert "expected_gpu_count_mismatch" in result.invalid_reasons + assert result.expected_num_gpus == 4 + assert result.observed_num_gpus == 2 + + +def test_integrate_power_requires_every_gpu_to_bracket_window(tmp_path: Path): + csv = tmp_path / "gpu_metrics.csv" + base = 1_700_000_000.0 + _write_nvidia_csv( + csv, + [ + (base - 1, 0, 500.0), + (base + 1, 0, 500.0), + (base + 3, 0, 500.0), + (base + 5, 0, 500.0), + (base + 7, 0, 500.0), + (base + 9, 0, 500.0), + (base + 11, 0, 500.0), + # GPU 1 starts after the formal benchmark start. + (base + 1, 1, 500.0), + (base + 3, 1, 500.0), + (base + 5, 1, 500.0), + (base + 7, 1, 500.0), + (base + 9, 1, 500.0), + (base + 11, 1, 500.0), + ], + ) + + result = integrate_power( + csv, + start_unix=base, + end_unix=base + 10, + expected_num_gpus=2, + ) + + assert result.power_valid is False + assert "benchmark_window_not_bracketed" in result.invalid_reasons + assert result.device_issues == {"1": ["benchmark_window_not_bracketed"]} + + +def test_integrate_power_rejects_sampling_gap(tmp_path: Path): csv = tmp_path / "gpu_metrics.csv" + base = 1_700_000_000.0 _write_nvidia_csv( csv, [ - (base + 1 + sample_idx, gpu, 500.0) - for sample_idx in range(2) - for gpu in range(8) + (base - 1, 0, 500.0), + (base + 1, 0, 500.0), + # The 8-second gap makes a 1 Hz telemetry stream unusable. + (base + 9, 0, 500.0), + (base + 11, 0, 500.0), ], ) + + result = integrate_power( + csv, + start_unix=base, + end_unix=base + 10, + expected_num_gpus=1, + ) + + assert result.power_valid is False + assert "sampling_gap_exceeded" in result.invalid_reasons + assert result.device_issues == {"0": ["sampling_gap_exceeded"]} + + +def test_integrate_power_ignores_invalid_samples_far_outside_window(tmp_path: Path): + """Warmup/eval corruption must not invalidate the formal benchmark slice.""" + csv = tmp_path / "gpu_metrics.csv" + base = 1_700_000_000.0 + samples = [ + # Far-outside negative and conflicting duplicate samples are irrelevant. + (base - 100, 0, -1.0), + (base + 100, 0, 100.0), + (base + 100, 0, 200.0), + ] + samples.extend((base + offset, 0, 500.0) for offset in range(-1, 12)) + _write_nvidia_csv(csv, samples) + + result = integrate_power( + csv, + start_unix=base, + end_unix=base + 10, + expected_num_gpus=1, + ) + + assert result.power_valid is True + assert result.invalid_reasons == () + assert result.total_gpu_energy_j == pytest.approx(5_000.0) + + +@pytest.mark.parametrize( + ("malformed_row", "expected_reason"), + [ + ("not-a-timestamp, 0, 500 W", "invalid_timestamp_sample"), + ("{timestamp}, 0, [N/A]", "invalid_power_sample"), + ], + ids=["timestamp", "power"], +) +def test_run_rejects_malformed_telemetry_inside_window( + tmp_path: Path, + malformed_row: str, + expected_reason: str, +): + """A rejected 1 Hz row must not be interpolated into valid energy metrics.""" + base = 1_700_000_000.0 + csv = tmp_path / "gpu_metrics.csv" + rows = ["timestamp, index, power.draw [W]"] + for offset in range(-1, 12): + timestamp = _nvidia_ts(base + offset) + if offset == 5: + rows.append(malformed_row.format(timestamp=timestamp)) + else: + rows.append(f"{timestamp}, 0, 500 W") + csv.write_text("\n".join(rows) + "\n", encoding="utf-8") + + bench = tmp_path / "bench.json" + agg = tmp_path / "agg.json" + validation = tmp_path / "power_validation.json" + _write_bench_result( + bench, + start=base, + end=base + 10, + duration=10.0, + completed=1, + total_input=1_000, + total_output=1_000, + ) + agg.write_text(json.dumps({"hw": "h200"}), encoding="utf-8") + + exit_code = run( + csv, + bench, + agg, + expected_num_gpus=1, + validation_result=validation, + ) + + assert exit_code == 0 + patched = json.loads(agg.read_text()) + assert patched["power_valid"] == 0 + for metric in ( + "avg_power_w", + "avg_total_gpu_power_w", + "total_gpu_energy_j", + "joules_per_successful_query", + "joules_per_input_token", + "joules_per_output_token", + "joules_per_total_token", + ): + assert metric not in patched + audit = json.loads(validation.read_text()) + assert audit["reasons"] == [expected_reason] + + +def test_run_patches_agg_with_power_and_joules(tmp_path: Path): + base = 1_700_000_000.0 + csv = tmp_path / "gpu_metrics.csv" + _write_constant_window_samples( + csv, + start=base, + end=base + 10, + watts_per_gpu=500.0, + num_gpus=8, + ) bench = tmp_path / "bench.json" agg = tmp_path / "agg.json" - _write_bench_result(bench, start=base, end=base + 10, duration=10.0, total_output=20_000) + _write_bench_result( + bench, + start=base, + end=base + 10, + duration=10.0, + total_output=20_000, + total_input=20_000, + ) agg.write_text(json.dumps({"hw": "h200", "conc": 64}), encoding="utf-8") exit_code = run(csv, bench, agg) @@ -334,8 +587,8 @@ def test_run_patches_agg_with_power_and_joules(tmp_path: Path): assert patched["avg_power_w"] == pytest.approx(500.0) # J/output_token = 500W × 8 GPUs × 10s / 20_000 tokens = 2.0 assert patched["joules_per_output_token"] == pytest.approx(2.0) - # No input tokens were supplied -> J/total_token falls back to J/output_token. - assert patched["joules_per_total_token"] == pytest.approx(2.0) + # J/total_token = 40_000 J / (20_000 input + 20_000 output). + assert patched["joules_per_total_token"] == pytest.approx(1.0) def test_run_computes_j_per_total_token_with_input_tokens(tmp_path: Path): @@ -346,13 +599,12 @@ def test_run_computes_j_per_total_token_with_input_tokens(tmp_path: Path): """ base = 1_700_000_000.0 csv = tmp_path / "gpu_metrics.csv" - _write_nvidia_csv( + _write_constant_window_samples( csv, - [ - (base + 1 + sample_idx, gpu, 500.0) - for sample_idx in range(2) - for gpu in range(8) - ], + start=base, + end=base + 10, + watts_per_gpu=500.0, + num_gpus=8, ) bench = tmp_path / "bench.json" agg = tmp_path / "agg.json" @@ -397,7 +649,10 @@ def test_run_skips_when_bench_window_missing(tmp_path: Path): patched = json.loads(agg.read_text()) assert "avg_power_w" not in patched - assert patched == {"hw": "h200"} + assert patched == { + "hw": "h200", + "power_valid": 0, + } def test_run_skips_when_csv_missing(tmp_path: Path): @@ -445,3 +700,195 @@ def test_patch_agg_result_is_atomic_via_tempfile(tmp_path: Path): assert data["joules_per_total_token"] == 0.5 # No .tmp leftover. assert not (tmp_path / "agg.json.tmp").exists() + + +def test_run_emits_complete_whole_deployment_metric_contract(tmp_path: Path): + base = 1_700_000_000.0 + csv = tmp_path / "gpu_metrics.csv" + _write_constant_window_samples( + csv, + start=base, + end=base + 10, + watts_per_gpu=500.0, + num_gpus=8, + ) + bench = tmp_path / "bench.json" + agg = tmp_path / "agg.json" + validation = tmp_path / "power_validation.json" + _write_bench_result( + bench, + start=base, + end=base + 10, + duration=10.0, + completed=10, + total_input=10_000, + total_output=2_000, + ) + agg.write_text(json.dumps({"hw": "h200"}), encoding="utf-8") + + exit_code = run( + csv, + bench, + agg, + expected_num_gpus=8, + validation_result=validation, + ) + + assert exit_code == 0 + patched = json.loads(agg.read_text()) + assert type(patched["power_valid"]) is int + assert patched["power_valid"] == 1 + assert "power_invalid_reasons" not in patched + assert patched["avg_power_w"] == pytest.approx(500.0) + assert patched["avg_total_gpu_power_w"] == pytest.approx(4_000.0) + assert patched["total_gpu_energy_j"] == pytest.approx(40_000.0) + assert patched["joules_per_successful_query"] == pytest.approx(4_000.0) + assert patched["joules_per_input_token"] == pytest.approx(4.0) + assert patched["joules_per_output_token"] == pytest.approx(20.0) + assert patched["joules_per_total_token"] == pytest.approx(40_000 / 12_000) + + audit = json.loads(validation.read_text()) + assert audit["schema_version"] == 1 + assert audit["power_valid"] is True + assert audit["reasons"] == [] + assert audit["expected_gpu_count"] == 8 + assert audit["observed_gpu_count"] == 8 + assert audit["benchmark_window"]["start_time_unix"] == base + assert audit["benchmark_window"]["end_time_unix"] == base + 10 + assert audit["benchmark_window"]["integration_duration_s"] == 10.0 + assert audit["integration_method"] == ( + "per_device_trapezoidal_with_linear_boundary_interpolation" + ) + + +def test_run_best_effort_marks_invalid_power_and_preserves_benchmark(tmp_path: Path): + bench = tmp_path / "bench.json" + agg = tmp_path / "agg.json" + validation = tmp_path / "power_validation.json" + _write_bench_result( + bench, + start=1_700_000_000.0, + end=1_700_000_010.0, + duration=10.0, + completed=10, + total_input=10_000, + total_output=2_000, + ) + agg.write_text(json.dumps({"hw": "h200", "conc": 4}), encoding="utf-8") + + exit_code = run( + tmp_path / "missing.csv", + bench, + agg, + expected_num_gpus=8, + validation_result=validation, + ) + + assert exit_code == 0 + patched = json.loads(agg.read_text()) + assert patched["hw"] == "h200" + assert patched["conc"] == 4 + assert type(patched["power_valid"]) is int + assert patched["power_valid"] == 0 + assert "power_invalid_reasons" not in patched + for metric in ( + "avg_power_w", + "avg_total_gpu_power_w", + "total_gpu_energy_j", + "joules_per_successful_query", + "joules_per_input_token", + "joules_per_output_token", + "joules_per_total_token", + ): + assert metric not in patched + + audit = json.loads(validation.read_text()) + assert audit["power_valid"] is False + assert audit["reasons"] == ["telemetry_file_missing"] + + +def test_run_strict_mode_fails_after_writing_validation(tmp_path: Path): + bench = tmp_path / "bench.json" + agg = tmp_path / "agg.json" + validation = tmp_path / "power_validation.json" + _write_bench_result( + bench, + start=1_700_000_000.0, + end=1_700_000_010.0, + duration=10.0, + completed=10, + total_input=10_000, + total_output=2_000, + ) + agg.write_text(json.dumps({"hw": "h200"}), encoding="utf-8") + + exit_code = run( + tmp_path / "missing.csv", + bench, + agg, + expected_num_gpus=8, + validation_result=validation, + require_power=True, + ) + + assert exit_code == 1 + assert json.loads(agg.read_text())["power_valid"] == 0 + assert json.loads(validation.read_text())["reasons"] == ["telemetry_file_missing"] + + +def test_run_invalid_benchmark_denominator_is_auditable(tmp_path: Path): + base = 1_700_000_000.0 + csv = tmp_path / "gpu_metrics.csv" + _write_constant_window_samples( + csv, + start=base, + end=base + 10, + watts_per_gpu=500.0, + num_gpus=1, + ) + bench = tmp_path / "bench.json" + agg = tmp_path / "agg.json" + validation = tmp_path / "power_validation.json" + _write_bench_result( + bench, + start=base, + end=base + 10, + duration=10.0, + completed=0, + total_input=10_000, + total_output=2_000, + ) + agg.write_text(json.dumps({"hw": "h200"}), encoding="utf-8") + + exit_code = run( + csv, + bench, + agg, + expected_num_gpus=1, + validation_result=validation, + ) + + assert exit_code == 0 + assert json.loads(agg.read_text())["power_valid"] == 0 + assert json.loads(validation.read_text())["reasons"] == [ + "invalid_successful_query_count" + ] + + +def test_cli_requires_expected_gpu_count(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr( + sys, + "argv", + [ + "aggregate_power.py", + "--bench-result", + "bench.json", + "--agg-result", + "agg.json", + ], + ) + + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 2 diff --git a/utils/test_process_result.py b/utils/test_process_result.py index ec6d245cb..78212098a 100644 --- a/utils/test_process_result.py +++ b/utils/test_process_result.py @@ -4,12 +4,13 @@ 1. Testing the get_required_env_vars function directly 2. Running the script as a subprocess with mocked environment and files """ -import pytest import json import subprocess import sys from pathlib import Path +import pytest + SCRIPT_PATH = Path(__file__).parent / "process_result.py" @@ -104,6 +105,34 @@ def run_script(tmp_path, env, benchmark_result, result_filename="benchmark_resul ) +def run_script_with_broken_aggregator( + tmp_path, env, benchmark_result, result_filename="benchmark_result" +): + """Run process_result with the real aggregator patched to raise unexpectedly.""" + result_file = tmp_path / f"{result_filename}.json" + result_file.write_text(json.dumps(benchmark_result)) + env = {**env, "RESULT_FILENAME": result_filename} + wrapper = f""" +import runpy +import sys + +sys.path.insert(0, {str(SCRIPT_PATH.parent)!r}) +import aggregate_power + +def broken_run(**kwargs): + raise RuntimeError("forced aggregation failure") +aggregate_power.run = broken_run +runpy.run_path({str(SCRIPT_PATH)!r}, run_name="__main__") +""" + return subprocess.run( + [sys.executable, "-c", wrapper], + cwd=tmp_path, + env=env, + capture_output=True, + text=True, + ) + + # ============================================================================= # Test get_required_env_vars function # ============================================================================= @@ -622,10 +651,10 @@ def test_conc_from_benchmark_result(self, tmp_path, single_node_env_vars): class TestPowerAggregationIntegration: """End-to-end wiring: process_result.py invokes aggregate_power.py and - patches avg_power_w + joules_per_output_token into the agg JSON. + patches the validated whole-deployment power contract into the agg JSON. Exercises the env-var path resolution (GPU_METRICS_CSV), the subprocess - boundary, and the best-effort try/except that wraps the aggregator call. + boundary, topology validation, and best-effort/strict modes. """ @staticmethod @@ -668,6 +697,8 @@ def test_agg_json_gets_patched_with_power_and_joules(self, tmp_path, single_node "benchmark_start_time_unix": start, "benchmark_end_time_unix": end, "duration": 60.0, + "completed": 30, + "total_input_tokens": 240_000, "total_output_tokens": 30_000, } env = {**single_node_env_vars, "GPU_METRICS_CSV": str(csv_path)} @@ -684,9 +715,15 @@ def test_agg_json_gets_patched_with_power_and_joules(self, tmp_path, single_node assert patched["tp"] == 8 assert patched["conc"] == 64 # New power fields. + assert patched["power_valid"] == 1 assert patched["avg_power_w"] == pytest.approx(600.0, abs=0.5) + assert patched["avg_total_gpu_power_w"] == pytest.approx(4_800.0, abs=0.5) + assert patched["total_gpu_energy_j"] == pytest.approx(288_000.0, abs=0.5) + assert patched["joules_per_successful_query"] == pytest.approx(9_600.0, abs=0.05) + assert patched["joules_per_input_token"] == pytest.approx(1.2, abs=0.01) # 600W × 8 GPUs × 60s / 30_000 tokens = 9.6 J/tok assert patched["joules_per_output_token"] == pytest.approx(9.6, abs=0.05) + assert (tmp_path / "power_validation_benchmark_result.json").is_file() def test_missing_csv_does_not_break_process_result(self, tmp_path, single_node_env_vars): """Without GPU_METRICS_CSV (or with a missing file), process_result.py @@ -697,6 +734,12 @@ def test_missing_csv_does_not_break_process_result(self, tmp_path, single_node_e "max_concurrency": 64, "total_token_throughput": 1000.0, "output_throughput": 500.0, + "benchmark_start_time_unix": 1_700_000_100.0, + "benchmark_end_time_unix": 1_700_000_110.0, + "duration": 10.0, + "completed": 4, + "total_input_tokens": 32_768, + "total_output_tokens": 4_096, } result = run_script(tmp_path, single_node_env_vars, benchmark_result) @@ -706,6 +749,13 @@ def test_missing_csv_does_not_break_process_result(self, tmp_path, single_node_e patched = json.loads(agg_path.read_text()) assert "avg_power_w" not in patched assert "joules_per_output_token" not in patched + assert patched["power_valid"] == 0 + assert "power_invalid_reasons" not in patched + + validation = json.loads( + (tmp_path / "power_validation_benchmark_result.json").read_text() + ) + assert validation["reasons"] == ["telemetry_file_missing"] def test_missing_bench_timestamps_does_not_patch(self, tmp_path, single_node_env_vars): """A CSV is present but the bench JSON predates the timestamp fields @@ -730,3 +780,334 @@ def test_missing_bench_timestamps_does_not_patch(self, tmp_path, single_node_env patched = json.loads(agg_path.read_text()) assert "avg_power_w" not in patched assert "joules_per_output_token" not in patched + assert patched["power_valid"] == 0 + assert "power_invalid_reasons" not in patched + + def test_expected_gpu_count_mismatch_is_invalid(self, tmp_path, single_node_env_vars): + """TP/PP/PCP topology is checked against the observed device IDs.""" + start, end = 1_700_000_100.0, 1_700_000_110.0 + csv_path = tmp_path / "gpu_metrics.csv" + self._write_nvidia_csv(csv_path, start, end, watts_per_gpu=600.0, num_gpus=4) + benchmark_result = { + "model_id": "test-model", + "max_concurrency": 4, + "total_token_throughput": 1000.0, + "output_throughput": 500.0, + "benchmark_start_time_unix": start, + "benchmark_end_time_unix": end, + "duration": 10.0, + "completed": 4, + "total_input_tokens": 32_768, + "total_output_tokens": 4_096, + } + env = {**single_node_env_vars, "GPU_METRICS_CSV": str(csv_path)} + + result = run_script(tmp_path, env, benchmark_result) + + assert result.returncode == 0, f"Script failed: {result.stderr}" + patched = json.loads((tmp_path / "agg_benchmark_result.json").read_text()) + assert patched["power_valid"] == 0 + assert "power_invalid_reasons" not in patched + assert "total_gpu_energy_j" not in patched + + def test_require_power_propagates_validation_failure( + self, tmp_path, single_node_env_vars + ): + """Study/CI mode must fail after preserving validation artifacts.""" + benchmark_result = { + "model_id": "test-model", + "max_concurrency": 4, + "total_token_throughput": 1000.0, + "output_throughput": 500.0, + "benchmark_start_time_unix": 1_700_000_100.0, + "benchmark_end_time_unix": 1_700_000_110.0, + "duration": 10.0, + "completed": 4, + "total_input_tokens": 32_768, + "total_output_tokens": 4_096, + } + env = {**single_node_env_vars, "REQUIRE_POWER": "1"} + + result = run_script(tmp_path, env, benchmark_result) + + assert result.returncode != 0 + assert "Power validation failed" in result.stderr + validation = json.loads( + (tmp_path / "power_validation_benchmark_result.json").read_text() + ) + assert validation["reasons"] == ["telemetry_file_missing"] + + def test_require_power_accepts_valid_single_node_measurement( + self, tmp_path, single_node_env_vars + ): + """The strict H100/H200 canary path also has a protected success case.""" + start, end = 1_700_000_100.0, 1_700_000_110.0 + csv_path = tmp_path / "gpu_metrics.csv" + self._write_nvidia_csv( + csv_path, + start, + end, + watts_per_gpu=500.0, + num_gpus=8, + ) + benchmark_result = { + "model_id": "test-model", + "max_concurrency": 4, + "total_token_throughput": 1000.0, + "output_throughput": 500.0, + "benchmark_start_time_unix": start, + "benchmark_end_time_unix": end, + "duration": 10.0, + "completed": 4, + "total_input_tokens": 32_768, + "total_output_tokens": 4_096, + } + env = { + **single_node_env_vars, + "GPU_METRICS_CSV": str(csv_path), + "REQUIRE_POWER": "1", + } + + result = run_script(tmp_path, env, benchmark_result) + + assert result.returncode == 0, result.stderr + agg = json.loads((tmp_path / "agg_benchmark_result.json").read_text()) + assert agg["power_valid"] == 1 + assert agg["total_gpu_energy_j"] == pytest.approx(40_000.0) + validation = json.loads( + (tmp_path / "power_validation_benchmark_result.json").read_text() + ) + assert validation["power_valid"] is True + assert validation["reasons"] == [] + + @pytest.mark.parametrize( + ("require_power", "expected_returncode"), + [(False, 0), (True, 1)], + ) + def test_internal_aggregation_error_is_always_auditable( + self, + tmp_path, + single_node_env_vars, + require_power, + expected_returncode, + ): + benchmark_result = { + "model_id": "test-model", + "max_concurrency": 4, + "total_token_throughput": 1000.0, + "output_throughput": 500.0, + "benchmark_start_time_unix": 1_700_000_100.0, + "benchmark_end_time_unix": 1_700_000_110.0, + "duration": 10.0, + "completed": 4, + "total_input_tokens": 32_768, + "total_output_tokens": 4_096, + } + env = single_node_env_vars.copy() + if require_power: + env["REQUIRE_POWER"] = "1" + + result = run_script_with_broken_aggregator(tmp_path, env, benchmark_result) + + assert result.returncode == expected_returncode + agg = json.loads((tmp_path / "agg_benchmark_result.json").read_text()) + assert agg["power_valid"] == 0 + assert "power_invalid_reasons" not in agg + validation = json.loads( + (tmp_path / "power_validation_benchmark_result.json").read_text() + ) + assert validation["power_valid"] is False + assert validation["reasons"] == ["aggregation_internal_error"] + assert validation["internal_error"]["type"] == "RuntimeError" + + def test_stop_gpu_monitor_appends_final_nvidia_sample(self, tmp_path): + """Stopping between 1 Hz ticks still records one post-benchmark sample.""" + fake_bin = tmp_path / "bin" + fake_bin.mkdir() + args_log = tmp_path / "nvidia_args.txt" + fake_nvidia_smi = fake_bin / "nvidia-smi" + fake_nvidia_smi.write_text( + "#!/usr/bin/env bash\n" + f"printf '%s\\n' \"$*\" > {str(args_log)!r}\n" + "printf '%s\\n' " + "'2026/07/23 12:00:11.000, 0, 500.00 W, 65, 1000, 1000, 90 %, 10 %'\n" + ) + fake_nvidia_smi.chmod(0o755) + metrics = tmp_path / "gpu_metrics.csv" + metrics.write_text( + "timestamp, index, power.draw [W], temperature.gpu, " + "clocks.current.sm [MHz], clocks.current.memory [MHz], " + "utilization.gpu [%], utilization.memory [%]\n" + ) + benchmark_lib = Path(__file__).parents[1] / "benchmarks/benchmark_lib.sh" + script = f""" +source {str(benchmark_lib)!r} +kill() {{ return 0; }} +wait() {{ return 0; }} +GPU_MONITOR_PID=999 +GPU_MONITOR_VENDOR=nvidia +GPU_METRICS_CSV={str(metrics)!r} +stop_gpu_monitor +""" + env = { + "PATH": f"{fake_bin}:/usr/bin:/bin", + "PYTHONDONTWRITEBYTECODE": "1", + } + + result = subprocess.run( + ["bash", "-c", script], + env=env, + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert "--format=csv,noheader" in args_log.read_text() + assert "2026/07/23 12:00:11.000, 0, 500.00 W" in metrics.read_text() + + def test_stop_gpu_monitor_drops_truncated_row_before_final_sample(self, tmp_path): + """A killed monitor cannot concatenate its partial row with the final sample.""" + fake_bin = tmp_path / "bin" + fake_bin.mkdir() + fake_nvidia_smi = fake_bin / "nvidia-smi" + final_sample = ( + "2026/07/23 12:00:11.000, 0, 500.00 W, " + "65, 1000, 1000, 90 %, 10 %" + ) + fake_nvidia_smi.write_text( + "#!/usr/bin/env bash\n" + f"printf '%s\\n' {final_sample!r}\n" + ) + fake_nvidia_smi.chmod(0o755) + + header = ( + "timestamp, index, power.draw [W], temperature.gpu, " + "clocks.current.sm [MHz], clocks.current.memory [MHz], " + "utilization.gpu [%], utilization.memory [%]" + ) + complete_sample = ( + "2026/07/23 12:00:09.000, 0, 490.00 W, " + "64, 990, 990, 89 %, 9 %" + ) + truncated_sample = "2026/07/23 12:00:10.000, 0, 52" + metrics = tmp_path / "gpu_metrics.csv" + metrics.write_text( + f"{header}\n{complete_sample}\n{truncated_sample}" + ) + + benchmark_lib = Path(__file__).parents[1] / "benchmarks/benchmark_lib.sh" + script = f""" +source {str(benchmark_lib)!r} +kill() {{ return 0; }} +wait() {{ return 0; }} +GPU_MONITOR_PID=999 +GPU_MONITOR_VENDOR=nvidia +GPU_METRICS_CSV={str(metrics)!r} +stop_gpu_monitor +""" + env = { + "PATH": f"{fake_bin}:/usr/bin:/bin", + "PYTHONDONTWRITEBYTECODE": "1", + } + + result = subprocess.run( + ["bash", "-c", script], + env=env, + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert metrics.read_text().splitlines() == [ + header, + complete_sample, + final_sample, + ] + + +# ============================================================================= +# Static CI/artifact contract +# ============================================================================= + + +class TestPowerWorkflowContract: + """Keep power audit inputs separate from the canonical agg artifact.""" + + @staticmethod + def _repo_root(): + return Path(__file__).parents[1] + + def test_gpu_monitor_default_matches_uploaded_artifact_path(self): + benchmark_lib = self._repo_root() / "benchmarks/benchmark_lib.sh" + script = f""" +unset GPU_METRICS_CSV +source {str(benchmark_lib)!r} +printf '%s' "$GPU_METRICS_CSV" +""" + + result = subprocess.run( + ["bash", "-c", script], + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert result.stdout == "gpu_metrics.csv" + + def test_benchmark_workflow_uploads_independent_power_audit_bundle(self): + workflow = ( + self._repo_root() / ".github/workflows/benchmark-tmpl.yml" + ).read_text() + + assert "name: power_audit_${{ env.RESULT_FILENAME }}" in workflow + assert "${{ env.RESULT_FILENAME }}.json" in workflow + assert "agg_${{ env.RESULT_FILENAME }}.json" in workflow + assert "gpu_metrics.csv" in workflow + assert "power_validation_${{ env.RESULT_FILENAME }}.json" in workflow + assert ( + "always() && !inputs.eval-only && " + "inputs.scenario-type != 'agentic-coding'" + ) in workflow + assert "require-power:" in workflow + assert "REQUIRE_POWER: ${{ inputs.require-power && '1' || '0' }}" in workflow + + # The canonical bmk artifact remains aggregate-only so downstream + # result collection cannot mistake the raw benchmark JSON for an agg. + upload_result = workflow.split("- name: Upload result", 1)[1].split( + "- name: Upload agentic aggregated result", 1 + )[0] + assert "path: agg_${{ env.RESULT_FILENAME }}.json" in upload_result + assert "power_validation_" not in upload_result + + def test_cpu_workflow_covers_power_code_and_both_test_suites(self): + workflow = ( + self._repo_root() / ".github/workflows/test-process-result.yml" + ).read_text() + + for watched_path in ( + "utils/aggregate_power.py", + "utils/test_aggregate_power.py", + "utils/process_result.py", + "utils/test_process_result.py", + ".github/workflows/benchmark-tmpl.yml", + ".github/workflows/e2e-tests.yml", + ".github/workflows/test-process-result.yml", + ): + assert watched_path in workflow + assert "test_aggregate_power.py" in workflow + assert "test_process_result.py" in workflow + assert "github.event.pull_request.draft" not in workflow + + def test_e2e_dispatch_can_enable_strict_power_validation(self): + workflow = ( + self._repo_root() / ".github/workflows/e2e-tests.yml" + ).read_text() + + assert workflow.count("require-power:") >= 3 + single_node_job = workflow.split("test-sweep-single-node:", 1)[1].split( + "test-sweep-evals:", 1 + )[0] + assert "require-power: ${{ inputs.require-power }}" in single_node_job