fix(health): fail TransferBench healthcheck test when TransferBench reports a failure

CHANGELOG.md

@@ -48,6 +48,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Kernel-log health checks no longer echo each node's full dmesg or journal into the run log and HTML report. Each scan logs one summary line, and detected errors are still reported.
 - TorchTitan no longer writes the Hugging Face token into tokenizer-download or training commands. Each node reads `paths.hf_token_file` inside the container when that command runs. If that file is missing, empty, or not mounted, the stage fails and later stages are skipped.
 - Megatron and Primus no longer write the Hugging Face token into training, tokenizer-download, or job-launch commands. Each node reads `paths.hf_token_file` inside the container when that command runs. If that file is missing, empty, or not mounted, Megatron fails the test and the remaining training stages are skipped. Primus exports an empty `HF_TOKEN` and the run continues.
+- The health TransferBench `healthcheck` test fails when TransferBench reports a failed sub-test or exits with an error. It previously passed, and only the Run Deck showed the failure.
 - ibperf:
   - Runs on nodes with fewer than 8 GPUs instead of failing with `KeyError: 'card4'`.
   - `verify_bw` now enforces `expected_results`; previously tests passed regardless of measured bandwidth. Configs with `verify_bw` set to `"True"` may now fail; re-baseline thresholds for your cluster.

cvs/lib/unittests/test_transferbench_health.py

@@ -234,5 +234,78 @@ def test_missing_per_link_min_key_fails(self):
         self.assertIn('gpu_to_gpu_a2a_per_link_min', fail_test.call_args.args[0])
 
 
+HEALTHCHECK_ONE_FAIL = """\
+Testing HBM performance [READ]             ........PASS
+Testing bidirectional host<->device copies ........PASS
+Testing all-to-all XGMI copies             ........FAIL (56 test(s))
+"""
+
+HEALTHCHECK_ALL_PASS = HEALTHCHECK_ONE_FAIL.replace('FAIL (56 test(s))', 'PASS')
+
+# VERBOSE=1 prints the verdict on its own line, so no "Testing ....FAIL" line exists.
+HEALTHCHECK_VERBOSE_FAIL = """\
+Testing all-to-all XGMI copies
+   GPU 00 to GPU 01: : Measured  40.17 Limit  43.65
+FAIL (56 test(s))
+"""
+
+
+class TestParseTbHealthcheck(unittest.TestCase):
+    def test_all_pass_does_not_fail(self):
+        with patch.object(tb, 'fail_test') as fail_test:
+            tb.parse_tb_healthcheck({'nodeA': {'output': HEALTHCHECK_ALL_PASS, 'exit_code': 0}})
+        fail_test.assert_not_called()
+
+    def test_failed_subtest_is_named(self):
+        with patch.object(tb, 'fail_test') as fail_test:
+            tb.parse_tb_healthcheck({'nodeA': {'output': HEALTHCHECK_ONE_FAIL, 'exit_code': 2}})
+        fail_test.assert_called_once()
+        message = fail_test.call_args.args[0]
+        self.assertIn('all-to-all XGMI copies', message)
+        self.assertIn('56 test(s)', message)
+        self.assertIn('nodeA', message)
+
+    def test_nonzero_exit_without_fail_line_fails(self):
+        with patch.object(tb, 'fail_test') as fail_test:
+            tb.parse_tb_healthcheck({'nodeA': {'output': HEALTHCHECK_VERBOSE_FAIL, 'exit_code': 2}})
+        fail_test.assert_called_once()
+        self.assertIn('exited with code 2', fail_test.call_args.args[0])
+
+    def test_expected_host_without_result_fails(self):
+        with patch.object(tb, 'fail_test') as fail_test:
+            tb.parse_tb_healthcheck(
+                {'nodeA': {'output': HEALTHCHECK_ALL_PASS, 'exit_code': 0}}, expected_hosts=['nodeA', 'nodeB']
+            )
+        fail_test.assert_called_once()
+        self.assertIn('no result for node nodeB', fail_test.call_args.args[0])
+
+    def test_no_results_fails_every_expected_host(self):
+        with patch.object(tb, 'fail_test') as fail_test:
+            tb.parse_tb_healthcheck({}, expected_hosts=['nodeA', 'nodeB'])
+        self.assertEqual(fail_test.call_count, 2)
+
+    def test_generic_scan_misses_healthcheck_failure(self):
+        from cvs.lib import utils_lib
+
+        with patch('cvs.lib.utils_lib.fail_test') as fail_test:
+            utils_lib.scan_test_results({'nodeA': HEALTHCHECK_ONE_FAIL})
+        fail_test.assert_not_called()
+
+    def test_run_transferbench_requests_detailed_results(self):
+        orch = MagicMock()
+        orch.orchestrator_type = 'baremetal'
+        orch.sudo_prefix.return_value = 'sudo -n '
+        with patch.object(tb, 'detect_rocm_path', return_value='/opt/rocm'):
+            tb.run_transferbench(
+                orch,
+                {'path': '/tb', 'rocm_path': '/opt/rocm', 'num_cpu_devices': 2},
+                'healthcheck',
+                timeout=30,
+                detailed=True,
+            )
+        self.assertTrue(orch.exec.call_args.kwargs['detailed'])
+        self.assertFalse(orch.exec.call_args.kwargs['print_console'])
+
+
 if __name__ == '__main__':
     unittest.main()

cvs/tests/health/transferbench_cvs.py

@@ -184,7 +184,7 @@ def build_transferbench_command(path, rocm_path, preset, extra_env=None, sudo_pr
     return f'{sudo_prefix}bash -c {shlex.quote(inner)}'
 
 
-def run_transferbench(orch, config_dict, preset, timeout, extra_env=None, lifecycle=None, stage=None):
+def run_transferbench(orch, config_dict, preset, timeout, extra_env=None, lifecycle=None, stage=None, detailed=False):
     """Resolve ROCm/env and execute one TransferBench preset on all orch hosts."""
     path = config_dict['path']
     rocm_path = detect_rocm_path(orch, config_dict.get('rocm_path', ''))
@@ -194,7 +194,8 @@ def run_transferbench(orch, config_dict, preset, timeout, extra_env=None, lifecy
     cmd = build_transferbench_command(path, rocm_path, preset, env, sudo_prefix=_payload_sudo_prefix(orch))
     log.info('TransferBench command: %s', cmd)
     with timed_stage(lifecycle, stage or preset):
-        return orch.exec(cmd, timeout=timeout)
+        # The sharded SSH handle can only print string results, so detailed callers log the output themselves.
+        return orch.exec(cmd, timeout=timeout, detailed=detailed, print_console=not detailed)
 
 
 def _tb_output_excerpt(out, tail=6000):
@@ -462,6 +463,30 @@ def parse_tb_schmoo_bw(out_dict, exp_dict):
             )
 
 
+# e.g. "Testing all-to-all XGMI copies             ........FAIL (56 test(s))"
+_TB_HEALTHCHECK_FAIL_RE = re.compile(r'(?m)^[ \t]*Testing\s+(.+?)\s*\.{2,}\s*FAIL(?:\s*\((\d+)\s+test\(s\)\))?')
+
+
+def parse_tb_healthcheck(results, expected_hosts=()):
+    """Fail on a non-zero TransferBench healthcheck exit code, naming failed sub-tests from the output.
+
+    Sharded SSH runs omit hosts pruned as unreachable earlier, so every expected host must have a result.
+    """
+    for node in expected_hosts:
+        if node not in results:
+            fail_test(f"TransferBench healthcheck returned no result for node {node}")
+    for node, res in results.items():
+        output = res.get('output') or ''
+        failed = _TB_HEALTHCHECK_FAIL_RE.findall(output)
+        for name, count in failed:
+            fail_test(f"TransferBench healthcheck '{name.strip()}' FAIL ({count or '?'} test(s)) on node {node}")
+        if res.get('exit_code') != 0 and not failed:
+            fail_test(
+                f"TransferBench healthcheck exited with code {res.get('exit_code')} on node {node}. "
+                f"Output: {_tb_output_excerpt(output)}"
+            )
+
+
 def _capture_tb_rundeck(res_dict, cluster_dict, recorder, *args):
     """Best-effort: a reporting problem must not change the TransferBench pass/fail."""
     if res_dict is None:
@@ -532,9 +557,13 @@ def test_transfer_bench_healthcheck(
     globals.error_list = []
     log.info('Testcase Run TransferBench healthcheck')
     preset = config_dict.get('healthcheck_preset') or 'healthcheck'
-    out_dict = run_transferbench(orch, config_dict, preset, timeout=(60 * 3), lifecycle=lifecycle, stage='healthcheck')
+    results = run_transferbench(
+        orch, config_dict, preset, timeout=(60 * 3), lifecycle=lifecycle, stage='healthcheck', detailed=True
+    )
+    out_dict = {node: res.get('output') or '' for node, res in results.items()}
     print_test_output(log, out_dict)
     scan_test_results(out_dict)
+    parse_tb_healthcheck(results, orch.hosts)
     _capture_tb_rundeck(
         transferbench_res_dict,
         cluster_dict,