diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/communication_baseline.py b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/communication_baseline.py
new file mode 100644
index 0000000..dcc3b6d
--- /dev/null
+++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/communication_baseline.py
@@ -0,0 +1,227 @@
+#!/usr/bin/env python3
+"""Small PCIe P2P and NCCL AllReduce baseline for the Phase 2 entry."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import os
+import statistics
+from typing import Any
+
+import torch
+import torch.distributed as dist
+
+
+RESULT_PREFIX = "COMM_RESULT "
+
+
+def parse_size(value: str) -> int:
+    text = value.strip().upper()
+    units = {"K": 1 << 10, "M": 1 << 20, "G": 1 << 30}
+    if text[-1:] in units:
+        return int(float(text[:-1]) * units[text[-1]])
+    return int(text)
+
+
+def percentile(values: list[float], quantile: float) -> float:
+    ordered = sorted(values)
+    if not ordered:
+        return math.nan
+    if len(ordered) == 1:
+        return ordered[0]
+    rank = (len(ordered) - 1) * quantile
+    lower = math.floor(rank)
+    upper = math.ceil(rank)
+    weight = rank - lower
+    return ordered[lower] * (1 - weight) + ordered[upper] * weight
+
+
+def emit(value: dict[str, Any]) -> None:
+    print(RESULT_PREFIX + json.dumps(value, sort_keys=True), flush=True)
+
+
+def run_p2p(args: argparse.Namespace) -> None:
+    device_count = torch.cuda.device_count()
+    size_bytes = parse_size(args.size)
+    elements = max(1, size_bytes // torch.tensor([], dtype=torch.float16).element_size())
+    for source in range(device_count):
+        for destination in range(device_count):
+            if source == destination:
+                continue
+            supported = torch.cuda.can_device_access_peer(source, destination)
+            if not supported:
+                emit(
+                    {
+                        "test": "p2p_copy",
+                        "node": args.node,
+                        "source_gpu": source,
+                        "destination_gpu": destination,
+                        "peer_access": False,
+                        "size_bytes": size_bytes,
+                    }
+                )
+                continue
+
+            with torch.cuda.device(source):
+                source_tensor = torch.ones(elements, dtype=torch.float16, device=source)
+            with torch.cuda.device(destination):
+                destination_tensor = torch.empty(
+                    elements,
+                    dtype=torch.float16,
+                    device=destination,
+                )
+                for _ in range(args.warmup):
+                    destination_tensor.copy_(source_tensor, non_blocking=True)
+                torch.cuda.synchronize(destination)
+
+                samples_ms: list[float] = []
+                for _ in range(args.iterations):
+                    start = torch.cuda.Event(enable_timing=True)
+                    end = torch.cuda.Event(enable_timing=True)
+                    start.record()
+                    destination_tensor.copy_(source_tensor, non_blocking=True)
+                    end.record()
+                    end.synchronize()
+                    samples_ms.append(start.elapsed_time(end))
+
+            mean_ms = statistics.fmean(samples_ms)
+            emit(
+                {
+                    "test": "p2p_copy",
+                    "node": args.node,
+                    "source_gpu": source,
+                    "destination_gpu": destination,
+                    "peer_access": True,
+                    "size_bytes": size_bytes,
+                    "iterations": args.iterations,
+                    "mean_ms": mean_ms,
+                    "p50_ms": percentile(samples_ms, 0.50),
+                    "p95_ms": percentile(samples_ms, 0.95),
+                    "bandwidth_GBps": size_bytes / (mean_ms / 1000.0) / 1e9,
+                }
+            )
+
+
+def run_all_reduce(args: argparse.Namespace) -> None:
+    dist.init_process_group("nccl")
+    rank = dist.get_rank()
+    world_size = dist.get_world_size()
+    local_rank = int(os.environ["LOCAL_RANK"])
+    torch.cuda.set_device(local_rank)
+
+    for size_text in args.sizes.split(","):
+        size_bytes = parse_size(size_text)
+        elements = max(
+            1,
+            size_bytes // torch.tensor([], dtype=torch.float32).element_size(),
+        )
+        tensor = torch.empty(elements, dtype=torch.float32, device=local_rank)
+        expected = world_size * (world_size + 1) / 2
+
+        for repetition in range(1, args.repetitions + 1):
+            for _ in range(args.warmup):
+                tensor.fill_(rank + 1)
+                dist.all_reduce(tensor)
+            torch.cuda.synchronize(local_rank)
+
+            local_samples_ms: list[float] = []
+            for _ in range(args.iterations):
+                tensor.fill_(rank + 1)
+                torch.cuda.synchronize(local_rank)
+                dist.barrier()
+                start = torch.cuda.Event(enable_timing=True)
+                end = torch.cuda.Event(enable_timing=True)
+                start.record()
+                dist.all_reduce(tensor)
+                end.record()
+                end.synchronize()
+                local_samples_ms.append(start.elapsed_time(end))
+
+            wrong_values = int(
+                not torch.allclose(
+                    tensor[0],
+                    torch.tensor(expected, device=local_rank),
+                )
+            )
+            gathered_samples: list[list[float] | None] = [None] * world_size
+            gathered_wrong: list[int | None] = [None] * world_size
+            dist.all_gather_object(gathered_samples, local_samples_ms)
+            dist.all_gather_object(gathered_wrong, wrong_values)
+            if rank == 0:
+                per_iteration_max = [
+                    max(
+                        float(samples[index])
+                        for samples in gathered_samples
+                        if samples is not None
+                    )
+                    for index in range(args.iterations)
+                ]
+                mean_ms = statistics.fmean(per_iteration_max)
+                algbw = size_bytes / (mean_ms / 1000.0) / 1e9
+                emit(
+                    {
+                        "test": "all_reduce",
+                        "scope": args.scope,
+                        "world_size": world_size,
+                        "size_bytes": size_bytes,
+                        "repetition": repetition,
+                        "iterations": args.iterations,
+                        "mean_ms": mean_ms,
+                        "p50_ms": percentile(per_iteration_max, 0.50),
+                        "p95_ms": percentile(per_iteration_max, 0.95),
+                        "algbw_GBps": algbw,
+                        "busbw_GBps": algbw * 2 * (world_size - 1) / world_size,
+                        "wrong_values": sum(
+                            int(value or 0) for value in gathered_wrong
+                        ),
+                        "nccl_cross_nic": os.environ.get(
+                            "NCCL_CROSS_NIC",
+                            "",
+                        ),
+                        "nccl_socket_ifname": os.environ.get(
+                            "NCCL_SOCKET_IFNAME",
+                            "",
+                        ),
+                        "nccl_ib_hca": os.environ.get("NCCL_IB_HCA", ""),
+                        "torch_version": torch.__version__,
+                        "cuda_version": torch.version.cuda,
+                        "nccl_version": ".".join(
+                            str(part) for part in torch.cuda.nccl.version()
+                        ),
+                    }
+                )
+
+    dist.destroy_process_group()
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser()
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    p2p = subparsers.add_parser("p2p")
+    p2p.add_argument("--node", required=True)
+    p2p.add_argument("--size", default="256M")
+    p2p.add_argument("--warmup", type=int, default=3)
+    p2p.add_argument("--iterations", type=int, default=10)
+
+    all_reduce = subparsers.add_parser("all-reduce")
+    all_reduce.add_argument("--scope", required=True)
+    all_reduce.add_argument("--sizes", default="1M,64M,1G")
+    all_reduce.add_argument("--repetitions", type=int, default=3)
+    all_reduce.add_argument("--warmup", type=int, default=5)
+    all_reduce.add_argument("--iterations", type=int, default=10)
+    return parser
+
+
+def main() -> None:
+    args = build_parser().parse_args()
+    if args.command == "p2p":
+        run_p2p(args)
+    else:
+        run_all_reduce(args)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/config.env b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/config.env
index 424559c..d9514a9 100644
--- a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/config.env
+++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/config.env
@@ -21,6 +21,10 @@ RUN_MIXED_CASE="${RUN_MIXED_CASE:-1}"
 
 # Monitoring policy.
 SAMPLE_INTERVAL_S="${SAMPLE_INTERVAL_S:-1}"
+CPU_SAMPLE_INTERVAL_S="${CPU_SAMPLE_INTERVAL_S:-5}"
+PROCESS_SAMPLE_INTERVAL_S="${PROCESS_SAMPLE_INTERVAL_S:-5}"
+NET_SAMPLE_INTERVAL_S="${NET_SAMPLE_INTERVAL_S:-5}"
+PERF_INTERVAL_MS="${PERF_INTERVAL_MS:-5000}"
 IDLE_BASELINE_S="${IDLE_BASELINE_S:-15}"
 POST_RUN_COOLDOWN_S="${POST_RUN_COOLDOWN_S:-15}"
 CASE_COOLDOWN_S="${CASE_COOLDOWN_S:-5}"
@@ -30,6 +34,23 @@ PERF_EVENTS="${PERF_EVENTS:-cycles,instructions,cache-misses,context-switches,cp
 RDMA_HCAS="${RDMA_HCAS:-mlx5_0 mlx5_3}"
 NUMASTAT_INTERVAL_S="${NUMASTAT_INTERVAL_S:-5}"
 CLOCK_SKEW_TOLERANCE_S="${CLOCK_SKEW_TOLERANCE_S:-2}"
+REQUIRE_PRECISE_WINDOWS="${REQUIRE_PRECISE_WINDOWS:-1}"
+
+# One-time communication baseline. It runs before the model service starts.
+RUN_COMMUNICATION_BASELINE="${RUN_COMMUNICATION_BASELINE:-1}"
+COMMUNICATION_IMAGE="${COMMUNICATION_IMAGE:-lmsysorg/sglang:nightly-dev-cu13-20260720-b3570a45}"
+COMMUNICATION_MASTER_PORT="${COMMUNICATION_MASTER_PORT:-29620}"
+COMMUNICATION_SIZES="${COMMUNICATION_SIZES:-1M,64M,1G}"
+COMMUNICATION_REPETITIONS="${COMMUNICATION_REPETITIONS:-3}"
+COMMUNICATION_WARMUP="${COMMUNICATION_WARMUP:-5}"
+COMMUNICATION_ITERATIONS="${COMMUNICATION_ITERATIONS:-10}"
+P2P_SIZE="${P2P_SIZE:-256M}"
+P2P_WARMUP="${P2P_WARMUP:-3}"
+P2P_ITERATIONS="${P2P_ITERATIONS:-10}"
+CROSS_NIC_VALUES="${CROSS_NIC_VALUES:-0 1 2}"
+NCCL_SOCKET_IFNAME="${NCCL_SOCKET_IFNAME:-=eth0}"
+NCCL_IB_HCA="${NCCL_IB_HCA:-=mlx5_0:1,mlx5_3:1}"
+RDMA_DEVICE_PATHS="${RDMA_DEVICE_PATHS:-/dev/infiniband/rdma_cm /dev/infiniband/uverbs0 /dev/infiniband/uverbs3}"
 
 SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 REPO_ROOT="$(cd "${SCRIPT_DIR}/../../.." && pwd)"
@@ -37,4 +58,4 @@ RESULT_BASE="${RESULT_BASE:-${SCRIPT_DIR}/results}"
 RUNTIME_BASE="${RUNTIME_BASE:-${SCRIPT_DIR}/runtime}"
 
 DRY_RUN="${DRY_RUN:-0}"
-ALLOW_PARTIAL_COLLECTORS="${ALLOW_PARTIAL_COLLECTORS:-1}"
+ALLOW_PARTIAL_COLLECTORS="${ALLOW_PARTIAL_COLLECTORS:-0}"
diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/hardware_contention_attribution.py b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/hardware_contention_attribution.py
index c2eb9c0..2b8148d 100755
--- a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/hardware_contention_attribution.py
+++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/hardware_contention_attribution.py
@@ -7,6 +7,7 @@ import argparse
 import csv
 import json
 import math
+import re
 import statistics
 import time
 from datetime import datetime
@@ -50,6 +51,27 @@ RDMA_FIELDS = [
     "rp_cnp_handled",
 ]
 
+DCGM_FIELDS = [
+    "gr_engine_active",
+    "sm_active",
+    "sm_occupancy",
+    "tensor_active",
+    "dram_active",
+    "pcie_tx_bytes_per_s",
+    "pcie_rx_bytes_per_s",
+]
+
+NUMA_FIELDS = [
+    "wall_time_ns",
+    "node",
+    "node0_mib",
+    "node1_mib",
+    "total_mib",
+    "processes",
+]
+
+COMMUNICATION_RESULT_PREFIX = "COMM_RESULT "
+
 
 def now_iso() -> str:
     return datetime.now().astimezone().isoformat(timespec="seconds")
@@ -98,6 +120,260 @@ def read_csv_rows(path: Path) -> list[dict[str, str]]:
         return list(csv.DictReader(handle))
 
 
+def read_timestamped_lines(path: Path) -> list[dict[str, str]]:
+    rows: list[dict[str, str]] = []
+    if not path.exists() or path.stat().st_size == 0:
+        return rows
+    for raw_line in path.read_text(
+        encoding="utf-8",
+        errors="replace",
+    ).splitlines():
+        parts = raw_line.split("\t", 2)
+        if len(parts) != 3 or parse_number(parts[0]) is None:
+            continue
+        rows.append(
+            {
+                "wall_time_ns": parts[0],
+                "node": parts[1],
+                "payload": parts[2],
+            }
+        )
+    return rows
+
+
+def numeric_stats(rows: list[dict[str, Any]], field: str) -> dict[str, Any]:
+    values = [
+        value
+        for value in (parse_number(row.get(field)) for row in rows)
+        if value is not None
+    ]
+    return {
+        f"{field}_mean": statistics.fmean(values) if values else None,
+        f"{field}_p95": percentile(values, 0.95),
+        f"{field}_max": max(values) if values else None,
+    }
+
+
+def summarize_numeric_rows(
+    rows: list[dict[str, Any]],
+    fields: Iterable[str],
+) -> dict[str, Any]:
+    result: dict[str, Any] = {"samples": len(rows)}
+    for field in fields:
+        result.update(numeric_stats(rows, field))
+    return result
+
+
+def parse_dcgm(path: Path) -> list[dict[str, Any]]:
+    result: list[dict[str, Any]] = []
+    if not path.exists():
+        return result
+    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
+        parts = line.split(",", 2)
+        if len(parts) != 3 or parse_number(parts[0]) is None:
+            continue
+        match = re.match(r"GPU\s+(\d+)\s+(.+)$", parts[2].strip())
+        if not match:
+            continue
+        values = match.group(2).split()
+        if len(values) < len(DCGM_FIELDS):
+            continue
+        row: dict[str, Any] = {
+            "wall_time_ns": parts[0],
+            "node": parts[1],
+            "gpu": match.group(1),
+        }
+        for field, value in zip(DCGM_FIELDS, values):
+            row[field] = parse_number(value)
+        if any(row[field] is not None for field in DCGM_FIELDS):
+            result.append(row)
+    return result
+
+
+def parse_mpstat(path: Path) -> list[dict[str, Any]]:
+    result: list[dict[str, Any]] = []
+    for record in read_timestamped_lines(path):
+        fields = record["payload"].split()
+        cpu_index = next(
+            (
+                index
+                for index, field in enumerate(fields)
+                if field == "all" or field.isdigit()
+            ),
+            None,
+        )
+        if cpu_index is None or len(fields) < cpu_index + 11:
+            continue
+        values = [parse_number(value) for value in fields[cpu_index + 1 : cpu_index + 11]]
+        if any(value is None for value in values):
+            continue
+        idle = float(values[-1])
+        result.append(
+            {
+                "wall_time_ns": record["wall_time_ns"],
+                "node": record["node"],
+                "cpu": fields[cpu_index],
+                "cpu_active_pct": 100.0 - idle,
+                "iowait_pct": values[3],
+            }
+        )
+    return result
+
+
+def parse_pidstat(path: Path) -> list[dict[str, Any]]:
+    result: list[dict[str, Any]] = []
+    mode = ""
+    for record in read_timestamped_lines(path):
+        payload = record["payload"]
+        if "%usr" in payload and "%CPU" in payload:
+            mode = "cpu"
+            continue
+        if "kB_rd/s" in payload:
+            mode = "disk"
+            continue
+        if "minflt/s" in payload:
+            mode = "memory"
+            continue
+        if "cswch/s" in payload:
+            mode = "context"
+            continue
+        fields = payload.split()
+        if not fields or mode == "":
+            continue
+        row: dict[str, Any] = {
+            "wall_time_ns": record["wall_time_ns"],
+            "node": record["node"],
+            "kind": mode,
+        }
+        try:
+            if mode == "cpu" and len(fields) >= 9:
+                row.update(
+                    {
+                        "pid": int(fields[-8]),
+                        "cpu_pct": parse_number(fields[-3]),
+                        "wait_pct": parse_number(fields[-4]),
+                    }
+                )
+            elif mode == "disk" and len(fields) >= 7:
+                row.update(
+                    {
+                        "pid": int(fields[-6]),
+                        "read_kib_s": parse_number(fields[-5]),
+                        "write_kib_s": parse_number(fields[-4]),
+                        "io_delay": parse_number(fields[-2]),
+                    }
+                )
+            elif mode == "memory" and len(fields) >= 8:
+                row.update(
+                    {
+                        "pid": int(fields[-7]),
+                        "major_faults_s": parse_number(fields[-5]),
+                        "rss_kib": parse_number(fields[-3]),
+                    }
+                )
+            elif mode == "context" and len(fields) >= 5:
+                row.update(
+                    {
+                        "pid": int(fields[-4]),
+                        "voluntary_switches_s": parse_number(fields[-3]),
+                        "involuntary_switches_s": parse_number(fields[-2]),
+                    }
+                )
+            else:
+                continue
+        except ValueError:
+            continue
+        result.append(row)
+    return result
+
+
+def parse_perf(path: Path) -> list[dict[str, Any]]:
+    result: list[dict[str, Any]] = []
+    pattern = re.compile(
+        r"^\s*[\d.]+\s+([\d,]+)\s+"
+        r"(cycles|instructions|cache-misses|context-switches|cpu-migrations|page-faults)\b"
+    )
+    for record in read_timestamped_lines(path):
+        match = pattern.match(record["payload"])
+        if not match:
+            continue
+        result.append(
+            {
+                "wall_time_ns": record["wall_time_ns"],
+                "node": record["node"],
+                "event": match.group(2),
+                "count": parse_number(match.group(1).replace(",", "")),
+            }
+        )
+    return result
+
+
+def parse_sar_net(path: Path) -> list[dict[str, Any]]:
+    result: list[dict[str, Any]] = []
+    mode = ""
+    for record in read_timestamped_lines(path):
+        payload = record["payload"]
+        if "IFACE" in payload and "rxpck/s" in payload:
+            mode = "traffic"
+            continue
+        if "IFACE" in payload and "rxerr/s" in payload:
+            mode = "errors"
+            continue
+        fields = payload.split()
+        iface_index = next(
+            (index for index, field in enumerate(fields) if field in {"eth0", "eth3"}),
+            None,
+        )
+        if iface_index is None:
+            continue
+        values = [parse_number(value) for value in fields[iface_index + 1 :]]
+        row: dict[str, Any] = {
+            "wall_time_ns": record["wall_time_ns"],
+            "node": record["node"],
+            "iface": fields[iface_index],
+            "kind": mode,
+        }
+        if mode == "traffic" and len(values) >= 8:
+            row.update(
+                {
+                    "rx_gbps": values[2] * 8 / 1e6 if values[2] is not None else None,
+                    "tx_gbps": values[3] * 8 / 1e6 if values[3] is not None else None,
+                    "ifutil_pct": values[7],
+                }
+            )
+        elif mode == "errors" and len(values) >= 9:
+            row.update(
+                {
+                    "rx_errors_s": values[0],
+                    "tx_errors_s": values[1],
+                    "rx_drops_s": values[3],
+                    "tx_drops_s": values[4],
+                }
+            )
+        else:
+            continue
+        result.append(row)
+    return result
+
+
+def load_communication_rows(result_dir: Path) -> list[dict[str, Any]]:
+    rows: list[dict[str, Any]] = []
+    for path in sorted((result_dir / "communication").glob("*.log")):
+        for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
+            marker_index = line.find(COMMUNICATION_RESULT_PREFIX)
+            if marker_index < 0:
+                continue
+            try:
+                value = json.loads(
+                    line[marker_index + len(COMMUNICATION_RESULT_PREFIX) :]
+                )
+            except json.JSONDecodeError:
+                continue
+            value["source_log"] = str(path)
+            rows.append(value)
+    return rows
+
+
 def summarize_gpu_rows(rows: list[dict[str, str]]) -> list[dict[str, Any]]:
     grouped: dict[tuple[str, str], list[dict[str, str]]] = {}
     for row in rows:
@@ -237,8 +513,16 @@ def load_case_windows(result_dir: Path) -> list[dict[str, Any]]:
             meta = json.loads(meta_path.read_text(encoding="utf-8"))
         except (OSError, json.JSONDecodeError):
             continue
-        started_ns = iso_to_ns(str(meta.get("started_at", "")))
-        ended_ns = iso_to_ns(str(meta.get("ended_at", "")))
+        window_source = str(meta.get("measurement_window_source", ""))
+        if window_source == "bench_main_marker_plus_duration":
+            started_at = str(meta.get("measurement_started_at", ""))
+            ended_at = str(meta.get("measurement_ended_at", ""))
+        else:
+            started_at = str(meta.get("started_at", ""))
+            ended_at = str(meta.get("ended_at", ""))
+            window_source = "case_process_fallback"
+        started_ns = iso_to_ns(started_at)
+        ended_ns = iso_to_ns(ended_at)
         if started_ns is None or ended_ns is None or ended_ns < started_ns:
             continue
         windows.append(
@@ -249,11 +533,12 @@ def load_case_windows(result_dir: Path) -> list[dict[str, Any]]:
                 "stage": str(meta.get("stage", "")),
                 "repetition": int(meta.get("repetition", 0) or 0),
                 "status": str(meta.get("status", "")),
-                "started_at": str(meta.get("started_at", "")),
-                "ended_at": str(meta.get("ended_at", "")),
+                "started_at": started_at,
+                "ended_at": ended_at,
                 "started_ns": started_ns,
                 "ended_ns": ended_ns,
                 "duration_s": (ended_ns - started_ns) / 1_000_000_000,
+                "window_source": window_source,
                 "meta_path": str(meta_path),
             }
         )
@@ -297,6 +582,7 @@ def summarize_case_hardware(
         "started_at",
         "ended_at",
         "duration_s",
+        "window_source",
     )
     for window in windows:
         identity = {field: window[field] for field in identity_fields}
@@ -319,6 +605,327 @@ def summarize_case_hardware(
     return case_gpu, case_rdma
 
 
+def window_identity(window: dict[str, Any]) -> dict[str, Any]:
+    return {
+        field: window[field]
+        for field in (
+            "phase2_bench_run",
+            "case_id",
+            "role",
+            "stage",
+            "repetition",
+            "status",
+            "started_at",
+            "ended_at",
+            "duration_s",
+            "window_source",
+        )
+    }
+
+
+def sums_by_second(
+    rows: list[dict[str, Any]],
+    field: str,
+) -> list[float]:
+    buckets: dict[int, float] = {}
+    for row in rows:
+        timestamp = parse_number(row.get("wall_time_ns"))
+        value = parse_number(row.get(field))
+        if timestamp is None or value is None:
+            continue
+        second = int(timestamp // 1_000_000_000)
+        buckets[second] = buckets.get(second, 0.0) + value
+    return list(buckets.values())
+
+
+def value_summary(values: list[float], prefix: str) -> dict[str, Any]:
+    return {
+        f"{prefix}_mean": statistics.fmean(values) if values else None,
+        f"{prefix}_p95": percentile(values, 0.95),
+        f"{prefix}_max": max(values) if values else None,
+    }
+
+
+def summarize_case_metrics(
+    result_dir: Path,
+    windows: list[dict[str, Any]],
+) -> dict[str, list[dict[str, Any]]]:
+    raw_by_node: dict[str, dict[str, list[dict[str, Any]]]] = {}
+    for node in ("head", "worker"):
+        node_dir = result_dir / node
+        raw_by_node[node] = {
+            "gpu": read_csv_rows(node_dir / "gpu_samples.csv"),
+            "dcgm": parse_dcgm(node_dir / "dcgm_dmon.log"),
+            "cpu": parse_mpstat(node_dir / "mpstat.log"),
+            "process": parse_pidstat(node_dir / "pidstat.log"),
+            "perf": parse_perf(node_dir / "perf_stat.log"),
+            "numa": read_csv_rows(node_dir / "numa_samples.csv"),
+            "netdev": parse_sar_net(node_dir / "sar_net.log"),
+        }
+
+    output: dict[str, list[dict[str, Any]]] = {
+        "gpu_node": [],
+        "dcgm": [],
+        "cpu": [],
+        "process": [],
+        "perf": [],
+        "numa": [],
+        "netdev": [],
+    }
+    gpu_fields = (
+        "gpu_util_pct",
+        "memory_util_pct",
+        "memory_used_mib",
+        "power_w",
+        "temperature_c",
+        "sm_clock_mhz",
+        "memory_clock_mhz",
+    )
+    for window in windows:
+        identity = window_identity(window)
+        for node in ("head", "worker"):
+            sources = raw_by_node[node]
+            sliced = {
+                name: rows_in_window(
+                    rows,
+                    window["started_ns"],
+                    window["ended_ns"],
+                )
+                for name, rows in sources.items()
+            }
+
+            gpu_rows = sliced["gpu"]
+            gpu_summary = summarize_numeric_rows(gpu_rows, gpu_fields)
+            gpu_summary["gpus"] = len(
+                {str(row.get("gpu", "")) for row in gpu_rows}
+            )
+            output["gpu_node"].append(
+                {**identity, "node": node, **gpu_summary}
+            )
+
+            output["dcgm"].append(
+                {
+                    **identity,
+                    "node": node,
+                    "gpus": len(
+                        {str(row.get("gpu", "")) for row in sliced["dcgm"]}
+                    ),
+                    **summarize_numeric_rows(sliced["dcgm"], DCGM_FIELDS),
+                }
+            )
+
+            cpu_all = [
+                row for row in sliced["cpu"] if str(row.get("cpu")) == "all"
+            ]
+            cpu_cores = [
+                row for row in sliced["cpu"] if str(row.get("cpu")) != "all"
+            ]
+            hot_by_second: dict[int, int] = {}
+            for row in cpu_cores:
+                timestamp = parse_number(row.get("wall_time_ns"))
+                active = parse_number(row.get("cpu_active_pct"))
+                if timestamp is None or active is None:
+                    continue
+                second = int(timestamp // 1_000_000_000)
+                if active >= 80:
+                    hot_by_second[second] = hot_by_second.get(second, 0) + 1
+                else:
+                    hot_by_second.setdefault(second, 0)
+            output["cpu"].append(
+                {
+                    **identity,
+                    "node": node,
+                    **summarize_numeric_rows(
+                        cpu_all,
+                        ("cpu_active_pct", "iowait_pct"),
+                    ),
+                    **value_summary(
+                        [float(value) for value in hot_by_second.values()],
+                        "hot_cores_ge80",
+                    ),
+                }
+            )
+
+            process_summary: dict[str, Any] = {
+                **identity,
+                "node": node,
+                "samples": len(sliced["process"]),
+            }
+            for field in (
+                "cpu_pct",
+                "wait_pct",
+                "read_kib_s",
+                "write_kib_s",
+                "major_faults_s",
+                "rss_kib",
+                "voluntary_switches_s",
+                "involuntary_switches_s",
+            ):
+                process_summary.update(
+                    value_summary(
+                        sums_by_second(sliced["process"], field),
+                        f"process_{field}",
+                    )
+                )
+            output["process"].append(process_summary)
+
+            perf_summary: dict[str, Any] = {
+                **identity,
+                "node": node,
+                "samples": len(sliced["perf"]),
+            }
+            event_values: dict[str, list[float]] = {}
+            for row in sliced["perf"]:
+                value = parse_number(row.get("count"))
+                if value is not None:
+                    event_values.setdefault(str(row.get("event")), []).append(value)
+            cycles = sum(event_values.get("cycles", []))
+            instructions = sum(event_values.get("instructions", []))
+            perf_summary["ipc"] = instructions / cycles if cycles else None
+            for event in (
+                "cache-misses",
+                "context-switches",
+                "cpu-migrations",
+                "page-faults",
+            ):
+                values = event_values.get(event, [])
+                perf_summary.update(value_summary(values, event.replace("-", "_")))
+            output["perf"].append(perf_summary)
+
+            numa_rows = sliced["numa"]
+            numa_summary = {
+                **identity,
+                "node": node,
+                **summarize_numeric_rows(
+                    numa_rows,
+                    ("node0_mib", "node1_mib", "total_mib", "processes"),
+                ),
+            }
+            imbalance = []
+            for row in numa_rows:
+                node0 = parse_number(row.get("node0_mib"))
+                node1 = parse_number(row.get("node1_mib"))
+                total = parse_number(row.get("total_mib"))
+                if node0 is not None and node1 is not None and total:
+                    imbalance.append(abs(node0 - node1) / total * 100.0)
+            numa_summary.update(value_summary(imbalance, "numa_imbalance_pct"))
+            output["numa"].append(numa_summary)
+
+            for iface in ("eth0", "eth3"):
+                iface_rows = [
+                    row
+                    for row in sliced["netdev"]
+                    if row.get("iface") == iface
+                ]
+                output["netdev"].append(
+                    {
+                        **identity,
+                        "node": node,
+                        "iface": iface,
+                        **summarize_numeric_rows(
+                            iface_rows,
+                            (
+                                "rx_gbps",
+                                "tx_gbps",
+                                "ifutil_pct",
+                                "rx_errors_s",
+                                "tx_errors_s",
+                                "rx_drops_s",
+                                "tx_drops_s",
+                            ),
+                        ),
+                    }
+                )
+    return output
+
+
+def aggregate_communication_rows(
+    rows: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    groups: dict[tuple[Any, ...], list[dict[str, Any]]] = {}
+    for row in rows:
+        if row.get("test") == "p2p_copy":
+            source = int(row.get("source_gpu", -1))
+            destination = int(row.get("destination_gpu", -1))
+            relation = (
+                "same_pcie_switch"
+                if source // 4 == destination // 4
+                else "cross_numa_sys"
+            )
+            key = ("p2p_copy", row.get("node"), relation, row.get("size_bytes"))
+        else:
+            key = (
+                "all_reduce",
+                row.get("scope"),
+                row.get("nccl_cross_nic", ""),
+                row.get("size_bytes"),
+            )
+        groups.setdefault(key, []).append(row)
+
+    result: list[dict[str, Any]] = []
+    for key, group in sorted(groups.items(), key=lambda item: str(item[0])):
+        if key[0] == "p2p_copy":
+            bandwidth = [
+                value
+                for value in (
+                    parse_number(row.get("bandwidth_GBps")) for row in group
+                )
+                if value is not None
+            ]
+            result.append(
+                {
+                    "test": key[0],
+                    "scope": key[1],
+                    "path_class": key[2],
+                    "size_bytes": key[3],
+                    "samples": len(group),
+                    "bandwidth_GBps_mean": (
+                        statistics.fmean(bandwidth) if bandwidth else None
+                    ),
+                    "bandwidth_GBps_p05": percentile(bandwidth, 0.05),
+                    "bandwidth_GBps_min": min(bandwidth) if bandwidth else None,
+                }
+            )
+        else:
+            mean_ms = [
+                value
+                for value in (parse_number(row.get("mean_ms")) for row in group)
+                if value is not None
+            ]
+            algbw = [
+                value
+                for value in (
+                    parse_number(row.get("algbw_GBps")) for row in group
+                )
+                if value is not None
+            ]
+            busbw = [
+                value
+                for value in (
+                    parse_number(row.get("busbw_GBps")) for row in group
+                )
+                if value is not None
+            ]
+            result.append(
+                {
+                    "test": key[0],
+                    "scope": key[1],
+                    "nccl_cross_nic": key[2],
+                    "size_bytes": key[3],
+                    "repetitions": len(group),
+                    "mean_ms": statistics.fmean(mean_ms) if mean_ms else None,
+                    "algbw_GBps_mean": statistics.fmean(algbw) if algbw else None,
+                    "busbw_GBps_mean": statistics.fmean(busbw) if busbw else None,
+                    "busbw_GBps_min": min(busbw) if busbw else None,
+                    "wrong_values": sum(
+                        int(parse_number(row.get("wrong_values")) or 0)
+                        for row in group
+                    ),
+                }
+            )
+    return result
+
+
 def write_csv(path: Path, rows: list[dict[str, Any]]) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     if not rows:
@@ -381,10 +988,18 @@ def create_manifest(args: argparse.Namespace) -> None:
             "fixed_case_ids": args.fixed_case_ids,
             "run_mixed_case": bool(args.run_mixed_case),
             "sample_interval_s": args.sample_interval_s,
+            "cpu_sample_interval_s": args.cpu_sample_interval_s,
+            "process_sample_interval_s": args.process_sample_interval_s,
+            "net_sample_interval_s": args.net_sample_interval_s,
+            "perf_interval_ms": args.perf_interval_ms,
             "numastat_interval_s": args.numastat_interval_s,
             "clock_skew_tolerance_s": args.clock_skew_tolerance_s,
             "idle_baseline_s": args.idle_baseline_s,
             "post_run_cooldown_s": args.post_run_cooldown_s,
+            "require_precise_windows": bool(args.require_precise_windows),
+            "run_communication_baseline": bool(
+                args.run_communication_baseline
+            ),
             "dry_run": bool(args.dry_run),
         },
     )
@@ -402,7 +1017,26 @@ def check_bench(summary_path: Path) -> bool:
     return bool(rows) and all(row.get("status") == "COMPLETED" for row in rows)
 
 
-def summarize(result_dir: Path) -> dict[str, Any]:
+def format_metric(value: Any, digits: int = 2) -> str:
+    parsed = parse_number(value)
+    return "-" if parsed is None else f"{parsed:.{digits}f}"
+
+
+def summarize_communication(result_dir: Path) -> list[dict[str, Any]]:
+    communication_rows = load_communication_rows(result_dir)
+    communication_aggregate = aggregate_communication_rows(communication_rows)
+    write_csv(result_dir / "communication_summary.csv", communication_rows)
+    write_csv(
+        result_dir / "communication_aggregate.csv",
+        communication_aggregate,
+    )
+    return communication_aggregate
+
+
+def summarize(
+    result_dir: Path,
+    require_precise_windows: bool = False,
+) -> dict[str, Any]:
     gpu_rows: list[dict[str, Any]] = []
     rdma_rows: list[dict[str, Any]] = []
     for node in ("head", "worker"):
@@ -414,6 +1048,18 @@ def summarize(result_dir: Path) -> dict[str, Any]:
         result_dir,
         case_windows,
     )
+    if require_precise_windows:
+        fallback = [
+            row
+            for row in case_windows
+            if row.get("window_source") != "bench_main_marker_plus_duration"
+        ]
+        if fallback:
+            raise ValueError(
+                f"{len(fallback)} benchmark windows are not precise main-run windows"
+            )
+    case_metrics = summarize_case_metrics(result_dir, case_windows)
+    communication_aggregate = summarize_communication(result_dir)
 
     write_csv(result_dir / "gpu_summary.csv", gpu_rows)
     write_csv(result_dir / "rdma_summary.csv", rdma_rows)
@@ -421,6 +1067,8 @@ def summarize(result_dir: Path) -> dict[str, Any]:
     write_csv(result_dir / "case_windows.csv", case_windows)
     write_csv(result_dir / "case_gpu_summary.csv", case_gpu_rows)
     write_csv(result_dir / "case_rdma_summary.csv", case_rdma_rows)
+    for name, rows in case_metrics.items():
+        write_csv(result_dir / f"case_{name}_summary.csv", rows)
 
     failed_bench = [row for row in bench_rows if row.get("status") != "COMPLETED"]
     collector_status_rows = read_csv_rows(result_dir / "collector_status.csv")
@@ -446,6 +1094,14 @@ def summarize(result_dir: Path) -> dict[str, Any]:
         "case_windows": len(case_windows),
         "case_gpu_summary_rows": len(case_gpu_rows),
         "case_rdma_summary_rows": len(case_rdma_rows),
+        "case_metric_rows": {
+            name: len(rows) for name, rows in case_metrics.items()
+        },
+        "communication_aggregate_rows": len(communication_aggregate),
+        "precise_windows": sum(
+            row.get("window_source") == "bench_main_marker_plus_duration"
+            for row in case_windows
+        ),
         "collector_status_counts": collector_status_counts,
         "collector_files": collector_files,
     }
@@ -460,9 +1116,10 @@ def summarize(result_dir: Path) -> dict[str, Any]:
         f"- GPU summary rows: `{len(gpu_rows)}`",
         f"- RDMA summary rows: `{len(rdma_rows)}`",
         f"- Case windows: `{len(case_windows)}`",
+        f"- Precise main-run windows: `{summary['precise_windows']}/{len(case_windows)}`",
         f"- Collector status counts: `{json.dumps(collector_status_counts, sort_keys=True)}`",
         "",
-        "## Bench",
+        "## 1. Benchmark results",
         "",
         "| Run | Case | Status | Input TPS | Output TPS | TTFT P95 (ms) | TPOT P95 (ms) |",
         "|---|---|---|---:|---:|---:|---:|",
@@ -482,21 +1139,236 @@ def summarize(result_dir: Path) -> dict[str, Any]:
                 }
             )
         )
+
     report.extend(
         [
             "",
-            "## Machine-readable summaries",
+            "## 2. Measurement-window validity",
+            "",
+            "| Case | Role | Duration (s) | Window source |",
+            "|---|---|---:|---|",
+        ]
+    )
+    for row in case_windows:
+        report.append(
+            f"| {row['case_id']} | {row['role'] or '-'} | "
+            f"{format_metric(row['duration_s'])} | {row['window_source']} |"
+        )
+
+    report.extend(
+        [
+            "",
+            "## 3. GPU basic state (`nvidia-smi`)",
+            "",
+            "Data: `case_gpu_node_summary.csv`; raw: `head|worker/gpu_samples.csv`.",
+            "",
+            "| Case | Node | Samples | GPU util mean/p95 (%) | Memory used mean (MiB) | "
+            "Power mean (W) | SM clock mean (MHz) |",
+            "|---|---|---:|---:|---:|---:|---:|",
+        ]
+    )
+    for row in case_metrics["gpu_node"]:
+        report.append(
+            f"| {row['case_id']} | {row['node']} | {row.get('samples', 0)} | "
+            f"{format_metric(row.get('gpu_util_pct_mean'))}/"
+            f"{format_metric(row.get('gpu_util_pct_p95'))} | "
+            f"{format_metric(row.get('memory_used_mib_mean'))} | "
+            f"{format_metric(row.get('power_w_mean'))} | "
+            f"{format_metric(row.get('sm_clock_mhz_mean'))} |"
+        )
+
+    report.extend(
+        [
+            "",
+            "## 4. GPU profiling counters (DCGM)",
+            "",
+            "Data: `case_dcgm_summary.csv`; raw: `head|worker/dcgm_dmon.log`.",
+            "",
+            "| Case | Node | Samples | GR active | SM active | SM occupancy | Tensor active | "
+            "DRAM active | PCIe TX/RX mean (GB/s) |",
+            "|---|---|---:|---:|---:|---:|---:|---:|---:|",
+        ]
+    )
+    for row in case_metrics["dcgm"]:
+        tx = parse_number(row.get("pcie_tx_bytes_per_s_mean"))
+        rx = parse_number(row.get("pcie_rx_bytes_per_s_mean"))
+        report.append(
+            f"| {row['case_id']} | {row['node']} | {row.get('samples', 0)} | "
+            f"{format_metric(row.get('gr_engine_active_mean'), 3)} | "
+            f"{format_metric(row.get('sm_active_mean'), 3)} | "
+            f"{format_metric(row.get('sm_occupancy_mean'), 3)} | "
+            f"{format_metric(row.get('tensor_active_mean'), 3)} | "
+            f"{format_metric(row.get('dram_active_mean'), 3)} | "
+            f"{format_metric(tx / 1e9 if tx is not None else None, 3)}/"
+            f"{format_metric(rx / 1e9 if rx is not None else None, 3)} |"
+        )
+
+    process_by_key = {
+        (row["case_id"], row["role"], row["node"]): row
+        for row in case_metrics["process"]
+    }
+    perf_by_key = {
+        (row["case_id"], row["role"], row["node"]): row
+        for row in case_metrics["perf"]
+    }
+    report.extend(
+        [
+            "",
+            "## 5. CPU, process, and `perf`",
+            "",
+            "Data: `case_cpu_summary.csv`, `case_process_summary.csv`, "
+            "`case_perf_summary.csv`; raw: `mpstat.log`, `pidstat.log`, `perf_stat.log`.",
+            "",
+            "| Case | Node | Samples CPU/process/perf | CPU active mean/p95 (%) | Hot cores max | "
+            "Process CPU max (%) | Process wait max (%) | IPC | Context switches mean/interval |",
+            "|---|---|---|---:|---:|---:|---:|---:|---:|",
+        ]
+    )
+    for row in case_metrics["cpu"]:
+        key = (row["case_id"], row["role"], row["node"])
+        process = process_by_key.get(key, {})
+        perf = perf_by_key.get(key, {})
+        report.append(
+            f"| {row['case_id']} | {row['node']} | "
+            f"{row.get('samples', 0)}/{process.get('samples', 0)}/"
+            f"{perf.get('samples', 0)} | "
+            f"{format_metric(row.get('cpu_active_pct_mean'))}/"
+            f"{format_metric(row.get('cpu_active_pct_p95'))} | "
+            f"{format_metric(row.get('hot_cores_ge80_max'), 0)} | "
+            f"{format_metric(process.get('process_cpu_pct_max'))} | "
+            f"{format_metric(process.get('process_wait_pct_max'))} | "
+            f"{format_metric(perf.get('ipc'), 3)} | "
+            f"{format_metric(perf.get('context_switches_mean'))} |"
+        )
+
+    report.extend(
+        [
+            "",
+            "## 6. NUMA memory placement",
+            "",
+            "Data/raw: `case_numa_summary.csv`, `head|worker/numa_samples.csv`.",
+            "",
+            "| Case | Node | Samples | Node 0 mean (MiB) | Node 1 mean (MiB) | "
+            "Total mean (MiB) | Imbalance mean/max (%) |",
+            "|---|---|---:|---:|---:|---:|---:|",
+        ]
+    )
+    for row in case_metrics["numa"]:
+        report.append(
+            f"| {row['case_id']} | {row['node']} | {row.get('samples', 0)} | "
+            f"{format_metric(row.get('node0_mib_mean'))} | "
+            f"{format_metric(row.get('node1_mib_mean'))} | "
+            f"{format_metric(row.get('total_mib_mean'))} | "
+            f"{format_metric(row.get('numa_imbalance_pct_mean'))}/"
+            f"{format_metric(row.get('numa_imbalance_pct_max'))} |"
+        )
+
+    report.extend(
+        [
+            "",
+            "## 7. Linux netdev and RDMA data path",
+            "",
+            "Netdev data: `case_netdev_summary.csv`; RDMA data: "
+            "`case_rdma_summary.csv`; raw: `sar_net.log`, `rdma.csv`.",
+            "",
+            "### Linux interfaces",
+            "",
+            "| Case | Node | Interface | Samples | RX mean/max (Gbit/s) | TX mean/max (Gbit/s) | "
+            "Util max (%) | RX/TX error max (/s) |",
+            "|---|---|---|---:|---:|---:|---:|---:|",
+        ]
+    )
+    for row in case_metrics["netdev"]:
+        report.append(
+            f"| {row['case_id']} | {row['node']} | {row['iface']} | "
+            f"{row.get('samples', 0)} | "
+            f"{format_metric(row.get('rx_gbps_mean'), 3)}/"
+            f"{format_metric(row.get('rx_gbps_max'), 3)} | "
+            f"{format_metric(row.get('tx_gbps_mean'), 3)}/"
+            f"{format_metric(row.get('tx_gbps_max'), 3)} | "
+            f"{format_metric(row.get('ifutil_pct_max'), 3)} | "
+            f"{format_metric(row.get('rx_errors_s_max'))}/"
+            f"{format_metric(row.get('tx_errors_s_max'))} |"
+        )
+    report.extend(
+        [
+            "",
+            "### RDMA HCAs",
+            "",
+            "| Case | Node | HCA | Samples | TX/RX (Gbit/s) | Wait delta | Discard/error delta | "
+            "Retry exceeded delta |",
+            "|---|---|---|---:|---:|---:|---:|---:|",
+        ]
+    )
+    for row in case_rdma_rows:
+        report.append(
+            f"| {row['case_id']} | {row['node']} | {row['hca']} | "
+            f"{row.get('samples', 0)} | "
+            f"{format_metric(row.get('xmit_gbps'))}/"
+            f"{format_metric(row.get('rcv_gbps'))} | "
+            f"{format_metric(row.get('port_xmit_wait_delta'), 0)} | "
+            f"{format_metric(row.get('port_xmit_discards_delta'), 0)}/"
+            f"{format_metric(row.get('port_rcv_errors_delta'), 0)} | "
+            f"{format_metric(row.get('req_transport_retries_exceeded_delta'), 0)} |"
+        )
+
+    report.extend(
+        [
+            "",
+            "## 8. PCIe P2P and NCCL communication baseline",
+            "",
+            "Data: `communication_aggregate.csv`; raw: `communication/*.log`.",
+            "",
+            "| Test | Scope | Path/CROSS_NIC | Samples/repetitions | Size (MiB) | Mean latency (ms) | "
+            "Bandwidth / busbw (GB/s) | Minimum | Wrong values |",
+            "|---|---|---|---:|---:|---:|---:|---:|---:|",
+        ]
+    )
+    for row in communication_aggregate:
+        if row.get("test") == "p2p_copy":
+            bandwidth = row.get("bandwidth_GBps_mean")
+            minimum = row.get("bandwidth_GBps_min")
+            path_or_cross = row.get("path_class")
+            wrong = "-"
+            sample_count = row.get("samples", 0)
+        else:
+            bandwidth = row.get("busbw_GBps_mean")
+            minimum = row.get("busbw_GBps_min")
+            path_or_cross = row.get("nccl_cross_nic") or "single-node"
+            wrong = str(row.get("wrong_values", ""))
+            sample_count = row.get("repetitions", 0)
+        size_mib = (parse_number(row.get("size_bytes")) or 0) / (1 << 20)
+        report.append(
+            f"| {row.get('test')} | {row.get('scope')} | {path_or_cross} | "
+            f"{sample_count} | "
+            f"{format_metric(size_mib, 0)} | {format_metric(row.get('mean_ms'))} | "
+            f"{format_metric(bandwidth)} | {format_metric(minimum)} | {wrong} |"
+        )
+
+    report.extend(
+        [
+            "",
+            "## 9. Machine-readable summaries",
             "",
             "- `gpu_summary.csv`",
             "- `rdma_summary.csv`",
             "- `bench_summary.csv`",
             "- `case_windows.csv`",
             "- `case_gpu_summary.csv`",
+            "- `case_gpu_node_summary.csv`",
+            "- `case_dcgm_summary.csv`",
+            "- `case_cpu_summary.csv`",
+            "- `case_process_summary.csv`",
+            "- `case_perf_summary.csv`",
+            "- `case_numa_summary.csv`",
+            "- `case_netdev_summary.csv`",
             "- `case_rdma_summary.csv`",
+            "- `communication_summary.csv`",
+            "- `communication_aggregate.csv`",
             "- `summary.json`",
             "",
-            "The final bottleneck conclusion is written only after aligning these files "
-            "with `markers.csv`, raw DCGM/sysstat logs, and SGLang service logs.",
+            "Each conclusion must cite the corresponding table above and its raw file; "
+            "missing samples are reported as `-`, never interpreted as zero.",
             "",
         ]
     )
@@ -527,10 +1399,28 @@ def build_parser() -> argparse.ArgumentParser:
     manifest_parser.add_argument("--fixed-case-ids", required=True)
     manifest_parser.add_argument("--run-mixed-case", type=int, required=True)
     manifest_parser.add_argument("--sample-interval-s", type=int, required=True)
+    manifest_parser.add_argument("--cpu-sample-interval-s", type=int, required=True)
+    manifest_parser.add_argument(
+        "--process-sample-interval-s",
+        type=int,
+        required=True,
+    )
+    manifest_parser.add_argument("--net-sample-interval-s", type=int, required=True)
+    manifest_parser.add_argument("--perf-interval-ms", type=int, required=True)
     manifest_parser.add_argument("--numastat-interval-s", type=int, required=True)
     manifest_parser.add_argument("--clock-skew-tolerance-s", type=int, required=True)
     manifest_parser.add_argument("--idle-baseline-s", type=int, required=True)
     manifest_parser.add_argument("--post-run-cooldown-s", type=int, required=True)
+    manifest_parser.add_argument(
+        "--require-precise-windows",
+        type=int,
+        required=True,
+    )
+    manifest_parser.add_argument(
+        "--run-communication-baseline",
+        type=int,
+        required=True,
+    )
     manifest_parser.add_argument("--dry-run", type=int, required=True)
 
     finish_parser = subparsers.add_parser("finish")
@@ -542,6 +1432,16 @@ def build_parser() -> argparse.ArgumentParser:
 
     summarize_parser = subparsers.add_parser("summarize")
     summarize_parser.add_argument("result_dir")
+    summarize_parser.add_argument(
+        "--require-precise-windows",
+        type=int,
+        default=0,
+    )
+
+    summarize_communication_parser = subparsers.add_parser(
+        "summarize-communication"
+    )
+    summarize_communication_parser.add_argument("result_dir")
     return parser
 
 
@@ -565,7 +1465,13 @@ def main() -> int:
     if args.command == "check-bench":
         return 0 if check_bench(Path(args.summary)) else 1
     if args.command == "summarize":
-        summarize(Path(args.result_dir))
+        summarize(
+            Path(args.result_dir),
+            bool(args.require_precise_windows),
+        )
+        return 0
+    if args.command == "summarize-communication":
+        summarize_communication(Path(args.result_dir))
         return 0
     raise AssertionError(args.command)
 
diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/run_hardware_contention_attribution.sh b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/run_hardware_contention_attribution.sh
index 96b78a2..034225e 100755
--- a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/run_hardware_contention_attribution.sh
+++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/run_hardware_contention_attribution.sh
@@ -11,6 +11,7 @@ ACTION="${1:-all}"
 RUN_ID="${RUN_ID:-dsv4pro-phase2-$(date +%Y%m%d-%H%M%S)}"
 RESULT_DIR="${RESULT_BASE}/${RUN_ID}"
 RESULT_TOOL="${SCRIPT_DIR}/hardware_contention_attribution.py"
+COMMUNICATION_TOOL="${SCRIPT_DIR}/communication_baseline.py"
 MARKERS_PATH="${RESULT_DIR}/markers.csv"
 SERVICE_DIR="${RESULT_DIR}/service"
 COMMAND_DIR="${RESULT_DIR}/commands"
@@ -72,6 +73,10 @@ validate_config() {
     log "ERROR: result tool is missing or not executable: ${RESULT_TOOL}"
     return 1
   }
+  [[ -f "${COMMUNICATION_TOOL}" ]] || {
+    log "ERROR: communication tool is missing: ${COMMUNICATION_TOOL}"
+    return 1
+  }
   case "${RUN_MIXED_CASE}" in
     0|1) ;;
     *)
@@ -86,6 +91,20 @@ validate_config() {
       return 1
       ;;
   esac
+  case "${REQUIRE_PRECISE_WINDOWS}" in
+    0|1) ;;
+    *)
+      log "ERROR: REQUIRE_PRECISE_WINDOWS must be 0 or 1"
+      return 1
+      ;;
+  esac
+  case "${RUN_COMMUNICATION_BASELINE}" in
+    0|1) ;;
+    *)
+      log "ERROR: RUN_COMMUNICATION_BASELINE must be 0 or 1"
+      return 1
+      ;;
+  esac
   [[ "${SAMPLE_INTERVAL_S}" =~ ^[1-9][0-9]*$ ]] || {
     log "ERROR: SAMPLE_INTERVAL_S must be a positive integer"
     return 1
@@ -94,6 +113,14 @@ validate_config() {
     log "ERROR: NUMASTAT_INTERVAL_S must be a positive integer"
     return 1
   }
+  local interval_name
+  for interval_name in CPU_SAMPLE_INTERVAL_S PROCESS_SAMPLE_INTERVAL_S \
+    NET_SAMPLE_INTERVAL_S PERF_INTERVAL_MS; do
+    [[ "${!interval_name}" =~ ^[1-9][0-9]*$ ]] || {
+      log "ERROR: ${interval_name} must be a positive integer"
+      return 1
+    }
+  done
   [[ "${CLOCK_SKEW_TOLERANCE_S}" =~ ^[0-9]+$ ]] || {
     log "ERROR: CLOCK_SKEW_TOLERANCE_S must be a non-negative integer"
     return 1
@@ -121,6 +148,13 @@ preflight_node_tools() {
       log "ERROR: ${node} is missing required tool: dcgmi"
       return 1
     fi
+  elif ! run_on_node "${node}" "dcgmi discovery -l >/dev/null 2>&1"; then
+    if [[ "${ALLOW_PARTIAL_COLLECTORS}" == "1" ]]; then
+      log "WARN: ${node} DCGM Host Engine is unavailable"
+    else
+      log "ERROR: ${node} DCGM Host Engine is unavailable; run systemctl start nvidia-dcgm"
+      return 1
+    fi
   fi
   local hca
   for hca in ${RDMA_HCAS}; do
@@ -149,6 +183,20 @@ preflight_clock_sync() {
   log "node clock skew check passed: delta=${delta}s"
 }
 
+preflight_gpus_idle() {
+  local node="$1"
+  local active
+  active="$(
+    run_on_node "${node}" \
+      "nvidia-smi --query-compute-apps=pid --format=csv,noheader,nounits 2>/dev/null | sed '/^[[:space:]]*$/d'" \
+      || true
+  )"
+  if [[ -n "${active}" ]]; then
+    log "ERROR: ${node} has active GPU compute processes: ${active//$'\n'/,}"
+    return 1
+  fi
+}
+
 write_manifest() {
   local git_commit git_dirty
   git_commit="$(git -C "${REPO_ROOT}" rev-parse HEAD 2>/dev/null || printf unknown)"
@@ -167,10 +215,16 @@ write_manifest() {
     --fixed-case-ids "${FIXED_CASE_IDS}" \
     --run-mixed-case "${RUN_MIXED_CASE}" \
     --sample-interval-s "${SAMPLE_INTERVAL_S}" \
+    --cpu-sample-interval-s "${CPU_SAMPLE_INTERVAL_S}" \
+    --process-sample-interval-s "${PROCESS_SAMPLE_INTERVAL_S}" \
+    --net-sample-interval-s "${NET_SAMPLE_INTERVAL_S}" \
+    --perf-interval-ms "${PERF_INTERVAL_MS}" \
     --numastat-interval-s "${NUMASTAT_INTERVAL_S}" \
     --clock-skew-tolerance-s "${CLOCK_SKEW_TOLERANCE_S}" \
     --idle-baseline-s "${IDLE_BASELINE_S}" \
     --post-run-cooldown-s "${POST_RUN_COOLDOWN_S}" \
+    --require-precise-windows "${REQUIRE_PRECISE_WINDOWS}" \
+    --run-communication-baseline "${RUN_COMMUNICATION_BASELINE}" \
     --dry-run "${DRY_RUN}"
 }
 
@@ -212,6 +266,185 @@ run_phase1_action() {
   "${command[@]}"
 }
 
+build_communication_docker_command() {
+  local output_name="$1"
+  local container_name="$2"
+  local entrypoint="$3"
+  local cross_nic="$4"
+  shift 4
+  local -a docker_cmd=(
+    docker run --rm
+    --name "${container_name}"
+    --gpus all
+    --network host
+    --ipc host
+    --shm-size 20g
+    --ulimit memlock=-1
+    --ulimit stack=67108864
+    -v "${REPO_ROOT}:${REPO_ROOT}:ro"
+    -e "NCCL_SOCKET_IFNAME=${NCCL_SOCKET_IFNAME}"
+    -e "NCCL_IB_HCA=${NCCL_IB_HCA}"
+    -e "NCCL_CROSS_NIC=${cross_nic}"
+    -e NCCL_DEBUG=INFO
+    -e NCCL_DEBUG_SUBSYS=INIT,NET,GRAPH,TUNING
+    -e TORCH_NCCL_ASYNC_ERROR_HANDLING=1
+  )
+  local device
+  for device in ${RDMA_DEVICE_PATHS}; do
+    docker_cmd+=(--device "${device}")
+  done
+  docker_cmd+=(--entrypoint "${entrypoint}" "${COMMUNICATION_IMAGE}" "$@")
+  printf -v "${output_name}" '%q ' "${docker_cmd[@]}"
+}
+
+run_p2p_baseline() {
+  local role="$1"
+  local node="$2"
+  local container="${EXPERIMENT}_comm_p2p_${role}_${RUN_ID}"
+  local command
+  build_communication_docker_command \
+    command "${container}" python3 2 \
+    "${COMMUNICATION_TOOL}" p2p \
+    --node "${role}" \
+    --size "${P2P_SIZE}" \
+    --warmup "${P2P_WARMUP}" \
+    --iterations "${P2P_ITERATIONS}"
+  printf '%s\n' "${command}" \
+    > "${COMMAND_DIR}/communication_p2p_${role}.cmd.txt"
+  if [[ "${DRY_RUN}" == "1" ]]; then
+    log "[DRY] P2P baseline role=${role}: ${command}"
+    return 0
+  fi
+  log "START P2P baseline role=${role}"
+  run_on_node "${node}" "${command}" \
+    > "${RESULT_DIR}/communication/p2p_${role}.log" 2>&1
+  log "DONE  P2P baseline role=${role}"
+}
+
+run_single_node_allreduce() {
+  local role="$1"
+  local node="$2"
+  local master_port="$3"
+  local container="${EXPERIMENT}_comm_ar8_${role}_${RUN_ID}"
+  local command
+  build_communication_docker_command \
+    command "${container}" torchrun 2 \
+    --standalone \
+    --nnodes=1 \
+    --nproc-per-node=8 \
+    --master-port "${master_port}" \
+    "${COMMUNICATION_TOOL}" all-reduce \
+    --scope "${role}_8gpu" \
+    --sizes "${COMMUNICATION_SIZES}" \
+    --repetitions "${COMMUNICATION_REPETITIONS}" \
+    --warmup "${COMMUNICATION_WARMUP}" \
+    --iterations "${COMMUNICATION_ITERATIONS}"
+  printf '%s\n' "${command}" \
+    > "${COMMAND_DIR}/communication_allreduce_${role}_8gpu.cmd.txt"
+  if [[ "${DRY_RUN}" == "1" ]]; then
+    log "[DRY] AllReduce baseline scope=${role}_8gpu: ${command}"
+    return 0
+  fi
+  log "START AllReduce baseline scope=${role}_8gpu"
+  run_on_node "${node}" "${command}" \
+    > "${RESULT_DIR}/communication/allreduce_${role}_8gpu.log" 2>&1
+  log "DONE  AllReduce baseline scope=${role}_8gpu"
+}
+
+run_two_node_allreduce() {
+  local cross_nic="$1"
+  local master_port="$(( COMMUNICATION_MASTER_PORT + cross_nic + 10 ))"
+  local head_container="${EXPERIMENT}_comm_ar16_head_x${cross_nic}_${RUN_ID}"
+  local worker_container="${EXPERIMENT}_comm_ar16_worker_x${cross_nic}_${RUN_ID}"
+  local head_command worker_command
+  local -a common_args=(
+    --nnodes=2
+    --nproc-per-node=8
+    --master-addr "${HEAD_NODE}"
+    --master-port "${master_port}"
+  )
+  build_communication_docker_command \
+    worker_command "${worker_container}" torchrun "${cross_nic}" \
+    "${common_args[@]}" \
+    --node-rank=1 \
+    "${COMMUNICATION_TOOL}" all-reduce \
+    --scope two_node_16gpu \
+    --sizes "${COMMUNICATION_SIZES}" \
+    --repetitions "${COMMUNICATION_REPETITIONS}" \
+    --warmup "${COMMUNICATION_WARMUP}" \
+    --iterations "${COMMUNICATION_ITERATIONS}"
+  build_communication_docker_command \
+    head_command "${head_container}" torchrun "${cross_nic}" \
+    "${common_args[@]}" \
+    --node-rank=0 \
+    "${COMMUNICATION_TOOL}" all-reduce \
+    --scope two_node_16gpu \
+    --sizes "${COMMUNICATION_SIZES}" \
+    --repetitions "${COMMUNICATION_REPETITIONS}" \
+    --warmup "${COMMUNICATION_WARMUP}" \
+    --iterations "${COMMUNICATION_ITERATIONS}"
+  printf '%s\n' "${head_command}" \
+    > "${COMMAND_DIR}/communication_allreduce_16gpu_crossnic${cross_nic}_head.cmd.txt"
+  printf '%s\n' "${worker_command}" \
+    > "${COMMAND_DIR}/communication_allreduce_16gpu_crossnic${cross_nic}_worker.cmd.txt"
+  if [[ "${DRY_RUN}" == "1" ]]; then
+    log "[DRY] AllReduce 16 GPU cross_nic=${cross_nic} Head: ${head_command}"
+    log "[DRY] AllReduce 16 GPU cross_nic=${cross_nic} Worker: ${worker_command}"
+    return 0
+  fi
+
+  log "START AllReduce baseline scope=two_node_16gpu cross_nic=${cross_nic}"
+  set +e
+  run_on_node "${WORKER_NODE}" "${worker_command}" \
+    > "${RESULT_DIR}/communication/allreduce_16gpu_crossnic${cross_nic}_worker.log" \
+    2>&1 &
+  local worker_pid=$!
+  sleep 3
+  run_on_node "${HEAD_NODE}" "${head_command}" \
+    > "${RESULT_DIR}/communication/allreduce_16gpu_crossnic${cross_nic}_head.log" \
+    2>&1
+  local head_rc=$?
+  wait "${worker_pid}"
+  local worker_rc=$?
+  set -e
+  if (( head_rc != 0 || worker_rc != 0 )); then
+    log "ERROR: AllReduce cross_nic=${cross_nic} head_rc=${head_rc} worker_rc=${worker_rc}"
+    return 1
+  fi
+  log "DONE  AllReduce baseline scope=two_node_16gpu cross_nic=${cross_nic}"
+}
+
+cleanup_communication_containers() {
+  [[ "${DRY_RUN}" == "1" ]] && return 0
+  local node
+  for node in "${HEAD_NODE}" "${WORKER_NODE}"; do
+    run_on_node "${node}" \
+      "docker ps -aq --filter 'name=^/${EXPERIMENT}_comm_' | xargs -r docker rm -f >/dev/null 2>&1" \
+      || true
+  done
+}
+
+run_communication_baseline() {
+  [[ "${RUN_COMMUNICATION_BASELINE}" == "1" ]] || {
+    log "SKIP communication baseline by configuration"
+    return 0
+  }
+  mkdir -p "${RESULT_DIR}/communication" "${COMMAND_DIR}"
+  if [[ "${DRY_RUN}" != "1" ]]; then
+    preflight_gpus_idle "${HEAD_NODE}"
+    preflight_gpus_idle "${WORKER_NODE}"
+  fi
+  run_p2p_baseline head "${HEAD_NODE}"
+  run_p2p_baseline worker "${WORKER_NODE}"
+  run_single_node_allreduce head "${HEAD_NODE}" "${COMMUNICATION_MASTER_PORT}"
+  run_single_node_allreduce worker "${WORKER_NODE}" "$(( COMMUNICATION_MASTER_PORT + 1 ))"
+  local cross_nic
+  for cross_nic in ${CROSS_NIC_VALUES}; do
+    run_two_node_allreduce "${cross_nic}"
+  done
+  cleanup_communication_containers
+}
+
 start_service() {
   mark_event service_start
   # Set this before launch so the EXIT trap also cleans a partially started pair.
@@ -364,7 +597,7 @@ trap 'exit 0' HUP TERM PIPE
 while :; do
   printf 'wall_time_ns=%s\\n' \"\$(date +%s%N)\"
   docker top '${container}' -eo pid,ppid,psr,pcpu,pmem,stat,comm,args
-  sleep '${SAMPLE_INTERVAL_S}'
+  sleep '${PROCESS_SAMPLE_INTERVAL_S}'
 done
 "
 }
@@ -383,20 +616,27 @@ fi
 }
 
 numastat_command() {
-  local container="$1"
+  local role="$1"
+  local container="$2"
   printf '%s' "
 trap 'exit 0' HUP TERM PIPE
+printf '%s\n' 'wall_time_ns,node,node0_mib,node1_mib,total_mib,processes'
 while :; do
-  printf 'wall_time_ns=%s\\n' \"\$(date +%s%N)\"
   PIDS=\$(docker top '${container}' -eo pid 2>/dev/null |
     awk 'NR > 1 {print \$1}')
   if [[ -z \"\${PIDS}\" ]]; then
     printf 'container has no visible processes: %s\\n' '${container}' >&2
     exit 1
   fi
+  values=\$(
   for pid in \${PIDS}; do
-    numastat -p \"\${pid}\"
-  done
+    numastat -p \"\${pid}\" 2>/dev/null |
+      awk '\$1 == \"Total\" {print \$2, \$3, \$4}'
+  done |
+    awk '{n0 += \$1; n1 += \$2; total += \$3; count += 1}
+         END {printf \"%.2f,%.2f,%.2f,%d\", n0, n1, total, count}'
+  )
+  printf '%s,%s,%s\\n' \"\$(date +%s%N)\" '${role}' \"\${values}\"
   sleep '${NUMASTAT_INTERVAL_S}'
 done
 "
@@ -416,23 +656,36 @@ dcgmi dmon -e '${DCGM_FIELD_IDS}' -d '$(( SAMPLE_INTERVAL_S * 1000 ))' |
 "
   local mpstat_command="
 trap 'exit 0' HUP TERM PIPE
-mpstat -P ALL '${SAMPLE_INTERVAL_S}'
+LC_ALL=C stdbuf -oL -eL mpstat -P ALL '${CPU_SAMPLE_INTERVAL_S}' |
+  while IFS= read -r line; do
+    printf '%s\\t%s\\t%s\\n' \"\$(date +%s%N)\" '${role}' \"\${line}\"
+  done
 "
   local pidstat_command
   pidstat_command="$(container_pid_preamble "${container}")"
   pidstat_command+="
 trap 'exit 0' HUP TERM PIPE
-pidstat -durwt -p \"\${PIDS}\" '${SAMPLE_INTERVAL_S}'
+LC_ALL=C stdbuf -oL -eL pidstat -durw -p \"\${PIDS}\" '${PROCESS_SAMPLE_INTERVAL_S}' |
+  while IFS= read -r line; do
+    printf '%s\\t%s\\t%s\\n' \"\$(date +%s%N)\" '${role}' \"\${line}\"
+  done
 "
   local sar_command="
 trap 'exit 0' HUP TERM PIPE
-sar -n DEV,EDEV '${SAMPLE_INTERVAL_S}'
+LC_ALL=C stdbuf -oL -eL sar -n DEV,EDEV '${NET_SAMPLE_INTERVAL_S}' |
+  while IFS= read -r line; do
+    printf '%s\\t%s\\t%s\\n' \"\$(date +%s%N)\" '${role}' \"\${line}\"
+  done
 "
   local perf_command
   perf_command="$(container_pid_preamble "${container}")"
   perf_command+="
 trap 'exit 0' HUP TERM PIPE
-perf stat -p \"\${PIDS}\" -I 1000 -e '${PERF_EVENTS}'
+LC_ALL=C stdbuf -oL -eL perf stat -p \"\${PIDS}\" \
+  -I '${PERF_INTERVAL_MS}' -e '${PERF_EVENTS}' 2>&1 |
+  while IFS= read -r line; do
+    printf '%s\\t%s\\t%s\\n' \"\$(date +%s%N)\" '${role}' \"\${line}\"
+  done
 "
 
   start_stream_collector \
@@ -452,7 +705,7 @@ perf stat -p \"\${PIDS}\" -I 1000 -e '${PERF_EVENTS}'
   start_stream_collector \
     "${role}" "${node}" docker_top.log "$(docker_top_command "${container}")"
   start_stream_collector \
-    "${role}" "${node}" numastat.log "$(numastat_command "${container}")"
+    "${role}" "${node}" numa_samples.csv "$(numastat_command "${role}" "${container}")"
 }
 
 start_collectors() {
@@ -581,7 +834,8 @@ run_mixed_case() {
 }
 
 summarize_results() {
-  python3 "${RESULT_TOOL}" summarize "${RESULT_DIR}"
+  python3 "${RESULT_TOOL}" summarize "${RESULT_DIR}" \
+    --require-precise-windows "${REQUIRE_PRECISE_WINDOWS}"
 }
 
 finish_manifest() {
@@ -595,6 +849,7 @@ cleanup() {
   local rc=$?
   stop_collectors || true
   stop_service || true
+  cleanup_communication_containers || true
   if (( rc != 0 )) && [[ -f "${RESULT_DIR}/manifest.json" ]]; then
     finish_manifest ABORTED || true
   fi
@@ -620,6 +875,7 @@ run_all() {
   fi
 
   trap cleanup EXIT INT TERM
+  run_communication_baseline
   start_service
   capture_static_snapshots before
   start_collectors
@@ -652,6 +908,10 @@ run_all() {
   mark_event cooldown_start
   sleep_if_real "${POST_RUN_COOLDOWN_S}"
   mark_event cooldown_end
+  if ! check_collectors; then
+    log "ERROR: one or more required collectors exited during the run"
+    ((failures+=1))
+  fi
   stop_collectors
   capture_static_snapshots after
   stop_service
@@ -666,6 +926,22 @@ run_all() {
   (( failures == 0 ))
 }
 
+run_communication_only() {
+  validate_config
+  enable_result_logging
+  mkdir -p "${RESULT_DIR}" "${COMMAND_DIR}" "${RESULT_DIR}/communication"
+  if [[ "${DRY_RUN}" != "1" ]]; then
+    preflight_node_tools "${HEAD_NODE}"
+    preflight_node_tools "${WORKER_NODE}"
+    preflight_clock_sync
+  fi
+  trap cleanup_communication_containers EXIT INT TERM
+  run_communication_baseline
+  python3 "${RESULT_TOOL}" summarize-communication "${RESULT_DIR}"
+  trap - EXIT INT TERM
+  log "Communication baseline complete: result=${RESULT_DIR}"
+}
+
 main() {
   case "${ACTION}" in
     all)
@@ -675,12 +951,15 @@ main() {
       enable_result_logging
       summarize_results
       ;;
+    communication)
+      run_communication_only
+      ;;
     stop)
       enable_result_logging
       stop_service
       ;;
     *)
-      printf 'Usage: %s {all|summarize|stop}\n' "$0" >&2
+      printf 'Usage: %s {all|communication|summarize|stop}\n' "$0" >&2
       return 2
       ;;
   esac
diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/tests/test_hardware_contention_attribution.py b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/tests/test_hardware_contention_attribution.py
index c3a2a56..66df919 100644
--- a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/tests/test_hardware_contention_attribution.py
+++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/tests/test_hardware_contention_attribution.py
@@ -113,6 +113,10 @@ class HardwareContentionAttributionTest(unittest.TestCase):
             rows = attribution.read_csv_rows(result_dir / "bench_summary.csv")
             self.assertEqual(rows[0]["phase2_bench_run"], "prefill")
             self.assertTrue((result_dir / "report.md").exists())
+            report = (result_dir / "report.md").read_text(encoding="utf-8")
+            self.assertIn("Samples CPU/process/perf", report)
+            self.assertIn("Samples/repetitions", report)
+            self.assertIn("missing samples are reported as `-`", report)
 
     def test_case_hardware_is_cut_by_phase1_meta_window(self) -> None:
         with tempfile.TemporaryDirectory() as temporary:
@@ -179,6 +183,113 @@ class HardwareContentionAttributionTest(unittest.TestCase):
                 "long_prefill_latency_128k_c1",
             )
 
+    def test_case_window_prefers_precise_main_benchmark_fields(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            result_dir = Path(temporary)
+            meta_path = (
+                result_dir
+                / "bench"
+                / "decode"
+                / "cases"
+                / "decode_case"
+                / "rep1"
+                / "meta.json"
+            )
+            meta_path.parent.mkdir(parents=True)
+            meta_path.write_text(
+                json.dumps(
+                    {
+                        "case_id": "decode_case",
+                        "role": "",
+                        "stage": "decode",
+                        "repetition": 1,
+                        "status": "COMPLETED",
+                        "started_at": "2026-07-31T00:00:00+00:00",
+                        "ended_at": "2026-07-31T00:01:00+00:00",
+                        "measurement_started_at": "2026-07-31T00:00:10+00:00",
+                        "measurement_ended_at": "2026-07-31T00:00:30+00:00",
+                        "measurement_window_source": "bench_main_marker_plus_duration",
+                    }
+                ),
+                encoding="utf-8",
+            )
+
+            windows = attribution.load_case_windows(result_dir)
+
+            self.assertEqual(len(windows), 1)
+            self.assertEqual(windows[0]["duration_s"], 20)
+            self.assertEqual(
+                windows[0]["window_source"],
+                "bench_main_marker_plus_duration",
+            )
+
+    def test_dcgm_and_timestamped_cpu_parsers(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            dcgm = root / "dcgm.log"
+            dcgm.write_text(
+                "wall_time_ns,node,dcgm_output\n"
+                "1000000000,head,GPU 0 0.900 0.700 0.300 0.200 0.500 1000000000 2000000000\n",
+                encoding="utf-8",
+            )
+            mpstat = root / "mpstat.log"
+            mpstat.write_text(
+                "1000000000\thead\t12:00:00 PM  all  10.00 0.00 5.00 1.00 "
+                "0.00 2.00 0.00 0.00 0.00 82.00\n",
+                encoding="utf-8",
+            )
+
+            dcgm_rows = attribution.parse_dcgm(dcgm)
+            cpu_rows = attribution.parse_mpstat(mpstat)
+
+            self.assertEqual(dcgm_rows[0]["sm_active"], 0.7)
+            self.assertEqual(dcgm_rows[0]["pcie_rx_bytes_per_s"], 2_000_000_000)
+            self.assertEqual(cpu_rows[0]["cpu_active_pct"], 18)
+            self.assertEqual(cpu_rows[0]["iowait_pct"], 1)
+
+    def test_communication_aggregate_separates_local_and_cross_numa(self) -> None:
+        rows = [
+            {
+                "test": "p2p_copy",
+                "node": "head",
+                "source_gpu": 0,
+                "destination_gpu": 1,
+                "size_bytes": 1024,
+                "bandwidth_GBps": 20,
+            },
+            {
+                "test": "p2p_copy",
+                "node": "head",
+                "source_gpu": 0,
+                "destination_gpu": 4,
+                "size_bytes": 1024,
+                "bandwidth_GBps": 10,
+            },
+            {
+                "test": "all_reduce",
+                "scope": "two_node_16gpu",
+                "nccl_cross_nic": "1",
+                "size_bytes": 1024,
+                "mean_ms": 1,
+                "algbw_GBps": 1,
+                "busbw_GBps": 1.875,
+                "wrong_values": 0,
+            },
+        ]
+
+        aggregate = attribution.aggregate_communication_rows(rows)
+
+        self.assertEqual(len(aggregate), 3)
+        path_classes = {
+            row.get("path_class")
+            for row in aggregate
+            if row["test"] == "p2p_copy"
+        }
+        self.assertEqual(
+            path_classes,
+            {"same_pcie_switch", "cross_numa_sys"},
+        )
+
     @staticmethod
     def _write_csv(
         path: Path,
diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/quick_map_results.py b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/quick_map_results.py
index 6e1878c..866a9b1 100755
--- a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/quick_map_results.py
+++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/quick_map_results.py
@@ -9,8 +9,9 @@ import json
 import math
 import re
 import statistics
+import time
 from collections import defaultdict
-from datetime import datetime
+from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Any
 
@@ -33,6 +34,10 @@ SUMMARY_FIELDS = [
     "started_at",
     "ended_at",
     "elapsed_s",
+    "measurement_started_at",
+    "measurement_ended_at",
+    "measurement_duration_s",
+    "measurement_window_source",
     "completed",
     "failed",
     "duration_s",
@@ -305,6 +310,7 @@ def parse_scenarios(path: Path) -> list[dict[str, Any]]:
 
 
 def write_case(args: argparse.Namespace) -> None:
+    measurement = measurement_window(args.measurement_marker, args.bench_file)
     value = {
         "run_id": args.run_id,
         "suite": args.suite,
@@ -323,6 +329,7 @@ def write_case(args: argparse.Namespace) -> None:
         "started_at": args.started_at,
         "ended_at": args.ended_at,
         "elapsed_s": args.elapsed_s,
+        **measurement,
         "bench_file": args.bench_file,
         "bench_log": args.bench_log,
         "note": args.note,
@@ -330,6 +337,51 @@ def write_case(args: argparse.Namespace) -> None:
     write_json(args.path, value)
 
 
+def write_measurement_start(path: Path) -> None:
+    write_json(
+        path,
+        {
+            "recorded_at": datetime.now()
+            .astimezone()
+            .isoformat(timespec="microseconds"),
+            "wall_time_ns": time.time_ns(),
+            "source": "bench_log_main_marker",
+        },
+    )
+
+
+def measurement_window(marker_path: str, bench_file: str) -> dict[str, Any]:
+    empty = {
+        "measurement_started_at": None,
+        "measurement_ended_at": None,
+        "measurement_duration_s": None,
+        "measurement_window_source": "unavailable",
+    }
+    marker = Path(marker_path) if marker_path else None
+    bench = Path(bench_file) if bench_file else None
+    if marker is None or bench is None or not marker.exists():
+        return empty
+    try:
+        marker_value = read_json(marker)
+        started_at = str(marker_value["recorded_at"])
+        started = datetime.fromisoformat(started_at)
+    except (OSError, KeyError, TypeError, ValueError, json.JSONDecodeError):
+        return empty
+    bench_value = read_bench_output(bench)
+    if bench_value is None:
+        return empty
+    duration_s = first_float(bench_value, "duration", "benchmark_duration")
+    if duration_s is None or duration_s <= 0:
+        return empty
+    ended = started + timedelta(seconds=duration_s)
+    return {
+        "measurement_started_at": started_at,
+        "measurement_ended_at": ended.isoformat(timespec="microseconds"),
+        "measurement_duration_s": duration_s,
+        "measurement_window_source": "bench_main_marker_plus_duration",
+    }
+
+
 def write_manifest(args: argparse.Namespace) -> None:
     value: dict[str, Any] = {}
     if args.path.exists():
@@ -618,6 +670,7 @@ def add_case_arguments(parser: argparse.ArgumentParser) -> None:
     parser.add_argument("--started-at", required=True)
     parser.add_argument("--ended-at", required=True)
     parser.add_argument("--elapsed-s", type=float, required=True)
+    parser.add_argument("--measurement-marker", default="")
     parser.add_argument("--bench-file", required=True)
     parser.add_argument("--bench-log", required=True)
     parser.add_argument("--note", default="")
@@ -678,6 +731,9 @@ def main() -> None:
     mark_failed_parser.add_argument("--error-type", required=True)
     mark_failed_parser.add_argument("--note", required=True)
 
+    measurement_start_parser = subparsers.add_parser("mark-measurement-start")
+    measurement_start_parser.add_argument("--path", type=Path, required=True)
+
     summarize_parser = subparsers.add_parser("summarize")
     summarize_parser.add_argument("result_dir", type=Path)
 
@@ -708,6 +764,8 @@ def main() -> None:
         complete_manifest(args.path, args.status)
     elif args.command == "mark-case-failed":
         mark_case_failed(args.path, args.error_type, args.note)
+    elif args.command == "mark-measurement-start":
+        write_measurement_start(args.path)
     elif args.command == "summarize":
         summarize(args.result_dir)
 
diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/run_quick_map.sh b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/run_quick_map.sh
index 12c1284..5f9411f 100755
--- a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/run_quick_map.sh
+++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/run_quick_map.sh
@@ -483,6 +483,7 @@ write_case_meta() {
   local bench_file="${18}"
   local bench_log="${19}"
   local note="${20}"
+  local measurement_marker="${21}"
 
   python3 "${RESULT_TOOL}" write-case \
     --path "${meta_path}" \
@@ -503,6 +504,7 @@ write_case_meta() {
     --started-at "${started_at}" \
     --ended-at "${ended_at}" \
     --elapsed-s "${elapsed_s}" \
+    --measurement-marker "${measurement_marker}" \
     --bench-file "${bench_file}" \
     --bench-log "${bench_log}" \
     --note "${note}"
@@ -542,6 +544,25 @@ detect_error_type() {
   fi
 }
 
+watch_bench_main_start() {
+  local bench_log="$1"
+  local bench_pid="$2"
+  local marker_path="$3"
+
+  while kill -0 "${bench_pid}" 2>/dev/null; do
+    if grep -Fq "Starting main benchmark run" "${bench_log}" 2>/dev/null; then
+      python3 "${RESULT_TOOL}" mark-measurement-start --path "${marker_path}"
+      return 0
+    fi
+    sleep 0.1
+  done
+  if grep -Fq "Starting main benchmark run" "${bench_log}" 2>/dev/null; then
+    python3 "${RESULT_TOOL}" mark-measurement-start --path "${marker_path}"
+    return 0
+  fi
+  return 1
+}
+
 run_bench_case() {
   local suite="$1"
   local case_id="$2"
@@ -563,6 +584,7 @@ run_bench_case() {
   local bench_log="${case_dir}/bench.log"
   local meta_path="${case_dir}/meta.json"
   local command_file="${case_dir}/bench_cmd.txt"
+  local measurement_marker="${case_dir}/measurement_start.json"
 
   if case_already_completed "${meta_path}" "${bench_file}"; then
     log "SKIP completed case=${case_id} rep=${repetition}"
@@ -591,14 +613,22 @@ run_bench_case() {
     "${BENCH_CMD[@]}" > "${command_file}"
 
   local started_at start_epoch ended_at elapsed_s rc status error_type
+  local bench_pid marker_pid
   started_at="$(iso_now)"
   start_epoch="$(date +%s)"
   log "START case=${case_id} rep=${repetition} isl=${isl} osl=${osl} c=${concurrency}"
+  rm -f "${measurement_marker}"
 
   set +e
   timeout --signal=TERM --kill-after=30s "${timeout_s}s" \
-    "${BENCH_CMD[@]}" > "${bench_log}" 2>&1
+    "${BENCH_CMD[@]}" > "${bench_log}" 2>&1 &
+  bench_pid=$!
+  watch_bench_main_start \
+    "${bench_log}" "${bench_pid}" "${measurement_marker}" &
+  marker_pid=$!
+  wait "${bench_pid}"
   rc=$?
+  wait "${marker_pid}" >/dev/null 2>&1 || true
   set -e
 
   ended_at="$(iso_now)"
@@ -618,7 +648,8 @@ run_bench_case() {
     "${meta_path}" "${suite}" "${case_id}" "${role}" "${stage}" \
     "${repetition}" "${isl}" "${osl}" "${concurrency}" "${num_prompts}" \
     "${warmup_requests}" "${status}" "${error_type}" "${rc}" "${started_at}" \
-    "${ended_at}" "${elapsed_s}" "${bench_file}" "${bench_log}" "${note}"
+    "${ended_at}" "${elapsed_s}" "${bench_file}" "${bench_log}" "${note}" \
+    "${measurement_marker}"
 
   LAST_CASE_STATUS="${status}"
   LAST_CASE_ERROR="${error_type}"
diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/tests/test_quick_map_results.py b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/tests/test_quick_map_results.py
index e1890ba..fae8944 100644
--- a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/tests/test_quick_map_results.py
+++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/tests/test_quick_map_results.py
@@ -4,6 +4,7 @@ import json
 import sys
 import tempfile
 import unittest
+from datetime import datetime
 from pathlib import Path
 
 
@@ -111,6 +112,35 @@ class QuickMapResultsTest(unittest.TestCase):
             self.assertIn("long_prefill_latency_128k_c1", report_text)
             self.assertIn("OOM", report_text)
 
+    def test_measurement_window_uses_main_marker_and_benchmark_duration(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            marker = root / "measurement_start.json"
+            bench = root / "bench.jsonl"
+            marker.write_text(
+                json.dumps(
+                    {
+                        "recorded_at": "2026-07-31T12:00:00.250000+08:00",
+                        "wall_time_ns": 1,
+                    }
+                ),
+                encoding="utf-8",
+            )
+            bench.write_text(json.dumps({"duration": 12.5}) + "\n", encoding="utf-8")
+
+            window = quick_map_results.measurement_window(str(marker), str(bench))
+
+            self.assertEqual(window["measurement_duration_s"], 12.5)
+            self.assertEqual(
+                window["measurement_window_source"],
+                "bench_main_marker_plus_duration",
+            )
+            ended = datetime.fromisoformat(str(window["measurement_ended_at"]))
+            self.assertEqual(
+                ended,
+                datetime.fromisoformat("2026-07-31T12:00:12.750000+08:00"),
+            )
+
     @staticmethod
     def _write_meta(
         path: Path,