diff --git a/.github/workflows/benchmark-multinode-tmpl.yml b/.github/workflows/benchmark-multinode-tmpl.yml
index d26d4e0bb..81f4cdb80 100644
--- a/.github/workflows/benchmark-multinode-tmpl.yml
+++ b/.github/workflows/benchmark-multinode-tmpl.yml
@@ -539,6 +539,7 @@ jobs:
             agg_${{ env.RESULT_FILENAME }}_*.json
             power_validation_${{ env.RESULT_FILENAME }}_*.json
             LOGS/power/**
+            LOGS/native_power/**
             result_processing_${{ env.RESULT_FILENAME }}.json
             LOGS/*/results_*.json
             LOGS/agentic/**/agentic_power_concurrency_*.json
diff --git a/.github/workflows/benchmark-tmpl.yml b/.github/workflows/benchmark-tmpl.yml
index e0fca9786..529b04b66 100644
--- a/.github/workflows/benchmark-tmpl.yml
+++ b/.github/workflows/benchmark-tmpl.yml
@@ -449,11 +449,13 @@ jobs:
           name: ${{ inputs.eval-only && 'eval_gpu_metrics_' || 'gpu_metrics_' }}${{ env.RESULT_FILENAME }}
           path: |
             gpu_metrics.csv
+            gpu_metrics*_context.json
             gpu_metrics_energy_start.csv
             gpu_metrics_energy_end.csv
             gpu_metrics_identity.json
             gpu_metrics_identity.csv
             results/gpu_metrics*.csv
+            results/gpu_metrics*_context.json
             results/gpu_metrics_identity.json
           if-no-files-found: ignore
 
@@ -466,12 +468,14 @@ jobs:
             ${{ env.RESULT_FILENAME }}.json
             agg_${{ env.RESULT_FILENAME }}.json
             gpu_metrics.csv
+            gpu_metrics*_context.json
             gpu_metrics_energy_start.csv
             gpu_metrics_energy_end.csv
             gpu_metrics_identity.json
             gpu_metrics_identity.csv
             power_validation_${{ env.RESULT_FILENAME }}.json
             results/gpu_metrics*.csv
+            results/gpu_metrics*_context.json
             results/gpu_metrics_identity.json
             results/agentic_power_window.json
             results/agentic_power_timezone_offset.txt
diff --git a/.github/workflows/test-process-result.yml b/.github/workflows/test-process-result.yml
index a4f30e0b3..1b1b5f5ab 100644
--- a/.github/workflows/test-process-result.yml
+++ b/.github/workflows/test-process-result.yml
@@ -10,6 +10,11 @@ on:
       - '.github/workflows/e2e-tests.yml'
       - '.github/workflows/test-process-result.yml'
       - 'benchmarks/benchmark_lib.sh'
+      - 'benchmarks/native_power_collect.sh'
+      - 'benchmarks/native_power_lifecycle.sh'
+      - 'runners/test_native_collector_barriers.py'
+      - 'runners/test_native_collector_receipts.py'
+      - 'utils/test_native_multinode_power.py'
       - 'benchmarks/multi_node/srt-slurm-recipes/sglang/deepseek-v4/**/*.yaml'
       - 'benchmarks/multi_node/srt-slurm-recipes/vllm/deepseek-v4/**/*.yaml'
       - 'benchmarks/multi_node/srt-slurm-recipes/vllm/kimi-k2.6/b200-fp4/**/*.yaml'
@@ -60,7 +65,7 @@ jobs:
         run: |
           cd utils
           uv run --no-project --exclude-newer PT12H --python 3.12 --with pytest --with pyyaml \
-            python -m pytest test_aggregate_power.py test_aggregate_power_multinode.py agentic/aggregation/ test_gb300_power_official_contract.py test_inject_srt_power_concurrencies.py test_process_result.py ../runners/test_kimik3_bh_power.py -v
+            python -m pytest test_aggregate_power.py test_aggregate_power_multinode.py agentic/aggregation/ test_gb300_power_official_contract.py test_inject_srt_power_concurrencies.py test_process_result.py test_native_multinode_power.py ../runners/test_native_collector_barriers.py ../runners/test_native_collector_receipts.py ../runners/test_kimik3_bh_power.py -v
 
       - name: Test serving client result persistence
         run: |
diff --git a/benchmarks/native_power_collect.sh b/benchmarks/native_power_collect.sh
new file mode 100644
index 000000000..058b8aeb2
--- /dev/null
+++ b/benchmarks/native_power_collect.sh
@@ -0,0 +1,76 @@
+#!/usr/bin/env bash
+# One native SMI collector per serving node. Raw files stay on node-local scratch;
+# the launcher stages them as its host user after containers stop.
+set -uo pipefail
+power_dir=$1
+control_dir=$2
+vendor=$3
+rank=$4
+role=$5
+gpu_indices=$6
+num_nodes=$7
+repo_root=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)
+export TZ=UTC
+export PYTHONPATH="$repo_root${PYTHONPATH:+:$PYTHONPATH}"
+source "$repo_root/benchmarks/benchmark_lib.sh"
+mkdir -p "$power_dir"
+collector_rc=0
+finished=0
+
+write_control() {
+    local path="$control_dir/$1"
+    local pending="$path.tmp"
+    printf '%s\n' "$2" > "$pending" || return
+    # The control directory is created by the host user before Docker starts.
+    # Containers must not strand root-owned files in the shared runner tree.
+    if [[ -n "${POWERX_HOST_UID:-}" && -n "${POWERX_HOST_GID:-}" ]]; then
+        chown "$POWERX_HOST_UID:$POWERX_HOST_GID" "$pending" || return
+    fi
+    mv -f "$pending" "$path"
+}
+
+finish() {
+    local incoming_rc=$?
+    [[ "$finished" == 0 ]] || return
+    if [[ "$incoming_rc" != 0 ]]; then collector_rc=$incoming_rc; fi
+    if ! _background_process_is_running "${GPU_MONITOR_PID:-}"; then collector_rc=1; fi
+    stop_gpu_monitor
+    if [[ "$vendor" == amd ]]; then
+        amd-smi list --json > "$power_dir/gpu_metrics_devices_end.json" || collector_rc=1
+    else
+        nvidia-smi --query-gpu=index,uuid,pci.bus_id,name,driver_version --format=csv \
+            > "$power_dir/gpu_metrics_identity_end.csv" || collector_rc=1
+    fi
+    python3 -m infx.results.power.native_multinode end --directory "$power_dir" \
+        --collector-exit-code "$collector_rc" || collector_rc=1
+    if [[ -n "${POWERX_HOST_UID:-}" && -n "${POWERX_HOST_GID:-}" ]]; then
+        chown -R "$POWERX_HOST_UID:$POWERX_HOST_GID" "$power_dir" || collector_rc=1
+    fi
+    write_control "done-$rank" "$collector_rc"
+    finished=1
+}
+trap finish EXIT
+trap 'collector_rc=130; AMD_MONITOR_STOP_TIMEOUT_S=0; exit 130' INT
+trap 'collector_rc=143; AMD_MONITOR_STOP_TIMEOUT_S=0; exit 143' TERM HUP
+
+case "${POWERX_CLOCK_SYNCHRONIZED:-false}" in
+    yes|true) clock_synchronized=true ;;
+    *) clock_synchronized=false ;;
+esac
+
+python3 -m infx.results.power.native_multinode begin --directory "$power_dir" \
+    --vendor "$vendor" --rank "$rank" --role "$role" --gpu-indices "$gpu_indices" \
+    --num-nodes "$num_nodes" --clock-synchronized "$clock_synchronized" || exit 1
+printf '{"timestamp_timezone":"UTC"}\n' > "$power_dir/gpu_metrics_context.json" || exit 1
+start_gpu_monitor --output "$power_dir/gpu_metrics.csv" || exit 1
+[[ "$GPU_MONITOR_VENDOR" == "$vendor" ]] || exit 1
+if [[ "$vendor" == amd ]]; then
+    _write_amd_smi_sidecar "$power_dir/gpu_metrics_devices.json" list --json
+fi
+_background_process_is_running "$GPU_MONITOR_PID" || exit 1
+write_control "ready-$rank" ready
+while [[ ! -f "$control_dir/stop" ]]; do
+    _background_process_is_running "$GPU_MONITOR_PID" || exit 1
+    sleep 1 &
+    wait $! || true
+done
diff --git a/benchmarks/native_power_lifecycle.sh b/benchmarks/native_power_lifecycle.sh
new file mode 100644
index 000000000..d38a0cc3f
--- /dev/null
+++ b/benchmarks/native_power_lifecycle.sh
@@ -0,0 +1,81 @@
+#!/usr/bin/env bash
+# Shared barriers for native collectors. Launchers own host scratch and mounts.
+
+powerx_start_collector() {
+    local power_dir="$1" control_dir="$2" vendor="$3" rank="$4" role="$5" gpus="$6" nodes="$7"
+    local indices
+    indices=$(seq -s, 0 "$((gpus - 1))") || return 1
+    POWERX_CONTROL_DIR="$control_dir"
+    POWERX_NUM_NODES="$nodes"
+    bash "$(dirname "${BASH_SOURCE[0]}")/native_power_collect.sh" \
+        "$power_dir" "$control_dir" "$vendor" "$rank" "$role" "$indices" "$nodes" &
+    POWERX_COLLECTOR_PID=$!
+}
+
+powerx_write_control() {
+    local path="$POWERX_CONTROL_DIR/$1"
+    local pending="$path.tmp"
+    printf '%s\n' "$2" > "$pending" || return
+    if [[ -n "${POWERX_HOST_UID:-}" && -n "${POWERX_HOST_GID:-}" ]]; then
+        chown "$POWERX_HOST_UID:$POWERX_HOST_GID" "$pending" || return
+    fi
+    mv -f "$pending" "$path"
+}
+
+powerx_wait_collectors() {
+    local phase="$1" deadline=$((SECONDS + ${POWERX_BARRIER_TIMEOUT_S:-60})) rank pending failed
+    while :; do
+        pending=0
+        failed=0
+        for ((rank=0; rank<POWERX_NUM_NODES; rank++)); do
+            if [[ "$phase" == ready && -f "$POWERX_CONTROL_DIR/done-$rank" ]]; then
+                echo "PowerX collector $rank stopped before benchmark readiness" >&2
+                return 1
+            fi
+            if [[ ! -f "$POWERX_CONTROL_DIR/$phase-$rank" ]]; then
+                pending=1
+            elif [[ "$phase" == done && "$(cat "$POWERX_CONTROL_DIR/done-$rank")" != 0 ]]; then
+                failed=1
+            fi
+        done
+        if [[ "$pending" == 0 ]]; then
+            [[ "$failed" == 0 ]] || echo "One or more PowerX collectors failed" >&2
+            return "$failed"
+        fi
+        if (( SECONDS >= deadline )); then
+            echo "Timed out waiting for PowerX $phase receipts" >&2
+            return 1
+        fi
+        sleep 1
+    done
+}
+
+powerx_stop_collectors() {
+    local rc=0
+    powerx_write_control stop stop || rc=$?
+    powerx_wait_collectors done || rc=$?
+    powerx_reap_collector || rc=$?
+    return "$rc"
+}
+
+powerx_reap_collector() {
+    [[ -n "${POWERX_COLLECTOR_PID:-}" ]] || return 0
+    local deadline=$((SECONDS + ${POWERX_BARRIER_TIMEOUT_S:-60})) rc=0
+    while kill -0 "$POWERX_COLLECTOR_PID" 2>/dev/null; do
+        if (( SECONDS >= deadline )); then
+            kill -TERM "$POWERX_COLLECTOR_PID" 2>/dev/null || true
+            # The shared AMD monitor drains for three seconds before writing receipts.
+            local grace_deadline=$((SECONDS + 5))
+            while kill -0 "$POWERX_COLLECTOR_PID" 2>/dev/null && (( SECONDS < grace_deadline )); do
+                sleep 1
+            done
+            kill -KILL "$POWERX_COLLECTOR_PID" 2>/dev/null || true
+            rc=1
+            break
+        fi
+        sleep 1
+    done
+    wait "$POWERX_COLLECTOR_PID" || rc=$?
+    POWERX_COLLECTOR_PID=""
+    return "$rc"
+}
diff --git a/docs/results-and-ingestion.md b/docs/results-and-ingestion.md
index 7ba8d08fb..725cac79d 100644
--- a/docs/results-and-ingestion.md
+++ b/docs/results-and-ingestion.md
@@ -108,6 +108,12 @@ The PR changelog selects representative NVIDIA and AMD coverage, not an exhausti
 
 Processing and diagnostic power-audit uploads run after launcher or validation failure, retaining raw and aggregate JSON. Normal `bmk_*` upload requires successful benchmark and processing steps, so an incomplete batch or failed Slurm job does not publish diagnostic rows. The main-branch ingest trigger can still publish other successful configurations from a partially failed sweep; it does not establish complete fleet coverage. Downstream importers can use the retained outcome to reject explicitly failed benchmarks.
 
+### Native multinode telemetry
+
+`native_power_collect.sh` and `native_power_lifecycle.sh` provide per-node collection and bounded ready/stop receipts. Launchers opt into the native package under `LOGS/native_power`; this prerequisite enables no new recipe. The adapter validates serving GPU identity, synchronized clocks, collector completion, and complete formal-window coverage. It preserves per-node failures, sample counts, and collector revision in the audit.
+
+The native collector sets UTC and records context beside its CSV for portable replay; existing benchmark monitors keep their current behavior. Its launcher integration requires separate hardware qualification. The offline adapter accepts this context without changing producers. Unusable samples outside the formal window do not establish coverage; `boundary_degenerate_rows` retains their per-GPU counts.
+
 ## Eval artifacts
 
 ### Per-config identity and collection
diff --git a/docs/results-and-ingestion_zh.md b/docs/results-and-ingestion_zh.md
index 1be0fbb0b..7d6ec24a0 100644
--- a/docs/results-and-ingestion_zh.md
+++ b/docs/results-and-ingestion_zh.md
@@ -108,6 +108,12 @@ PR changelog 选择具有代表性的 NVIDIA 和 AMD 覆盖，并非所有受影
 
 启动器或验证失败后仍会运行处理和功耗诊断上传，并在审计工件中保留原始及聚合 JSON。正常 `bmk_*` 上传要求基准和处理步骤成功，因此不完整批次或 Slurm 失败不会发布诊断数据。主分支的入库触发器仍可发布部分失败 sweep 中其他成功配置的数据；这并不证明整个硬件范围已完成覆盖。下游导入器可利用保留的状态拒绝明确失败的基准结果。
 
+### 原生多节点遥测
+
+`native_power_collect.sh` 和 `native_power_lifecycle.sh` 提供每节点采集及有时限的就绪/停止状态文件。启动器可使用 `LOGS/native_power` 下的原生产物；此前置改动不会启用新 recipe。适配器验证服务 GPU 身份、时钟同步、采集完成及正式窗口完整覆盖，并在审计中保留节点故障、样本数和采集器版本。
+
+原生采集器单独设置 UTC，并在 CSV 旁记录上下文以支持跨环境回放；现有基准监控行为保持不变。启动器接入需要另行完成硬件验证。离线适配器接受该上下文，不改变现有生产端。正式窗口外的无效样本不能构成覆盖；`boundary_degenerate_rows` 保留其逐 GPU 计数。
+
 ## 评测工件
 
 ### 单配置身份和收集
diff --git a/infx/results/fixed_sequence.py b/infx/results/fixed_sequence.py
index 3db53a3b7..3e858a268 100644
--- a/infx/results/fixed_sequence.py
+++ b/infx/results/fixed_sequence.py
@@ -248,6 +248,21 @@ def aggregate_power_result(
         expected_num_gpus = int(env['TP']) * int(env.get('PP_SIZE', '1')) * int(env.get('PCP_SIZE', '1'))
     try:
         if is_multinode:
+            native_dir = Path(env.get('POWERX_NATIVE_DIR', 'LOGS/native_power'))
+            if env.get('POWERX_NATIVE_DIR') or native_dir.is_dir():
+                if source.is_dir() and source != native_dir:
+                    raise ValueError('Both native and SRT power packages are present')
+                source = native_dir
+                from .power.native_multinode import run
+
+                return run(
+                    native_dir, bench_path, agg_path,
+                    expected_prefill_gpus=prefill_gpus,
+                    expected_decode_gpus=decode_gpus,
+                    expected_aggregate_gpus=aggregate_gpus,
+                    validation_result=validation_path,
+                    require_power=require_power,
+                )
             from .power.multinode import run
 
             return run(
diff --git a/infx/results/power/native_multinode.py b/infx/results/power/native_multinode.py
new file mode 100644
index 000000000..78ff89889
--- /dev/null
+++ b/infx/results/power/native_multinode.py
@@ -0,0 +1,287 @@
+"""Validate native SMI traces from each node of one fixed-sequence deployment.
+
+This format is owned by InferenceX. It does not claim the srt-slurm/DCGM wire
+contract. The launcher records real serving-device membership and synchronized
+host clocks; the normal result processor binds every trace to the client window.
+"""
+from __future__ import annotations
+
+import argparse
+import csv
+import hashlib
+import json
+import math
+import os
+import socket
+import time
+from datetime import timezone
+from pathlib import Path
+
+from . import ALL_POWER_METRIC_KEYS
+from .common import (
+    _load_benchmark_data, _write_json_atomic, benchmark_window_payload,
+    patch_power_metrics,
+)
+from .single_node import _derived_metrics, _detect_columns, _parse_timestamp, integrate_power
+
+
+def _identity(path: Path, vendor: str) -> dict[str, str]:
+    """Map the vendor enumeration index to a physical UUID; never use PCI alone."""
+    if vendor == "nvidia":
+        with path.open(newline="") as stream:
+            rows = []
+            for row in csv.DictReader(stream, skipinitialspace=True):
+                if any(not isinstance(k, str) or not isinstance(v, str) for k, v in row.items()):
+                    raise ValueError("invalid_device_identity")
+                rows.append({k.strip().lower(): v.strip() for k, v in row.items()})
+    elif vendor == "amd":
+        payload = json.loads(path.read_text())
+        rows = []
+
+        def visit(value: object) -> None:
+            if isinstance(value, list):
+                for item in value:
+                    visit(item)
+            elif isinstance(value, dict):
+                row = {str(k).lower(): v for k, v in value.items()}
+                if "gpu" in row and "uuid" in row:
+                    rows.append(row)
+                else:
+                    for item in value.values():
+                        visit(item)
+
+        visit(payload)
+    else:
+        raise ValueError("unsupported_native_vendor")
+    identities = {}
+    for row in rows:
+        index = str(row.get("index", row.get("gpu", ""))).strip()
+        uuid = str(row.get("uuid", "")).strip()
+        if not index.isdigit() or not uuid or uuid.lower() in {"n/a", "none", "null"}:
+            raise ValueError("invalid_device_identity")
+        if index in identities or uuid in identities.values():
+            raise ValueError("duplicate_device_identity")
+        identities[index] = uuid
+    if not identities:
+        raise ValueError("device_identity_missing")
+    return identities
+
+
+def record_begin(directory: Path, *, vendor: str, node: str, rank: int,
+                 role: str, gpu_indices: list[int], num_nodes: int, job_id: str,
+                 clock_synchronized: bool, revision: str) -> None:
+    if (role not in {"prefill", "decode", "aggregate"} or rank < 0 or
+            num_nodes <= rank or not gpu_indices or min(gpu_indices) < 0 or
+            len(set(gpu_indices)) != len(gpu_indices)):
+        raise ValueError("invalid_native_topology")
+    directory.mkdir(parents=True, exist_ok=True)
+    _write_json_atomic(directory / "manifest.json", {
+        "schema_version": 1, "collector": "inferencex-native-smi",
+        "vendor": vendor, "node": node, "rank": rank, "role": role,
+        "selected_gpu_indices": gpu_indices, "expected_num_nodes": num_nodes,
+        "job_id": job_id, "collector_revision": revision,
+        "clock_source": "utc_ntp", "clock_synchronized": clock_synchronized,
+        "clock_observation": "timedatectl NTPSynchronized on serving host; no measured clock offset",
+        "collection_start_unix": time.time(), "lifecycle": "collecting",
+    })
+
+
+def record_end(directory: Path, *, collector_exit_code: int) -> None:
+    manifest = json.loads((directory / "manifest.json").read_text())
+    manifest.update(collection_end_unix=time.time(),
+                    collector_exit_code=collector_exit_code,
+                    lifecycle="complete" if collector_exit_code == 0 else "failed")
+    _write_json_atomic(directory / "manifest.json", manifest)
+
+
+def run(power_dir: Path, bench_result: Path, agg_result: Path, *,
+        expected_prefill_gpus: int, expected_decode_gpus: int, expected_aggregate_gpus: int = 0,
+        validation_result: Path | None = None, require_power: bool = False) -> int:
+    validation_result = validation_result or bench_result.with_name(
+        f"power_validation_{bench_result.stem}.json")
+    benchmark, reasons = _load_benchmark_data(bench_result)
+    expected_gpus = expected_prefill_gpus + expected_decode_gpus + expected_aggregate_gpus
+    if expected_aggregate_gpus and (expected_prefill_gpus or expected_decode_gpus):
+        reasons.append("native_mixed_aggregate_role_topology")
+    roles: dict[str, list[str]] = {"prefill": [], "decode": [], "aggregate": []}
+    receipts = []
+    node_errors = []
+    samples = []
+    ranks = []
+    nodes = []
+    jobs = set()
+    revisions = set()
+    expected_nodes = set()
+    paths = sorted(power_dir.glob("node-*/manifest.json"))
+    if not paths:
+        reasons.append("native_manifests_missing")
+    if expected_gpus <= 0 or min(expected_prefill_gpus, expected_decode_gpus, expected_aggregate_gpus) < 0:
+        reasons.append("invalid_expected_gpu_count")
+    for path in paths:
+        try:
+            manifest = json.loads(path.read_text())
+            rank = manifest["rank"]
+            role = manifest["role"]
+            if (manifest.get("schema_version") != 1 or
+                    manifest.get("collector") != "inferencex-native-smi" or
+                    type(rank) is not int or rank < 0 or role not in roles or
+                    path.parent.name != f"node-{rank}"):
+                raise ValueError("invalid_native_manifest")
+            if (not isinstance(manifest.get("node"), str) or not manifest["node"] or
+                    not isinstance(manifest.get("job_id"), str) or
+                    not isinstance(manifest.get("collector_revision"), str) or
+                    type(manifest.get("expected_num_nodes")) is not int or
+                    manifest["expected_num_nodes"] <= 0):
+                raise ValueError("invalid_native_run_identity")
+            ranks.append(rank)
+            nodes.append(manifest["node"])
+            expected_nodes.add(manifest["expected_num_nodes"])
+            jobs.add(manifest["job_id"])
+            revisions.add(manifest["collector_revision"])
+            if (manifest.get("lifecycle") != "complete" or
+                    manifest.get("collector_exit_code") != 0):
+                reasons.append("native_collector_incomplete")
+            if (manifest.get("clock_source") != "utc_ntp" or
+                    manifest.get("clock_synchronized") is not True):
+                reasons.append("native_clock_not_synchronized")
+            start, end = manifest["collection_start_unix"], manifest["collection_end_unix"]
+            if (not all(type(x) in (int, float) and math.isfinite(x) for x in (start, end))
+                    or end <= start or (benchmark is not None and
+                    (start > benchmark.start_unix or end < benchmark.end_unix))):
+                reasons.append("native_collection_window_mismatch")
+            selected = manifest["selected_gpu_indices"]
+            if (not isinstance(selected, list) or not selected or
+                    any(type(i) is not int or i < 0 for i in selected) or
+                    len(set(selected)) != len(selected)):
+                raise ValueError("invalid_native_gpu_selection")
+            suffix = "csv" if manifest["vendor"] == "nvidia" else "json"
+            stem = "gpu_metrics_identity" if suffix == "csv" else "gpu_metrics_devices"
+            first_path = path.parent / f"{stem}.{suffix}"
+            last_path = path.parent / f"{stem}_end.{suffix}"
+            first = _identity(first_path, manifest["vendor"])
+            last = _identity(last_path, manifest["vendor"])
+            selected_ids = {str(i): first[str(i)] for i in selected}
+            if any(last.get(i) != uuid for i, uuid in selected_ids.items()):
+                reasons.append("native_device_identity_changed")
+            previous = {uuid for members in roles.values() for uuid in members}
+            if previous.intersection(selected_ids.values()):
+                reasons.append("native_duplicate_physical_gpu")
+            roles[role].extend(selected_ids.values())
+            csv_path = path.parent / "gpu_metrics.csv"
+            with csv_path.open(newline="") as stream:
+                reader = csv.DictReader(stream, skipinitialspace=True)
+                reader.fieldnames = [c.strip() for c in (reader.fieldnames or [])]
+                t_col, p_col, g_col = _detect_columns(reader.fieldnames)
+                if not all((t_col, p_col, g_col)):
+                    raise ValueError("native_telemetry_columns_missing")
+                for row in reader:
+                    gpu = (row.get(g_col) or "").strip()
+                    if gpu not in first:
+                        reasons.append("native_unknown_device_index")
+                        continue
+                    if gpu not in selected_ids:
+                        continue
+                    timestamp = _parse_timestamp((row.get(t_col) or ""), naive_timezone=timezone.utc)
+                    # Invalid rows are retained for the common validator to reject.
+                    samples.append((timestamp, selected_ids[gpu], row.get(p_col) or ""))
+            receipts.append({**manifest, "physical_gpu_ids": selected_ids,
+                             "manifest_sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
+                             "telemetry_sha256": hashlib.sha256(csv_path.read_bytes()).hexdigest(),
+                             "identity_sha256": hashlib.sha256(first_path.read_bytes()).hexdigest(),
+                             "identity_end_sha256": hashlib.sha256(last_path.read_bytes()).hexdigest()})
+        except (OSError, ValueError, KeyError, TypeError, csv.Error) as exc:
+            reasons.append("native_node_invalid")
+            node_errors.append({"node": path.parent.name, "type": type(exc).__name__, "detail": str(exc)})
+    if (len(expected_nodes) != 1 or not expected_nodes or
+            type(next(iter(expected_nodes), None)) is not int or
+            set(ranks) != set(range(next(iter(expected_nodes), 0))) or
+            len(set(nodes)) != len(nodes) or len(ranks) != len(set(ranks))):
+        reasons.append("native_node_topology_mismatch")
+    if len(jobs) != 1 or "" in jobs or len(revisions) != 1 or "" in revisions:
+        reasons.append("native_run_identity_mismatch")
+    if expected_aggregate_gpus:
+        if roles["prefill"] or roles["decode"] or len(roles["aggregate"]) != expected_aggregate_gpus:
+            reasons.append("native_role_gpu_count_mismatch")
+    elif (roles["aggregate"] or len(roles["prefill"]) != expected_prefill_gpus or
+          len(roles["decode"]) != expected_decode_gpus):
+        reasons.append("native_role_gpu_count_mismatch")
+    combined_path = validation_result.with_name(f"{validation_result.stem}_native.csv")
+    combined_path.parent.mkdir(parents=True, exist_ok=True)
+    with combined_path.open("w", newline="") as stream:
+        writer = csv.writer(stream)
+        writer.writerow(["timestamp", "gpu", "power"])
+        writer.writerows(samples)
+    integration = None
+    if benchmark is not None:
+        integration = integrate_power(combined_path, start_unix=benchmark.start_unix,
+                                      end_unix=benchmark.end_unix, expected_num_gpus=expected_gpus)
+        reasons.extend(integration.invalid_reasons)
+    metrics = {}
+    if not reasons and integration is not None and benchmark is not None:
+        metrics = _derived_metrics(integration, benchmark)
+        if roles["prefill"]:
+            prefill = sum(integration.per_gpu_energy_j[uuid] for uuid in roles["prefill"])
+            metrics.update(prefill_gpu_energy_j=prefill,
+                           prefill_avg_power_w=prefill / benchmark.integration_duration_s / expected_prefill_gpus,
+                           prefill_joules_per_input_token=prefill / benchmark.total_input_tokens)
+        if roles["decode"]:
+            decode = sum(integration.per_gpu_energy_j[uuid] for uuid in roles["decode"])
+            metrics.update(decode_gpu_energy_j=decode,
+                           decode_avg_power_w=decode / benchmark.integration_duration_s / expected_decode_gpus,
+                           decode_joules_per_output_token=decode / benchmark.total_output_tokens)
+    valid = not reasons
+    try:
+        patch_power_metrics(agg_result, metric_keys=ALL_POWER_METRIC_KEYS, power_valid=valid, metrics=metrics)
+    except (OSError, ValueError):
+        reasons.append("aggregate_result_unwritable")
+        valid, metrics = False, {}
+    audit = {"schema_version": 1, "telemetry_kind": "native_multinode_smi",
+             "power_valid": valid, "reasons": list(dict.fromkeys(reasons)),
+             "benchmark_result": str(bench_result),
+             "benchmark_result_sha256": hashlib.sha256(bench_result.read_bytes()).hexdigest() if bench_result.is_file() else None,
+             "benchmark_window": benchmark_window_payload(benchmark),
+             "expected_gpu_count": expected_gpus, "nodes": receipts,
+             "node_errors": node_errors,
+             "observed_gpu_count": integration.observed_num_gpus if integration else 0,
+             "per_gpu_role": {uuid: role for role, uuids in roles.items() for uuid in uuids},
+             "per_gpu_sample_counts": integration.per_gpu_sample_counts if integration else {},
+             "boundary_degenerate_rows": integration.boundary_degenerate_rows if integration else {},
+             "per_gpu_max_sample_gap_s": integration.per_gpu_max_sample_gap_s if integration else {},
+             "producer": {"name": "inferencex-native-smi", "revisions": sorted(revisions),
+                          "producer_git_commit": next(iter(revisions)) if len(revisions) == 1 else None},
+             "integration_method": "per_device_trapezoidal_with_linear_boundary_interpolation",
+             "power_percentile_method": "time_weighted_synchronized_total_piecewise_linear",
+             "metrics": metrics}
+    _write_json_atomic(validation_result, audit)
+    return int(require_power and not valid)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    sub = parser.add_subparsers(dest="action", required=True)
+    begin = sub.add_parser("begin")
+    begin.add_argument("--directory", type=Path, required=True)
+    begin.add_argument("--vendor", choices=("amd", "nvidia"), required=True)
+    begin.add_argument("--node", default=os.environ.get("POWERX_NODE_NAME", socket.gethostname()))
+    begin.add_argument("--rank", type=int, required=True)
+    begin.add_argument("--role", choices=("prefill", "decode", "aggregate"), required=True)
+    begin.add_argument("--gpu-indices", required=True)
+    begin.add_argument("--num-nodes", type=int, required=True)
+    begin.add_argument("--job-id", default=os.environ.get("SLURM_JOB_ID", ""))
+    begin.add_argument("--revision", default=os.environ.get("POWERX_COLLECTOR_REVISION", ""))
+    begin.add_argument("--clock-synchronized", choices=("true", "false"), default="false")
+    end = sub.add_parser("end")
+    end.add_argument("--directory", type=Path, required=True)
+    end.add_argument("--collector-exit-code", type=int, required=True)
+    args = vars(parser.parse_args())
+    action = args.pop("action")
+    if action == "begin":
+        args["gpu_indices"] = [int(i) for i in args["gpu_indices"].split(",")]
+        args["clock_synchronized"] = args["clock_synchronized"] == "true"
+        record_begin(**args)
+    else:
+        record_end(**args)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/infx/results/power/single_node.py b/infx/results/power/single_node.py
index 54669a1a8..4ce1bcaaf 100644
--- a/infx/results/power/single_node.py
+++ b/infx/results/power/single_node.py
@@ -9,7 +9,11 @@
 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.
+sidecar is the canonical source for boolean validity and reason codes. Rows in
+the ingest band but outside the formal window whose power is missing,
+non-finite, or <= 0 are teardown noise: they are skipped and counted in the
+sidecar's ``boundary_degenerate_rows`` instead of poisoning validity or faking
+window bracketing.
 """
 
 from __future__ import annotations
@@ -22,7 +26,7 @@
 import os
 import re
 import sys
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from pathlib import Path
 from statistics import mean
@@ -65,6 +69,10 @@ class PowerIntegration:
     per_gpu_max_sample_gap_s: dict[str, float]
     per_gpu_energy_j: dict[str, float]
     device_issues: dict[str, list[str]]
+    # Rows in the ingest band but outside the formal window whose power was
+    # missing/N-A/non-finite/<=0, skipped and counted per GPU; "unknown"
+    # buckets rows without a GPU identity.
+    boundary_degenerate_rows: dict[str, int] = field(default_factory=dict)
     avg_power_w: float | None = None
     p75_power_w: float | None = None
     p75_total_gpu_power_w: float | None = None
@@ -78,7 +86,7 @@ def observed_num_gpus(self) -> int:
         return len(self.observed_gpu_ids)
 
 
-def _parse_timestamp(value: str) -> float | None:
+def _parse_timestamp(value: str, *, naive_timezone: timezone | None = None) -> float | None:
     """Best-effort timestamp parse to Unix epoch seconds (local wall clock).
 
     Handles the formats observed in practice:
@@ -96,7 +104,7 @@ def _parse_timestamp(value: str) -> float | None:
     # nvidia-smi: "YYYY/MM/DD HH:MM:SS.ffffff"
     for fmt in ("%Y/%m/%d %H:%M:%S.%f", "%Y/%m/%d %H:%M:%S"):
         try:
-            return datetime.strptime(value, fmt).timestamp()
+            return datetime.strptime(value, fmt).replace(tzinfo=naive_timezone).timestamp()
         except ValueError:
             pass
     # ISO 8601 (amd-smi variants). fromisoformat tolerates 'T' or space separator
@@ -108,7 +116,7 @@ def _parse_timestamp(value: str) -> float | None:
         return None
     if dt.tzinfo is None:
         # Treat naive timestamps as local time (matches nvidia-smi convention).
-        return dt.timestamp()
+        return dt.replace(tzinfo=naive_timezone).timestamp()
     return dt.astimezone(timezone.utc).timestamp()
 
 
@@ -146,6 +154,16 @@ def _detect_columns(header: list[str]) -> tuple[str | None, str | None, str | No
     return timestamp_col, power_col, gpu_col
 
 
+def _telemetry_timezone(csv_path: Path) -> timezone | None:
+    context = csv_path.with_name(f"{csv_path.stem}_context.json")
+    if not context.is_file():
+        return None
+    payload = json.loads(context.read_text())
+    if not isinstance(payload, dict) or payload.get("timestamp_timezone") != "UTC":
+        raise ValueError("unsupported_telemetry_timezone")
+    return timezone.utc
+
+
 def aggregate_power(
     csv_path: Path,
     start_unix: float,
@@ -162,6 +180,7 @@ def aggregate_power(
         return None
 
     try:
+        timestamp_timezone = _telemetry_timezone(csv_path)
         with csv_path.open("r", newline="", encoding="utf-8", errors="replace") as f:
             reader = csv.DictReader(f, skipinitialspace=True)
             header = [c.strip() for c in (reader.fieldnames or [])]
@@ -189,7 +208,7 @@ def aggregate_power(
             for row in reader:
                 ts_raw = (row.get(timestamp_col) or "").strip()
                 pw_raw = (row.get(power_col) or "").strip()
-                ts = _parse_timestamp(ts_raw)
+                ts = _parse_timestamp(ts_raw, naive_timezone=timestamp_timezone)
                 pw = _parse_power(pw_raw)
                 if ts is None or pw is None:
                     continue
@@ -204,7 +223,7 @@ def aggregate_power(
                     if gpu_id:
                         per_sample_gpus.setdefault(bucket, set()).add(gpu_id)
                         gpu_keys.add(gpu_id)
-    except (OSError, csv.Error):
+    except (OSError, csv.Error, ValueError):
         return None
 
     if not per_sample_total:
@@ -237,6 +256,7 @@ def _empty_integration(
     *,
     expected_num_gpus: int | None,
     reasons: list[str],
+    boundary_degenerate_rows: dict[str, int] | None = None,
 ) -> PowerIntegration:
     """Build an invalid integration result when no device data is available."""
     return PowerIntegration(
@@ -248,6 +268,7 @@ def _empty_integration(
         per_gpu_max_sample_gap_s={},
         per_gpu_energy_j={},
         device_issues={},
+        boundary_degenerate_rows=boundary_degenerate_rows or {},
     )
 
 
@@ -308,8 +329,10 @@ def integrate_power(
     # 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]]] = {}
+    boundary_degenerate: dict[str, int] = {}
     saw_missing_gpu_identity = False
     try:
+        timestamp_timezone = _telemetry_timezone(csv_path)
         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 [])]
@@ -332,7 +355,7 @@ def integrate_power(
                 )
 
             for row in reader:
-                timestamp = _parse_timestamp((row.get(timestamp_col) or "").strip())
+                timestamp = _parse_timestamp((row.get(timestamp_col) or "").strip(), naive_timezone=timestamp_timezone)
                 if timestamp is None or not math.isfinite(timestamp):
                     _append_reason(reasons, "invalid_timestamp_sample")
                     continue
@@ -348,6 +371,15 @@ def integrate_power(
 
                 power = _parse_power((row.get(power_col) or "").strip())
                 gpu_id = (row.get(gpu_col) or "").strip()
+                if (power is None or not math.isfinite(power) or power <= 0.0) and (
+                    timestamp < start_unix or timestamp > end_unix
+                ):
+                    # SMI teardown rows can carry N/A or 0 W cells: outside the
+                    # formal window they are counted, never used to satisfy
+                    # bracketing or to poison in-window validity.
+                    key = gpu_id or "unknown"
+                    boundary_degenerate[key] = boundary_degenerate.get(key, 0) + 1
+                    continue
                 if power is None:
                     _append_reason(reasons, "invalid_power_sample")
                     continue
@@ -359,11 +391,12 @@ def integrate_power(
                     continue
                 values = raw_samples.setdefault(gpu_id, {}).setdefault(timestamp, [])
                 values.append(power)
-    except (OSError, csv.Error):
+    except (OSError, csv.Error, ValueError):
         _append_reason(reasons, "telemetry_file_unreadable")
         return _empty_integration(
             expected_num_gpus=expected_num_gpus,
             reasons=reasons,
+            boundary_degenerate_rows=boundary_degenerate,
         )
 
     if saw_missing_gpu_identity:
@@ -373,6 +406,7 @@ def integrate_power(
         return _empty_integration(
             expected_num_gpus=expected_num_gpus,
             reasons=reasons,
+            boundary_degenerate_rows=boundary_degenerate,
         )
 
     observed_gpu_ids = tuple(sorted(raw_samples, key=_gpu_sort_key))
@@ -450,6 +484,7 @@ def integrate_power(
         per_gpu_max_sample_gap_s=per_gpu_max_sample_gap_s,
         per_gpu_energy_j=per_gpu_energy_j,
         device_issues=device_issues,
+        boundary_degenerate_rows=boundary_degenerate,
         avg_power_w=avg_power_w,
         p75_power_w=p75_total / len(observed_gpu_ids) if p75_total is not None else None,
         p75_total_gpu_power_w=p75_total,
@@ -485,6 +520,7 @@ def _read_energy_snapshot(path: Path) -> dict[str, float] | None:
 def _stream_samples_by_gpu(csv_path: Path) -> dict[str, list[tuple[float, float]]]:
     """Group every parseable (timestamp, watt) sample per GPU, sorted in time."""
     samples: dict[str, list[tuple[float, float]]] = {}
+    timestamp_timezone = _telemetry_timezone(csv_path)
     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 [])]
@@ -493,7 +529,7 @@ def _stream_samples_by_gpu(csv_path: Path) -> dict[str, list[tuple[float, float]
         if not timestamp_col or not power_col or not gpu_col:
             return {}
         for row in reader:
-            timestamp = _parse_timestamp((row.get(timestamp_col) or "").strip())
+            timestamp = _parse_timestamp((row.get(timestamp_col) or "").strip(), naive_timezone=timestamp_timezone)
             power = _parse_power((row.get(power_col) or "").strip())
             gpu_id = (row.get(gpu_col) or "").strip()
             if timestamp is None or power is None or not gpu_id:
@@ -679,6 +715,7 @@ def _validation_payload(
         "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,
+        "boundary_degenerate_rows": integration.boundary_degenerate_rows,
         "accumulator_check": accumulator_check,
         "metrics": audit_metrics(metrics),
     }
@@ -754,7 +791,7 @@ def run(
 
     try:
         accumulator_check = cross_check_accumulator(csv_path)
-    except (OSError, csv.Error):
+    except (OSError, csv.Error, ValueError):
         accumulator_check = {"available": False, "reason": "cross_check_error"}
 
     try:
diff --git a/runners/test_native_collector_barriers.py b/runners/test_native_collector_barriers.py
new file mode 100644
index 000000000..14c8fe2ce
--- /dev/null
+++ b/runners/test_native_collector_barriers.py
@@ -0,0 +1,58 @@
+"""CPU-only lifecycle checks with controlled external collector/Slurm processes."""
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+LIFECYCLE = ROOT / 'benchmarks/native_power_lifecycle.sh'
+JOB = ROOT / 'benchmarks/multi_node/llm-d/job.slurm'
+
+
+@pytest.mark.parametrize('failed_rank', [None, 1])
+def test_stop_waits_for_every_collector_and_retains_failure(tmp_path, failed_rank):
+    command = '''
+source "$1"
+export POWERX_CONTROL_DIR="$2" POWERX_NUM_NODES=2 POWERX_BARRIER_TIMEOUT_S=5
+(while [[ ! -f "$2/stop" ]]; do sleep 0.01; done; sleep 0.1; echo 0 > "$2/done-0") &
+POWERX_COLLECTOR_PID=$!
+(while [[ ! -f "$2/stop" ]]; do sleep 0.01; done; sleep 0.2; echo "$3" > "$2/done-1") &
+remote_pid=$!
+powerx_stop_collectors
+rc=$?
+wait "$remote_pid"
+exit "$rc"
+'''
+    result = subprocess.run(['bash', '-c', command, 'bash', str(LIFECYCLE),
+                             str(tmp_path), '1' if failed_rank else '0'],
+                            capture_output=True, text=True, timeout=10)
+    assert result.returncode == (1 if failed_rank else 0), result.stderr
+    assert (tmp_path / 'done-0').read_text().strip() == '0'
+    assert (tmp_path / 'done-1').read_text().strip() == ('1' if failed_rank else '0')
+
+
+def test_ready_barrier_rejects_collector_that_already_stopped(tmp_path):
+    (tmp_path / 'ready-0').write_text('ready')
+    (tmp_path / 'done-0').write_text('1')
+    result = subprocess.run(['bash', '-c', 'source "$1"; POWERX_CONTROL_DIR="$2"; '
+                             'POWERX_NUM_NODES=1; powerx_wait_collectors ready',
+                             'bash', str(LIFECYCLE), str(tmp_path)], capture_output=True, text=True)
+    assert result.returncode == 1
+    assert 'stopped before benchmark readiness' in result.stderr
+
+
+def test_control_receipt_is_owned_before_publication(tmp_path):
+    command = '''source "$1"
+POWERX_CONTROL_DIR=$2
+POWERX_HOST_UID=1000 POWERX_HOST_GID=1000
+chown() { [[ ! -e "$POWERX_CONTROL_DIR/stop" ]]; }
+powerx_write_control stop stop
+'''
+    result = subprocess.run(['bash', '-c', command, 'bash', str(LIFECYCLE), str(tmp_path)],
+                            capture_output=True, text=True, timeout=5)
+    assert result.returncode == 0, result.stderr
+    assert (tmp_path / 'stop').read_text() == 'stop\n'
+    assert not list(tmp_path.glob('*.tmp'))
diff --git a/runners/test_native_collector_receipts.py b/runners/test_native_collector_receipts.py
new file mode 100644
index 000000000..d9b966296
--- /dev/null
+++ b/runners/test_native_collector_receipts.py
@@ -0,0 +1,140 @@
+import subprocess
+from pathlib import Path
+ROOT = Path(__file__).resolve().parents[1]
+
+def test_native_control_receipt_is_owned_before_publication(tmp_path):
+    source = (ROOT / 'benchmarks/native_power_collect.sh').read_text()
+    function = source[source.index('write_control() {'):source.index('\nfinish() {')]
+    result = subprocess.run(['bash', '-c', function + '''
+control_dir=$1
+POWERX_HOST_UID=1000 POWERX_HOST_GID=1000
+chown() { [[ ! -e "$control_dir/done-0" ]]; }
+write_control done-0 7
+''', 'bash', str(tmp_path)], capture_output=True, text=True, timeout=5)
+    assert result.returncode == 0, result.stderr
+    assert (tmp_path / 'done-0').read_text() == '7\n'
+    assert not list(tmp_path.glob('*.tmp'))
+
+
+def _collector_with_monitor(tmp_path, *, alive, end_identity_rc=0):
+    import shutil
+    import sys
+
+    scripts = tmp_path / 'repo/benchmarks'
+    scripts.mkdir(parents=True)
+    shutil.copyfile(ROOT / 'benchmarks/native_power_collect.sh', scripts / 'native_power_collect.sh')
+    library = (ROOT / 'benchmarks/benchmark_lib.sh').read_text()
+    functions = []
+    for name in ['_background_process_is_running', '_write_amd_smi_sidecar']:
+        start = library.index(name + '() {')
+        functions.append(library[start:library.index('\n}', start) + 2])
+    (scripts / 'benchmark_lib.sh').write_text('\n'.join(functions) + '''
+GPU_MONITOR_PID="" GPU_MONITOR_VENDOR=""
+start_gpu_monitor() {
+    GPU_MONITOR_VENDOR=amd
+    if [[ "$MONITOR_ALIVE" == 1 ]]; then
+        sleep 30 &
+        GPU_MONITOR_PID=$!
+    else
+        false &
+        GPU_MONITOR_PID=$!
+        wait "$GPU_MONITOR_PID" || true
+    fi
+}
+stop_gpu_monitor() {
+    kill "$GPU_MONITOR_PID" 2>/dev/null || true
+    wait "$GPU_MONITOR_PID" 2>/dev/null || true
+}
+amd-smi() {
+    if [[ -f "$IDENTITY_CALLED" && "$END_IDENTITY_RC" != 0 ]]; then return "$END_IDENTITY_RC"; fi
+    touch "$IDENTITY_CALLED"
+    printf '[{"gpu":0,"uuid":"test-gpu"}]\\n'
+}
+''')
+    control = tmp_path / 'control'
+    control.mkdir()
+    (control / 'stop').touch()
+    power = tmp_path / 'power'
+    result = subprocess.run(['bash', str(scripts / 'native_power_collect.sh'), str(power),
+                             str(control), 'amd', '0', 'prefill', '0', '1'],
+                            env={'PATH': f'{Path(sys.executable).parent}:/usr/bin:/bin',
+                                 'PYTHONPATH': str(ROOT), 'MONITOR_ALIVE': str(int(alive)),
+                                 'END_IDENTITY_RC': str(end_identity_rc),
+                                 'IDENTITY_CALLED': str(tmp_path / 'identity-called')},
+                            capture_output=True, text=True, timeout=5)
+    return result, control, power
+
+
+def test_dead_monitor_never_publishes_ready(tmp_path):
+    import json
+
+    result, control, power = _collector_with_monitor(tmp_path, alive=False)
+    assert not (control / 'ready-0').exists(), result.stderr
+    assert (control / 'done-0').read_text().strip() != '0'
+    assert json.loads((power / 'manifest.json').read_text())['lifecycle'] == 'failed'
+
+
+def test_amd_end_identity_failure_is_not_a_successful_done_receipt(tmp_path):
+    import json
+
+    result, control, power = _collector_with_monitor(tmp_path, alive=True, end_identity_rc=7)
+    assert (control / 'ready-0').exists(), result.stderr
+    assert (control / 'done-0').read_text().strip() != '0'
+    assert json.loads((power / 'manifest.json').read_text())['lifecycle'] == 'failed'
+
+
+def test_live_monitor_with_successful_identity_completes(tmp_path):
+    import json
+
+    result, control, power = _collector_with_monitor(tmp_path, alive=True)
+    assert result.returncode == 0, result.stderr
+    assert (control / 'ready-0').exists()
+    assert (control / 'done-0').read_text().strip() == '0'
+    assert json.loads((power / 'manifest.json').read_text())['lifecycle'] == 'complete'
+
+
+def test_native_timezone_is_scoped_to_collector(tmp_path):
+    import json
+    import os
+    import sys
+
+    binary = tmp_path / 'bin'
+    binary.mkdir()
+    smi = binary / 'nvidia-smi'
+    smi.write_text(f'''#!{sys.executable}
+import os, sys, time
+with open(os.environ['SMI_ENV_LOG'], 'a') as output:
+    print(os.environ.get('TZ', ''), file=output, flush=True)
+if '-l' in sys.argv:
+    print('timestamp, index, power.draw [W]', flush=True)
+    print('2026/09/12 00:00:00.000, 0, 100', flush=True)
+    time.sleep(30)
+else:
+    print('index, uuid, pci.bus_id, name, driver_version')
+    print('0, GPU-fixture, 0000:00:00.0, fixture, fixture')
+''')
+    smi.chmod(0o755)
+    env = {**os.environ, 'PATH': f'{binary}:{Path(sys.executable).parent}:/usr/bin:/bin',
+           'TZ': 'Pacific/Honolulu', 'SMI_ENV_LOG': str(tmp_path / 'ordinary-env')}
+    ordinary = tmp_path / 'ordinary.csv'
+    result = subprocess.run(['bash', '-c', '''source "$1"
+start_gpu_monitor --output "$2"
+stop_gpu_monitor
+''', 'bash', str(ROOT / 'benchmarks/benchmark_lib.sh'), str(ordinary)],
+                            env=env, capture_output=True, text=True, timeout=5)
+    assert result.returncode == 0, result.stderr
+    assert set((tmp_path / 'ordinary-env').read_text().splitlines()) == {'Pacific/Honolulu'}
+    assert not ordinary.with_name('ordinary_context.json').exists()
+
+    control = tmp_path / 'control'
+    control.mkdir()
+    (control / 'stop').touch()
+    power = tmp_path / 'native'
+    env['SMI_ENV_LOG'] = str(tmp_path / 'native-env')
+    result = subprocess.run(['bash', str(ROOT / 'benchmarks/native_power_collect.sh'),
+                             str(power), str(control), 'nvidia', '0', 'aggregate', '0', '1'],
+                            env=env, capture_output=True, text=True, timeout=5)
+    assert result.returncode == 0, result.stderr
+    assert set((tmp_path / 'native-env').read_text().splitlines()) == {'UTC'}
+    assert json.loads((power / 'gpu_metrics_context.json').read_text()) == {'timestamp_timezone': 'UTC'}
+    assert (control / 'done-0').read_text().strip() == '0'
diff --git a/utils/agentic/aggregation/test_power_lifecycle.py b/utils/agentic/aggregation/test_power_lifecycle.py
index 20710e68e..6d955af12 100644
--- a/utils/agentic/aggregation/test_power_lifecycle.py
+++ b/utils/agentic/aggregation/test_power_lifecycle.py
@@ -280,8 +280,13 @@ def test_signal_stops_monitor_once_without_replacing_parent_trap(
 }}
 stop_gpu_monitor() {{ printf 'monitor-stop\n' >> {str(event_log)!r}; }}
 fake_replay() {{
-    printf 'replay-ready\n' >> {str(event_log)!r}
-    exec sleep 30
+    exec {sys.executable!r} -c '
+import signal, sys, time
+signal.signal(signal.SIGINT, signal.SIG_DFL)
+signal.signal(signal.SIGTERM, signal.SIG_DFL)
+print("replay-ready", file=open(sys.argv[1], "a"), flush=True)
+time.sleep(30)
+' {str(event_log)!r}
 }}
 trap 'printf "parent-exit\\n" >> {str(event_log)!r}' EXIT
 trap 'printf "parent-int\\n" >> {str(event_log)!r}; exit 130' INT
@@ -301,7 +306,8 @@ def test_signal_stops_monitor_once_without_replacing_parent_trap(
     )
     try:
         # The monitor starts before the production signal traps are installed.
-        # Wait for replay so the signal actually exercises those traps.
+        # Publish readiness from the execed process after restoring signal handling;
+        # a shell marker before exec races with the group SIGINT.
         deadline = time.monotonic() + 5
         while time.monotonic() < deadline:
             if event_log.exists() and "replay-ready" in event_log.read_text().splitlines():
diff --git a/utils/test_aggregate_power.py b/utils/test_aggregate_power.py
index aa3bf23f9..5b8b76121 100644
--- a/utils/test_aggregate_power.py
+++ b/utils/test_aggregate_power.py
@@ -574,6 +574,201 @@ def test_run_rejects_malformed_telemetry_inside_window(
     assert audit["reasons"] == [expected_reason]
 
 
+
+# AMDSMI 26.2.0 `metric -p -c -t -u -w 1 --csv` header (order-faithful subset,
+# measured on MI355X; see test_detect_columns_amd_watch_mode_real_header).
+_MI355X_WATCH_HEADER = (
+    "timestamp,gpu,gfx_activity,umc_activity,mm_activity,vcn_activity,"
+    "jpeg_activity,gfx_busy_inst_xcp_0,jpeg_busy_xcp_0,vcn_busy_xcp_0,"
+    "socket_power,gfx_voltage,soc_voltage,mem_voltage,throttle_status,"
+    "power_management,gfx_0_clk,mem_0_clk,edge,hotspot,mem"
+)
+
+
+def _mi355x_watch_row(timestamp: int, gpu: int, socket_power: str) -> str:
+    """One data row in the shape captured from run 32433563482 (conc1):
+    integer-second epoch, quoted list cells with embedded commas, N/A cells,
+    and a trailing carriage return."""
+    return (
+        f"{timestamp},{gpu},0,0,N/A,\"['N/A', 'N/A', 'N/A', 'N/A']\","
+        "\"['N/A', 'N/A']\",\"[0, 0, 0, 0, 0, 0, 0, 0]\",\"[0, 0]\",\"[0, 0]\","
+        f"{socket_power},N/A,N/A,N/A,N/A,ENABLED,1404,2000,N/A,40,25\r"
+    )
+
+
+def test_integrate_power_skips_na_power_rows_outside_window(tmp_path: Path):
+    """N/A-power teardown rows past the window end are counted, not poisonous."""
+    csv = tmp_path / "gpu_metrics.csv"
+    base = 1_700_000_000.0
+    lines = ["timestamp,gpu,socket_power,temperature"]
+    for offset in range(-2, 13):
+        for gpu in range(2):
+            lines.append(f"{base + offset},{gpu},500.0,65")
+    for gpu in range(2):
+        lines.append(f"{base + 11},{gpu},N/A,65")
+    csv.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+    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 == ()
+    assert result.boundary_degenerate_rows == {"0": 1, "1": 1}
+    assert result.total_gpu_energy_j == pytest.approx(10_000.0)
+
+
+def test_integrate_power_does_not_bracket_with_zero_power_tail(tmp_path: Path):
+    """A 0 W teardown row past the window end must not fake end bracketing.
+
+    Legacy behavior accepted the 0 W row as a valid boundary sample, corrupting
+    the end interpolation with a bogus value; this intentionally flips that
+    case to an explicit benchmark_window_not_bracketed failure."""
+    csv = tmp_path / "gpu_metrics.csv"
+    base = 1_700_000_000.0
+    lines = ["timestamp,gpu,socket_power,temperature"]
+    # Good rows stop at end-4; the only post-end sample per GPU has power 0.
+    for offset in range(-1, 7):
+        for gpu in range(2):
+            lines.append(f"{base + offset},{gpu},500.0,65")
+    for gpu in range(2):
+        lines.append(f"{base + 11},{gpu},0,65")
+    csv.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+    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.boundary_degenerate_rows == {"0": 1, "1": 1}
+
+
+def test_integrate_power_preserves_boundary_counts_without_usable_samples(tmp_path: Path):
+    csv = tmp_path / "gpu_metrics.csv"
+    csv.write_text(
+        "timestamp,gpu,socket_power\n"
+        "1699999999,0,N/A\n"
+        "1700000011,0,0\n"
+        "1700000011,1,N/A\n",
+        encoding="utf-8",
+    )
+
+    result = integrate_power(
+        csv,
+        start_unix=1_700_000_000,
+        end_unix=1_700_000_010,
+        expected_num_gpus=2,
+    )
+
+    assert result.power_valid is False
+    assert "no_usable_power_samples" in result.invalid_reasons
+    assert result.boundary_degenerate_rows == {"0": 2, "1": 1}
+
+
+def test_integrate_power_keeps_zero_power_semantics_inside_window(tmp_path: Path):
+    """Frozen legacy behavior: an in-window 0 W sample still integrates."""
+    csv = tmp_path / "gpu_metrics.csv"
+    base = 1_700_000_000.0
+    lines = ["timestamp,gpu,socket_power,temperature"]
+    for offset in range(-1, 12):
+        watts = "0.0" if offset == 5 else "500.0"
+        lines.append(f"{base + offset},0,{watts},65")
+    csv.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+    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.boundary_degenerate_rows == {}
+    # Trapezoids dip to 0 at t=5: 8 x 500 + 2 x 250 = 4500 J.
+    assert result.total_gpu_energy_j == pytest.approx(4_500.0)
+
+
+def test_run_writes_boundary_degenerate_rows_to_sidecar(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=2,
+    )
+    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,
+        total_output=2_000,
+        total_input=10_000,
+    )
+    agg.write_text(json.dumps({"hw": "mi355x"}), encoding="utf-8")
+
+    exit_code = run(csv, bench, agg, expected_num_gpus=2, validation_result=validation)
+
+    assert exit_code == 0
+    audit = json.loads(validation.read_text())
+    # Present-and-empty for clean streams: readers can rely on the key.
+    assert audit["boundary_degenerate_rows"] == {}
+
+
+def test_integrate_power_regression_mi355x_integer_ticks_end_gap(tmp_path: Path):
+    """Run-32433563482 conc1 regression: amd-smi integer-second ticks stop 4 s
+    before the fractional aiperf window end (last tick 1787277605 vs end
+    ...609.157497), so bracketing fails and the producer-side telemetry loss is
+    attributed as benchmark_window_not_bracketed on every GPU.
+
+    The retrieved artifact's trailing rows all carry valid socket_power
+    (254-264 W) with N/A activity/voltage cells; the N/A- and 0-power teardown
+    rows appended past the window end are the documented synthetic degenerate
+    shapes, asserting they are counted rather than used for bracketing."""
+    csv = tmp_path / "gpu_metrics.csv"
+    start = 1_787_277_560.155891
+    end = 1_787_277_609.157497
+    last_tick = 1_787_277_605
+    powers = [259, 255, 263, 264, 256, 254, 259, 259]
+    lines = [_MI355X_WATCH_HEADER]
+    for tick in range(1_787_277_555, last_tick + 1):
+        for gpu in range(8):
+            lines.append(_mi355x_watch_row(tick, gpu, str(powers[gpu])))
+        # amd-smi watch mode emits a blank line between tick groups.
+        lines.append("")
+    for gpu in range(8):
+        lines.append(_mi355x_watch_row(1_787_277_610, gpu, "N/A"))
+    for gpu in range(8):
+        lines.append(_mi355x_watch_row(1_787_277_611, gpu, "0"))
+    csv.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+    result = integrate_power(
+        csv,
+        start_unix=start,
+        end_unix=end,
+        expected_num_gpus=8,
+    )
+
+    assert result.power_valid is False
+    assert result.invalid_reasons == ("benchmark_window_not_bracketed",)
+    assert result.device_issues == {
+        str(gpu): ["benchmark_window_not_bracketed"] for gpu in range(8)
+    }
+    assert result.boundary_degenerate_rows == {str(gpu): 2 for gpu in range(8)}
+
+
 def test_run_patches_agg_with_power_and_joules(tmp_path: Path):
     base = 1_700_000_000.0
     csv = tmp_path / "gpu_metrics.csv"
@@ -708,6 +903,7 @@ def test_power_replacement_removes_stale_metrics(tmp_path, patch_validated_power
     path.write_text(json.dumps({
         "hw": "fixture", "avg_power_w": 99, "total_gpu_energy_j": 50,
         "power_invalid_reasons": ["stale"],
+        "power_audit": {"source": "previous-run.json"},
     }))
     patch_validated_power(path, power_valid=valid, metrics={
         "avg_power_w": 12.34567, "joules_per_output_token": 0.12345678,
@@ -1433,12 +1629,12 @@ def test_power_percentiles_uses_synchronized_total_not_device_percentiles(tmp_pa
 
 def test_power_percentiles_weights_time_and_clips_the_validated_window(tmp_path):
     csv_path = tmp_path / "power.csv"
-    # Dense readings near the high end must not bias a uniform linear ramp.
-    _write_amd_csv(csv_path, [(0, 0, 0), (1, 0, 100), (1.9, 0, 190), (2, 0, 200)])
+    # Dense readings must not bias the uniform 150-250 W ramp inside the window.
+    _write_amd_csv(csv_path, [(0, 0, 100), (1, 0, 200), (1.9, 0, 290), (2, 0, 300)])
     result = integrate_power(csv_path, start_unix=0.5, end_unix=1.5, expected_num_gpus=1)
     assert result.power_valid
-    assert result.p75_power_w == pytest.approx(125)
-    assert result.p90_power_w == pytest.approx(140)
+    assert result.p75_power_w == pytest.approx(225)
+    assert result.p90_power_w == pytest.approx(240)
 
 
 def test_power_percentiles_is_withheld_for_invalid_telemetry(tmp_path):
@@ -1462,3 +1658,54 @@ def test_power_percentiles_aligns_asynchronous_gpu_samples(tmp_path):
     assert result.p75_power_w == pytest.approx(300)
     assert result.p90_total_gpu_power_w == pytest.approx(600)
     assert result.p90_power_w == pytest.approx(300)
+
+
+@pytest.mark.parametrize('step_name', ['Upload GPU metrics', 'Upload power audit bundle'])
+@pytest.mark.parametrize('directory', ['', 'results'])
+def test_uploaded_telemetry_replays_in_a_different_timezone(tmp_path, step_name, directory):
+    import os
+    import shutil
+    import yaml
+
+    repo = Path(__file__).resolve().parents[1]
+    source = tmp_path / 'source'
+    telemetry = source / directory
+    telemetry.mkdir(parents=True)
+    (telemetry / 'gpu_metrics.csv').write_text(
+        'timestamp,index,power.draw [W]\n'
+        '2024/01/01 00:00:00.000,0,100 W\n'
+        '2024/01/01 00:00:01.000,0,100 W\n'
+        '2024/01/01 00:00:02.000,0,100 W\n')
+    (telemetry / 'gpu_metrics_context.json').write_text('{"timestamp_timezone":"UTC"}')
+    workflow = yaml.safe_load((repo / '.github/workflows/benchmark-tmpl.yml').read_text())
+    step = next(step for job in workflow['jobs'].values() for step in job['steps']
+                if step.get('name') == step_name)
+    archive = tmp_path / 'downloaded'
+    for pattern in step['with']['path'].splitlines():
+        for file in source.glob(pattern):
+            destination = archive / file.relative_to(source)
+            destination.parent.mkdir(parents=True, exist_ok=True)
+            shutil.copy(file, destination)
+    result = subprocess.run(
+        [sys.executable, '-c', """
+import sys, time
+from pathlib import Path
+from infx.results.power.single_node import integrate_power
+time.tzset()
+result = integrate_power(Path(sys.argv[1]), start_unix=1704067200, end_unix=1704067202,
+                         expected_num_gpus=1)
+assert result.power_valid, result.invalid_reasons
+assert result.total_gpu_energy_j == 200
+""", str(archive / directory / 'gpu_metrics.csv')],
+        cwd=tmp_path, env={**os.environ, 'TZ': 'Etc/GMT+8', 'PYTHONPATH': str(repo)},
+        capture_output=True, text=True, timeout=10,
+    )
+    assert result.returncode == 0, result.stderr
+
+
+@pytest.mark.parametrize('context', ['{"timestamp_timezone":"PST"}', '{}', '{invalid'])
+def test_legacy_average_rejects_invalid_telemetry_context(tmp_path, context):
+    csv = tmp_path / 'gpu_metrics.csv'
+    csv.write_text('timestamp,index,power.draw [W]\n1,0,100\n2,0,100\n')
+    csv.with_name('gpu_metrics_context.json').write_text(context)
+    assert aggregate_power(csv, 1, 2) is None
diff --git a/utils/test_native_multinode_power.py b/utils/test_native_multinode_power.py
new file mode 100644
index 000000000..31e563737
--- /dev/null
+++ b/utils/test_native_multinode_power.py
@@ -0,0 +1,293 @@
+"""Native collector acceptance uses independent, hand-computed two-node traces."""
+import json
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+from infx.results.power.native_multinode import record_begin, record_end, run
+from infx.results.power.single_node import integrate_power
+
+REPO = Path(__file__).resolve().parents[1]
+
+
+def _package(tmp_path, vendor="amd"):
+    root = tmp_path / "native_power"
+    for rank, role, watts in ((0, "prefill", 100), (1, "decode", 300)):
+        node = root / f"node-{rank}"
+        record_begin(node, vendor=vendor, node=f"host-{rank}", rank=rank, role=role,
+                     gpu_indices=[0], num_nodes=2, job_id="job-123", revision="revision-abc",
+                     clock_synchronized=True)
+        record_end(node, collector_exit_code=0)
+        manifest = json.loads((node / "manifest.json").read_text())
+        manifest.update(collection_start_unix=0, collection_end_unix=5)
+        (node / "manifest.json").write_text(json.dumps(manifest))
+        for ending in ("", "_end"):
+            if vendor == "amd":
+                (node / f"gpu_metrics_devices{ending}.json").write_text(json.dumps([
+                    {"gpu": 0, "uuid": f"uuid-{rank}"}, {"gpu": 1, "uuid": f"unused-{rank}"}]))
+            else:
+                (node / f"gpu_metrics_identity{ending}.csv").write_text(
+                    f"index, uuid, pci.bus_id\n0, uuid-{rank}, 0000:01:00.0\n1, unused-{rank}, 0000:02:00.0\n")
+        # The spare physical GPU is visible but does not belong to the server.
+        (node / "gpu_metrics.csv").write_text("timestamp,gpu,power\n" + "".join(
+            f"{tick},0,{watts}\n{tick},1,900\n" for tick in range(5)))
+    bench = tmp_path / "result.json"
+    bench.write_text(json.dumps({"benchmark_start_time_unix": 1, "benchmark_end_time_unix": 3,
+                                "duration": 2, "completed": 2, "total_input_tokens": 20,
+                                "total_output_tokens": 10}))
+    agg = tmp_path / "agg.json"
+    agg.write_text(json.dumps({"avg_power_w": 999, "prefill_gpu_energy_j": 999}))
+    return root, bench, agg
+
+
+@pytest.mark.parametrize("vendor", ["amd", "nvidia"])
+def test_native_whole_fleet_and_role_energy_use_only_serving_devices(tmp_path, vendor):
+    root, bench, agg = _package(tmp_path, vendor)
+    assert run(root, bench, agg, expected_prefill_gpus=1, expected_decode_gpus=1, require_power=True) == 0
+    actual = json.loads(agg.read_text())
+    assert actual["power_valid"] == 1
+    assert actual["total_gpu_energy_j"] == 800
+    assert actual["avg_power_w"] == 200
+    assert actual["p90_power_w"] == 200
+    assert actual["p75_power_w"] == 200
+    assert actual["joules_per_successful_query"] == 400
+    assert actual["prefill_gpu_energy_j"] == 200
+    assert actual["decode_gpu_energy_j"] == 600
+    assert actual["prefill_joules_per_input_token"] == 10
+    assert actual["decode_joules_per_output_token"] == 60
+    audit = json.loads((tmp_path / "power_validation_result.json").read_text())
+    assert audit["observed_gpu_count"] == 2
+    assert audit["per_gpu_role"] == {"uuid-0": "prefill", "uuid-1": "decode"}
+    assert len(audit["nodes"][0]["telemetry_sha256"]) == 64
+    from infx.results.power.audit import audit_summary
+    summary = audit_summary(audit, "power_validation_result.json")["power_audit"]
+    assert summary["sample_count"] == 10  # Five samples on each of two participating GPUs.
+    assert summary["producer_sha"] == "revision-abc"
+
+
+def test_native_parse_failure_has_a_public_reason_and_retained_detail(tmp_path):
+    from infx.results.power.audit import audit_summary
+    root, bench, agg = _package(tmp_path)
+    (root / "node-1/manifest.json").write_text("not JSON")
+    assert run(root, bench, agg, expected_prefill_gpus=1, expected_decode_gpus=1, require_power=True) == 1
+    audit = json.loads((tmp_path / "power_validation_result.json").read_text())
+    summary = audit_summary(audit, "power_validation_result.json")
+    assert "native_node_invalid" in summary["power_invalid_reasons"]
+    assert audit["node_errors"][0]["node"] == "node-1"
+    assert audit["node_errors"][0]["detail"]
+
+
+@pytest.mark.parametrize(("field", "value", "reason"), [
+    ("clock_synchronized", False, "native_clock_not_synchronized"),
+    ("lifecycle", "collecting", "native_collector_incomplete"),
+    ("collection_end_unix", 2, "native_collection_window_mismatch"),
+    ("job_id", "other-job", "native_run_identity_mismatch"),
+    ("role", "prefill", "native_role_gpu_count_mismatch"),
+    ("expected_num_nodes", 3, "native_node_topology_mismatch"),
+])
+def test_native_invalid_evidence_clears_stale_metrics_and_writes_audit(tmp_path, field, value, reason):
+    root, bench, agg = _package(tmp_path)
+    path = root / "node-1/manifest.json"
+    manifest = json.loads(path.read_text()); manifest[field] = value
+    path.write_text(json.dumps(manifest))
+    assert run(root, bench, agg, expected_prefill_gpus=1, expected_decode_gpus=1, require_power=True) == 1
+    actual = json.loads(agg.read_text())
+    assert actual["power_valid"] == 0
+    assert "avg_power_w" not in actual
+    assert "prefill_gpu_energy_j" not in actual
+    audit = json.loads((tmp_path / "power_validation_result.json").read_text())
+    assert reason in audit["reasons"]
+
+
+def test_native_device_replacement_cannot_preserve_validity(tmp_path):
+    root, bench, agg = _package(tmp_path)
+    (root / "node-1/gpu_metrics_devices_end.json").write_text('[{"gpu":0,"uuid":"replacement"}]')
+    assert run(root, bench, agg, expected_prefill_gpus=1, expected_decode_gpus=1, require_power=True) == 1
+    assert "native_device_identity_changed" in json.loads((tmp_path / "power_validation_result.json").read_text())["reasons"]
+
+
+def test_native_aggregate_nodes_do_not_invent_prefill_decode_metrics(tmp_path):
+    root, bench, agg = _package(tmp_path)
+    for path in root.glob("*/manifest.json"):
+        manifest = json.loads(path.read_text()); manifest["role"] = "aggregate"
+        path.write_text(json.dumps(manifest))
+    assert run(root, bench, agg, expected_prefill_gpus=0, expected_decode_gpus=0,
+               expected_aggregate_gpus=2, require_power=True) == 0
+    actual = json.loads(agg.read_text())
+    assert actual["avg_power_w"] == 200
+    assert "prefill_gpu_energy_j" not in actual
+
+
+def test_utc_context_replays_in_a_different_timezone(tmp_path):
+    csv = tmp_path / "gpu_metrics.csv"
+    csv.write_text("timestamp,index,power.draw [W]\n2026/01/01 00:00:00,0,100\n2026/01/01 00:00:02,0,100\n")
+    (tmp_path / "gpu_metrics_context.json").write_text('{"timestamp_timezone":"UTC"}')
+    script = "from pathlib import Path; from infx.results.power.single_node import integrate_power; " + \
+        f"r=integrate_power(Path({str(csv)!r}),start_unix=1767225600,end_unix=1767225602,expected_num_gpus=1); assert r.power_valid; assert r.total_gpu_energy_j == 200"
+    subprocess.run([sys.executable, "-c", script], cwd=REPO,
+                   env={**os.environ, "TZ": "America/Los_Angeles"}, check=True, timeout=10)
+
+
+@pytest.mark.parametrize("clock_value, synchronized", [
+    ("yes", True), ("true", True), ("no", False), ("false", False), ("", False),
+])
+def test_native_supervisor_reaps_monitor_and_writes_completion(tmp_path, clock_value, synchronized):
+    binary = tmp_path / "bin"; binary.mkdir()
+    (binary / "python3").symlink_to(sys.executable)
+    fake = binary / "nvidia-smi"
+    fake.write_text(f'''#!{sys.executable}
+import datetime, os, sys, time
+if any("index,uuid" in arg for arg in sys.argv):
+    print("index, uuid, pci.bus_id"); print("0, gpu-0, 0000:01:00.0")
+else:
+    with open({str(tmp_path / 'monitor.pid')!r}, "w") as stream: stream.write(str(os.getpid()))
+    if "-l" in sys.argv: print("timestamp,index,power.draw [W]", flush=True)
+    while True:
+        print(datetime.datetime.now(datetime.timezone.utc).strftime("%Y/%m/%d %H:%M:%S.%f") + ",0,100", flush=True)
+        if "-l" not in sys.argv: break
+        time.sleep(0.1)
+''')
+    fake.chmod(0o755)
+    control = tmp_path / "control"; control.mkdir()
+    node = tmp_path / "node-0"
+    process = subprocess.Popen(["bash", str(REPO / "benchmarks/native_power_collect.sh"),
+                                str(node), str(control), "nvidia", "0", "aggregate", "0", "1"],
+                               env={**os.environ, "PATH": f"{binary}:{os.environ['PATH']}",
+                                    "SLURM_JOB_ID": "test-job", "POWERX_COLLECTOR_REVISION": "revision",
+                                    "POWERX_CLOCK_SYNCHRONIZED": clock_value}, stdout=subprocess.PIPE,
+                               stderr=subprocess.PIPE, text=True)
+    try:
+        deadline = time.monotonic() + 10
+        while not (control / "ready-0").exists():
+            assert process.poll() is None
+            assert time.monotonic() < deadline
+            time.sleep(0.02)
+        (control / "stop").write_text("stop")
+        stdout, stderr = process.communicate(timeout=10)
+        assert process.returncode == 0, (stdout, stderr)
+        assert (control / "done-0").read_text().strip() == "0"
+        manifest = json.loads((node / "manifest.json").read_text())
+        assert manifest["lifecycle"] == "complete"
+        assert manifest["clock_synchronized"] is synchronized
+        assert (node / "gpu_metrics_identity_end.csv").exists()
+    finally:
+        if process.poll() is None:
+            process.terminate(); process.communicate(timeout=10)
+
+
+def test_result_processor_discovers_staged_native_package(tmp_path, monkeypatch):
+    from infx.results.fixed_sequence import aggregate_power_result
+
+    root, bench, agg = _package(tmp_path)
+    logs = tmp_path / "LOGS"
+    logs.mkdir()
+    root.rename(logs / "native_power")
+    monkeypatch.chdir(tmp_path)
+    env = {"IS_MULTINODE": "true", "PREFILL_GPUS": "1", "DECODE_GPUS": "1",
+           "RESULT_FILENAME": "result", "REQUIRE_POWER": "1"}
+    assert aggregate_power_result(env, bench, agg) == 0
+    assert json.loads(agg.read_text())["total_gpu_energy_j"] == 800
+    assert (tmp_path / "power_validation_result.json").is_file()
+
+    # Two competing formats must not silently select one package.
+    (logs / "power").mkdir()
+    assert aggregate_power_result(env, bench, agg) == 1
+    invalid = json.loads(agg.read_text())
+    assert invalid["power_valid"] == 0
+    assert "total_gpu_energy_j" not in invalid
+
+
+@pytest.mark.parametrize('role', ['prefill', 'decode'])
+def test_native_single_role_preserves_whole_fleet_and_role_metrics(tmp_path, role):
+    root, bench, agg = _package(tmp_path)
+    for path in root.glob('*/manifest.json'):
+        manifest = json.loads(path.read_text())
+        manifest['role'] = role
+        path.write_text(json.dumps(manifest))
+    assert run(root, bench, agg, expected_prefill_gpus=2 if role == 'prefill' else 0,
+               expected_decode_gpus=2 if role == 'decode' else 0, require_power=True) == 0
+    actual = json.loads(agg.read_text())
+    assert actual['total_gpu_energy_j'] == 800
+    assert actual[f'{role}_gpu_energy_j'] == 800
+    assert actual[f'{role}_avg_power_w'] == 200
+    opposite = 'decode' if role == 'prefill' else 'prefill'
+    assert f'{opposite}_gpu_energy_j' not in actual
+    assert json.loads((tmp_path / 'power_validation_result.json').read_text())['power_valid']
+
+
+def test_native_amd_abort_publishes_receipt_before_reaper_deadline(tmp_path):
+    binary = tmp_path / 'bin'
+    binary.mkdir()
+    (binary / 'python3').symlink_to(sys.executable)
+    fake = binary / 'amd-smi'
+    fake.write_text(f'''#!{sys.executable}
+import sys, time
+if "-w" in sys.argv:
+    print("timestamp,gpu,socket_power", flush=True)
+    while True:
+        print(str(int(time.time())) + ",0,100", flush=True)
+        time.sleep(0.1)
+else:
+    print("[]")
+''')
+    fake.chmod(0o755)
+    control = tmp_path / 'control'
+    control.mkdir()
+    node = tmp_path / 'node-0'
+    process = subprocess.Popen(['bash', '-c', '\n'.join([
+        'source "$1"',
+        'bash "$4" "$2" "$3" amd 0 aggregate 0 1 &',
+        'POWERX_COLLECTOR_PID=$! POWERX_CONTROL_DIR=$3 POWERX_NUM_NODES=1',
+        'POWERX_BARRIER_TIMEOUT_S=5',
+        'powerx_wait_collectors ready || exit 1',
+        'POWERX_BARRIER_TIMEOUT_S=0',
+        'powerx_reap_collector',
+    ]), 'bash', str(REPO / 'benchmarks/native_power_lifecycle.sh'), str(node), str(control), str(REPO / 'benchmarks/native_power_collect.sh')],
+        env={**os.environ, 'PATH': f'{binary}:/usr/bin:/bin', 'SLURM_JOB_ID': 'test-job'},
+        stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True)
+    try:
+        stdout, stderr = process.communicate(timeout=8)
+        assert process.returncode == 143, (stdout, stderr)
+        assert (control / 'done-0').read_text().strip() == '143'
+        assert json.loads((node / 'manifest.json').read_text())['lifecycle'] == 'failed'
+    finally:
+        import signal
+        try:
+            os.killpg(process.pid, signal.SIGKILL)
+        except ProcessLookupError:
+            pass
+        process.communicate()
+
+
+def test_native_prefill_only_rejects_aggregate_role_devices(tmp_path):
+    root, bench, agg = _package(tmp_path)
+    manifest_path = root / 'node-1/manifest.json'
+    manifest = json.loads(manifest_path.read_text())
+    manifest['role'] = 'aggregate'
+    manifest_path.write_text(json.dumps(manifest))
+    assert run(root, bench, agg, expected_prefill_gpus=2, expected_decode_gpus=0,
+               require_power=True) == 1
+    audit = json.loads((tmp_path / 'power_validation_result.json').read_text())
+    assert 'native_role_gpu_count_mismatch' in audit['reasons']
+    assert json.loads(agg.read_text())['power_valid'] == 0
+    assert 'prefill_avg_power_w' not in json.loads(agg.read_text())
+
+
+@pytest.mark.parametrize('tick', [2, 4])
+def test_native_audit_retains_boundary_noise_without_relaxing_in_window_errors(tmp_path, tick):
+    root, bench, agg = _package(tmp_path)
+    for rank in [0, 1]:
+        with (root / f'node-{rank}/gpu_metrics.csv').open('a') as stream:
+            stream.write(f'{tick},0,N/A\n')
+            if rank == 0:
+                stream.write(f'{tick},0,0\n')
+    assert run(root, bench, agg, expected_prefill_gpus=1, expected_decode_gpus=1,
+               require_power=True) == int(tick == 2)
+    audit = json.loads((tmp_path / 'power_validation_result.json').read_text())
+    assert audit['boundary_degenerate_rows'] == ({'uuid-0': 2, 'uuid-1': 1} if tick == 4 else {})
+    if tick == 4:
+        assert json.loads(agg.read_text())['total_gpu_energy_j'] == 800
diff --git a/utils/test_process_result.py b/utils/test_process_result.py
index 3048116d6..8b8661043 100644
--- a/utils/test_process_result.py
+++ b/utils/test_process_result.py
@@ -1642,6 +1642,22 @@ def test_public_power_audit_bounds_text_and_device_identifiers():
     assert audit['observed_gpu_ids'][:2] == ['gpu0', 'gpu1']
 
 
+@pytest.mark.parametrize('step_name', ['Upload GPU metrics', 'Upload power audit bundle'])
+def test_workflow_retains_context_for_each_metrics_csv(tmp_path, step_name):
+    import yaml
+
+    workflow = yaml.safe_load((REPO_ROOT / '.github/workflows/benchmark-tmpl.yml').read_text())
+    step = next(step for job in workflow['jobs'].values() for step in job.get('steps', [])
+                if step.get('name') == step_name)
+    patterns = step['with']['path'].splitlines()
+    for relative in ['gpu_metrics_concurrency_4_context.json',
+                     'results/gpu_metrics_concurrency_4_context.json']:
+        path = tmp_path / relative
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text('{"timestamp_timezone":"UTC"}')
+        assert any(path in tmp_path.glob(pattern.strip()) for pattern in patterns)
+
+
 @pytest.mark.parametrize('sidecar', ['run_recipe_conc4_gpus_4_ctx_2_gen_2.pytorch.json',
                                     'run_gpu_metrics_context.json', 'run_gpu_metrics_identity.json'])
 @pytest.mark.parametrize('point_state', ['valid', 'missing', 'malformed'])