mlcommons · FileSystemGuy · Jul 21, 2026 · Jul 21, 2026 · Jul 21, 2026 · Jul 21, 2026
@@ -858,11 +858,19 @@ def _aggregate_option_results(
                 except Exception as e:
                     self.logger.warning(f"Failed to parse {result_file}: {e}")
                     missing_files.append(str(result_file))
-            all_read_bw.append(sum(trial_read_bw))
-            all_write_bw.append(sum(trial_write_bw))
-            all_avg_throughput.append(sum(trial_avg_throughput))
-            all_storage_throughput.append(sum(trial_storage_throughput))
-            all_p95_latency.append(max(trial_p95_latency) if trial_p95_latency else 0.0)
+            # A trial that produced no rank data (every rank file missing or
+            # unparseable) must not contribute a sum([])==0.0 into the
+            # across-trial aggregate — that 0.0 is a total-loss sentinel, not
+            # a measured value, and reportgen surfaces it verbatim (P5). The
+            # five per-rank lists are populated together, so any one being
+            # non-empty means the trial had at least one parsed rank; a
+            # present-rank 0.0 (CPU tier) is legitimate and still counted.
+            if trial_read_bw:
+                all_read_bw.append(sum(trial_read_bw))
+                all_write_bw.append(sum(trial_write_bw))
+                all_avg_throughput.append(sum(trial_avg_throughput))
+                all_storage_throughput.append(sum(trial_storage_throughput))
+                all_p95_latency.append(max(trial_p95_latency))
         if missing_files:
             hosts = getattr(self.args, 'hosts', None) or []
             multi_host = any(not _is_localhost(h.split(':')[0]) for h in hosts)
@@ -879,10 +887,12 @@ def _aggregate_option_results(
                 )
         return {
             'option': option,
-            'aggregated_read_bandwidth_gbps': fmean(all_read_bw) if all_read_bw else 0.0,
-            'aggregated_write_bandwidth_gbps': fmean(all_write_bw) if all_write_bw else 0.0,
-            'aggregated_avg_throughput_tokens_per_sec': fmean(all_avg_throughput) if all_avg_throughput else 0.0,
-            'aggregated_storage_throughput_tokens_per_sec': fmean(all_storage_throughput) if all_storage_throughput else 0.0,
+            # Total loss across every trial -> None (blank), never a fake 0.0
+            # (P5). aggregated_p95_latency_ms already did this.
+            'aggregated_read_bandwidth_gbps': fmean(all_read_bw) if all_read_bw else None,
+            'aggregated_write_bandwidth_gbps': fmean(all_write_bw) if all_write_bw else None,
+            'aggregated_avg_throughput_tokens_per_sec': fmean(all_avg_throughput) if all_avg_throughput else None,
+            'aggregated_storage_throughput_tokens_per_sec': fmean(all_storage_throughput) if all_storage_throughput else None,
             'aggregated_p95_latency_ms': max(all_p95_latency) if all_p95_latency else None,
             'rank_count': expected_rank_count,
             'trial_count': len(trial_dirs),

@@ -1014,10 +1014,10 @@ def _process_single_run(self, benchmark_run: BenchmarkRun) -> None:
     # ------------------------------------------------------------------
     #
     # ``_aggregate_workload_metrics`` is the single point where per-workload
-    # aggregation math lives. It is added here (Plan 06-02) but not yet
-    # wired into ``_process_workload_groups``; wiring happens in Plan 06-04
-    # (the TODO at ``metrics={}`` in this file's ``_process_workload_groups``
-    # is the eventual call site).
+    # aggregation math lives. It is wired into the reporting pass at the
+    # ``aggregated_metrics = self._aggregate_workload_metrics(...)`` call in
+    # ``accumulate_results`` (the ``metrics={}`` in ``_model_group_folder``'s
+    # proxy Result is a throwaway for folder derivation, not this call site).
     #
     # Dispatch on ``runs[0].benchmark_type``:
     #   - training       -> ``_aggregate_training``  (D-19, 5-run mean)
@@ -1040,8 +1040,8 @@ def _aggregate_workload_metrics(
         Compute aggregated metrics for one workload (per D-19..D-22).
 
         Dispatches on ``runs[0].benchmark_type`` and returns the aggregated
-        metric dict that eventually populates the ``metrics=`` slot on the
-        workload ``Result`` (see the TODO at ``_process_workload_groups``).
+        metric dict that populates the ``metrics=`` slot on the workload
+        ``Result`` (wired in ``accumulate_results``).
 
         Args:
             runs: The workload's ``BenchmarkRun`` invocations. For training,
@@ -1330,9 +1330,10 @@ def _aggregate_vdb(
         VDB-branch aggregation (D-21) — pass-through, NOT math.
 
         vdb's internal ``vdb-aggregate`` tool (see
-        ``benchmarks/vectordbbench.py:508``) owns the math contract per
-        the D-22 boundary; this helper copies pre-computed values from
-        the workload's ``summary.json`` into ``vdb_*`` columns.
+        ``VectorDBBenchmark._run_aggregate`` in ``benchmarks/vectordbbench.py``)
+        owns the math contract per the D-22 boundary; this helper copies
+        pre-computed values from the workload's ``summary.json`` into
+        ``vdb_*`` columns.
 
         Read fields (per ``submission_checker/checks/vdb_checks.py:44-51``
         ``_REQUIRED_METRIC_FIELDS``): ``throughput_qps``,

@@ -768,6 +768,42 @@ def extract(benchmark) -> BenchmarkRunData:
         )
 
 
+def _filter_list_metrics(metric_block, logger=None, context: str = ""):
+    """Filter a DLIO summary ``metric`` block to list-valued keys only.
+
+    The report-generator aggregation paths call ``fmean()`` over each metric
+    value, so scalar/string entries (``train_au_mean_percentage``,
+    ``train_au_meet_expectation``, ...) must be dropped or ``fmean`` raises
+    ``TypeError`` on a non-iterable. Returns the filtered dict, or ``None`` if
+    ``metric_block`` is not a dict or nothing survives the filter.
+
+    A diagnostic is logged (never silent) both when non-list keys are dropped
+    and when the block collapses to nothing, so a blank ``metrics`` column is
+    always traceable to a logged reason rather than disappearing quietly.
+    """
+    if not isinstance(metric_block, dict):
+        return None
+
+    filtered = {k: v for k, v in metric_block.items() if isinstance(v, list)}
+    dropped = [k for k, v in metric_block.items() if not isinstance(v, list)]
+
+    if dropped and logger:
+        logger.warning(
+            f"{context}dropped {len(dropped)} non-list metric key(s) not "
+            f"usable for aggregation: {sorted(dropped)}"
+        )
+
+    if not filtered:
+        if logger:
+            logger.warning(
+                f"{context}no list-valued metrics remain after filtering; "
+                f"metrics will be blank for this run"
+            )
+        return None
+
+    return filtered
+
+
 class DLIOResultParser:
     """Parses DLIO benchmark result files into BenchmarkRunData."""
 
@@ -843,7 +879,12 @@ def parse(self, result_dir: str, metadata: Optional[Dict] = None) -> BenchmarkRu
             parameters=hydra_workload_config,
             override_parameters=override_parameters,
             system_info=system_info,
-            metrics=summary.get("metric"),
+            # Filter to list-valued keys (matching the metadata-complete
+            # path) so report_generator's fmean(metric_list) never receives
+            # a scalar; dropped/blanked keys are logged, not silent.
+            metrics=_filter_list_metrics(
+                summary.get("metric"), self.logger, context=f"{result_dir}: "
+            ),
             result_dir=result_dir,
             accelerator=accelerator,
             run_args=(metadata or {}).get('args', {}) or {},
@@ -891,7 +932,7 @@ def extract(self, result_dir: str, logger=None) -> BenchmarkRunData:
         metadata = self._load_metadata(result_dir, logger)
 
         if metadata and self._is_complete_metadata(metadata):
-            return self._from_metadata(metadata, result_dir)
+            return self._from_metadata(metadata, result_dir, logger=logger)
 
         if self.result_parser is None:
             self.result_parser = DLIOResultParser(logger=logger)
@@ -916,7 +957,7 @@ def _is_complete_metadata(self, metadata: Dict) -> bool:
         required_fields = ['benchmark_type', 'run_datetime', 'num_processes', 'parameters']
         return all(field in metadata for field in required_fields)
 
-    def _from_metadata(self, metadata: Dict, result_dir: str) -> BenchmarkRunData:
+    def _from_metadata(self, metadata: Dict, result_dir: str, logger=None) -> BenchmarkRunData:
         """Create BenchmarkRunData from a complete metadata dict."""
         benchmark_type_str = metadata.get('benchmark_type', '')
         benchmark_type = None
@@ -944,16 +985,14 @@ def _from_metadata(self, metadata: Dict, result_dir: str) -> BenchmarkRunData:
                 if not end_datetime:
                     end_datetime = summary.get('end_time') or summary.get('end', '')
                 if not run_metrics:
-                    metric_block = summary.get('metric')
-                    if isinstance(metric_block, dict):
-                        # Filter to list-valued keys so _aggregate_training's
-                        # fmean(metric_list) doesn't blow up on scalars like
-                        # train_au_mean_percentage or strings like
-                        # train_au_meet_expectation.
-                        run_metrics = {
-                            k: v for k, v in metric_block.items()
-                            if isinstance(v, list)
-                        } or None
+                    # Filter to list-valued keys so _aggregate_training's
+                    # fmean(metric_list) doesn't blow up on scalars like
+                    # train_au_mean_percentage or strings like
+                    # train_au_meet_expectation. Dropped/blanked keys are
+                    # logged rather than silently discarded.
+                    run_metrics = _filter_list_metrics(
+                        summary.get('metric'), logger, context=f"{result_dir}: "
+                    )
 
         return BenchmarkRunData(
             benchmark_type=benchmark_type,

@@ -291,15 +291,19 @@ def closed_accelerators_per_host(self):
     @rule("4.3.4", "checkpointAggregateAcceleratorMemory")
     def aggregate_accelerator_memory(self):
         """
-        Verify total accelerator memory >= checkpoint size.
-        H100 has 80GB per accelerator.
+        Verify total accelerator memory >= checkpoint size (warnings-only).
+
+        The per-accelerator memory baseline is the H100's 80 GiB; checkpointing
+        metadata carries no accelerator-type field, so this is advisory. A
+        shortfall is surfaced to reviewers via ``warn_violation`` and never
+        flips a submission INVALID (late-window doctrine). Always returns True.
         (Rules.md 4.3.4)
         """
         valid = True
         if self.mode != "checkpointing":
             return valid
 
-        ACCELERATOR_MEMORY_GB = 80  # H100
+        ACCELERATOR_MEMORY_GB = 80  # H100 baseline (advisory)
 
         for summary, metadata, _ in self._iter_valid_files():
             checkpoint_size_gb = summary.get("metric", {}).get("checkpoint_size_GB", 0)
@@ -308,13 +312,15 @@ def aggregate_accelerator_memory(self):
             total_accelerator_memory = num_accelerators * ACCELERATOR_MEMORY_GB
 
             if total_accelerator_memory < checkpoint_size_gb:
-                self.log_violation(
+                self.warn_violation(
                     "4.3.4", "checkpointAggregateAcceleratorMemory", self.path,
-                    "aggregate accelerator memory %.2fGiB < checkpoint size %.2fGiB",
+                    "aggregate accelerator memory %.2fGiB (%d accelerators x 80 GiB "
+                    "H100 baseline) < checkpoint size %.2fGiB; verify against the "
+                    "actual accelerator's memory (Rules.md 4.3.4).",
                     total_accelerator_memory,
+                    num_accelerators,
                     checkpoint_size_gb,
                 )
-                valid = False
 
         return valid
 

@@ -49,7 +49,7 @@ def get_keys(self):
         return self.keys
 
     def __contains__(self, key):
-        return key in self.messages
+        return key in self.d
 
     def __repr__(self):
         return f"<SummaryParser path={self.path!r} keys={len(self.keys)}>"
@@ -475,6 +475,7 @@ def build_submission(tmp_path, **overrides) -> Path:
     chkpt_checkpoint_folder = overrides.pop("chkpt_checkpoint_folder", None)
     chkpt_results_dir = overrides.pop("chkpt_results_dir", None)
     chkpt_summary_checkpoint_size_GB = overrides.pop("chkpt_summary_checkpoint_size_GB", None)
+    chkpt_summary_num_accelerators = overrides.pop("chkpt_summary_num_accelerators", None)
     run_data_dir = overrides.pop("run_data_dir", None)
     run_results_dir = overrides.pop("run_results_dir", None)
 
@@ -818,6 +819,8 @@ def build_submission(tmp_path, **overrides) -> Path:
                     metric = dict(chkpt_summary.get("metric", {}))
                     metric["checkpoint_size_GB"] = chkpt_summary_checkpoint_size_GB
                     chkpt_summary["metric"] = metric
+                if chkpt_summary_num_accelerators is not None:
+                    chkpt_summary["num_accelerators"] = chkpt_summary_num_accelerators
 
                 (ts_dir / "summary.json").write_text(
                     json.dumps(chkpt_summary), encoding="utf-8"

@@ -136,6 +136,52 @@ def test_4_3_1_undersized_checkpoint_emits_warn_prefix_not_error(tmp_path, mock_
     )
 
 
+# ---------------------------------------------------------------------------
+# 4.3.4 aggregate_accelerator_memory — warnings-only (C2)
+# ---------------------------------------------------------------------------
+
+def test_4_3_4_insufficient_memory_warns_not_errors(tmp_path, mock_logger):
+    """4.3.4: aggregate accelerator memory < checkpoint size → WARNING, never fail.
+
+    Late-stage submission-window doctrine: this check must not flip a
+    submission INVALID. With 8 accelerators (× the 80 GiB baseline = 640 GiB)
+    and a 1000 GiB checkpoint, the aggregate is insufficient, so the rule
+    warns via warn_violation and still returns True.
+    """
+    from mlpstorage_py.tests.conftest import build_submission
+    root = build_submission(
+        tmp_path,
+        chkpt_summary_num_accelerators=8,
+        chkpt_summary_checkpoint_size_GB=1000,
+    )
+    check = _run_checkpointing_check(root, mock_logger)
+    result = check.aggregate_accelerator_memory()
+    assert result is True
+    assert any(
+        m.startswith("[4.3.4 checkpointAggregateAcceleratorMemory]")
+        for m in mock_logger.warnings
+    ), f"expected [4.3.4 ...] warning; got warnings={mock_logger.warnings}"
+    assert not any(
+        m.startswith("[4.3.4 ") for m in mock_logger.errors
+    ), f"4.3.4 must warn, not error; got errors={mock_logger.errors}"
+
+
+def test_4_3_4_sufficient_memory_no_warning(tmp_path, mock_logger):
+    """Aggregate accelerator memory >= checkpoint size → no warning, returns True."""
+    from mlpstorage_py.tests.conftest import build_submission
+    root = build_submission(
+        tmp_path,
+        chkpt_summary_num_accelerators=8,
+        chkpt_summary_checkpoint_size_GB=100,  # 8*80=640 GiB >= 100
+    )
+    check = _run_checkpointing_check(root, mock_logger)
+    result = check.aggregate_accelerator_memory()
+    assert result is True
+    assert not any(
+        m.startswith("[4.3.4 ") for m in mock_logger.warnings + mock_logger.errors
+    ), f"expected no 4.3.4 violation; got {mock_logger.warnings + mock_logger.errors}"
+
+
 # ---------------------------------------------------------------------------
 # Behavior-preservation: 4.4.1 checkpointPathArgs
 # ---------------------------------------------------------------------------

@@ -0,0 +1,31 @@
+"""Unit tests for the submission-checker JSONParser.
+
+Covers the ``in`` membership operator (``__contains__``), which referenced a
+non-existent ``self.messages`` attribute and raised ``AttributeError`` for any
+``key in parser`` test — a latent bug (callers use ``[]`` / ``.get()``).
+"""
+
+import json
+
+from mlpstorage_py.submission_checker.parsers.json_parser import JSONParser
+
+
+def _parser(tmp_path, payload):
+    path = tmp_path / "summary.json"
+    path.write_text(json.dumps(payload), encoding="utf-8")
+    return JSONParser(str(path))
+
+
+def test_contains_top_level_key(tmp_path):
+    """``key in parser`` is True for a present top-level key, False otherwise."""
+    parser = _parser(tmp_path, {"alpha": 1, "beta": 2})
+    assert "alpha" in parser
+    assert "beta" in parser
+    assert "missing" not in parser
+
+
+def test_contains_on_normalized_non_dict(tmp_path):
+    """A non-dict JSON body is normalized under ``summary`` and is found via ``in``."""
+    parser = _parser(tmp_path, [1, 2, 3])
+    assert "summary" in parser
+    assert "alpha" not in parser
@@ -620,7 +620,9 @@ def test_uses_glob_not_constructed_filename(self, tmp_path):
         assert result['partial_failure'] is False
 
     def test_none_p95_when_no_successful_reads(self, tmp_path):
-        """aggregated_p95_latency_ms is 0.0 when all rank files are missing."""
+        """P5: aggregated_p95_latency_ms is None when all rank files are
+        missing (total data loss) — a trial that produced no rank data must
+        not contribute a fake 0.0."""
         bm = _make_run_benchmark(tmp_path)
         trial_dir = tmp_path / 'trial_0'
         # Both rank dirs exist but have no json files
@@ -629,8 +631,48 @@ def test_none_p95_when_no_successful_reads(self, tmp_path):
 
         result = bm._aggregate_option_results(1, [str(trial_dir)], expected_rank_count=2)
 
-        # Empty trial contributes 0.0; fmean([0.0]) = 0.0
-        assert result['aggregated_p95_latency_ms'] == pytest.approx(0.0)
+        # Total loss -> None (blank), not a misleading measured 0.0.
+        assert result['aggregated_p95_latency_ms'] is None
+        assert result['partial_failure'] is True
+
+    def test_total_data_loss_emits_none_not_zero(self, tmp_path):
+        """P5: when NO rank produced data across all trials (total loss),
+        every aggregated bandwidth/throughput field must be None (blank),
+        not a misleading measured 0.0 that reportgen surfaces verbatim."""
+        bm = _make_run_benchmark(tmp_path)
+        trial_dir = tmp_path / 'trial_0'
+        # Rank dirs exist but carry no result JSON -> nothing parsed.
+        (trial_dir / 'rank_0').mkdir(parents=True)
+        (trial_dir / 'rank_1').mkdir(parents=True)
+
+        result = bm._aggregate_option_results(1, [str(trial_dir)], expected_rank_count=2)
+
+        assert result['aggregated_read_bandwidth_gbps'] is None
+        assert result['aggregated_write_bandwidth_gbps'] is None
+        assert result['aggregated_avg_throughput_tokens_per_sec'] is None
+        assert result['aggregated_storage_throughput_tokens_per_sec'] is None
+        assert result['aggregated_p95_latency_ms'] is None
+        assert result['partial_failure'] is True
+
+    def test_empty_trial_not_diluted_into_mean(self, tmp_path):
+        """P5: a trial where every rank is missing must not fold a 0.0 into
+        the across-trial mean — the mean is taken over the trials that
+        actually produced data."""
+        bm = _make_run_benchmark(tmp_path)
+        good = tmp_path / 'trial_0'
+        self._make_rank_file(good / 'rank_0', bw=3.0, p95=10.0)
+        self._make_rank_file(good / 'rank_1', bw=1.0, p95=12.0)  # trial sum = 4.0
+        empty = tmp_path / 'trial_1'
+        (empty / 'rank_0').mkdir(parents=True)
+        (empty / 'rank_1').mkdir(parents=True)
+
+        result = bm._aggregate_option_results(
+            1, [str(good), str(empty)], expected_rank_count=2
+        )
+
+        # Old behavior folded the empty trial's 0.0 in: fmean([4.0, 0.0]) = 2.0.
+        # Correct: mean over the single trial that had data = 4.0.
+        assert result['aggregated_read_bandwidth_gbps'] == pytest.approx(4.0)
         assert result['partial_failure'] is True