diff --git a/.automation_scripts/pytorch-unit-test-scripts/auto_classify_skip_reasons.py b/.automation_scripts/pytorch-unit-test-scripts/auto_classify_skip_reasons.py
index cf948495ec04e..d77ab51a8dfdf 100644
--- a/.automation_scripts/pytorch-unit-test-scripts/auto_classify_skip_reasons.py
+++ b/.automation_scripts/pytorch-unit-test-scripts/auto_classify_skip_reasons.py
@@ -46,6 +46,15 @@
     # TIER 1: High-specificity combined rules (message + file/class)
     # ==================================================================
 
+    # --- PT2.0 - Convolution: conv2d backward parametrized skipped on ROCm ---
+    # skipIfRocmArch(...) skips test_conv2d_backward_parametrized (see upstream
+    # #188671, CI timeout). Categorize it as a convolution issue rather than the
+    # generic Misc "test skipped on ('gfx...')" bucket below. Must precede that
+    # Misc arch rule.
+    {"reason": "PT2.0 - Convolution",
+     "msg": r"test skipped on \('gfx",
+     "name": r"(?i)conv2d_backward"},
+
     # --- bfloat16_SDPA_ME: dropout mask in test_transformers with bfloat16 in TEST NAME ---
     # Must be before generic SDPA_ME rule
     {"reason": "bfloat16_SDPA_ME",
diff --git a/.automation_scripts/pytorch-unit-test-scripts/detect_log_failures.py b/.automation_scripts/pytorch-unit-test-scripts/detect_log_failures.py
index b563fcf74bc1e..2b456f5954676 100755
--- a/.automation_scripts/pytorch-unit-test-scripts/detect_log_failures.py
+++ b/.automation_scripts/pytorch-unit-test-scripts/detect_log_failures.py
@@ -39,10 +39,19 @@
 RE_INDIVIDUAL_TEST = re.compile(
     r"(?P<test_path>\S+\.py::(?P<cls>\w+)::(?P<method>\w+))"
 )
-RE_INDIV_PASSED = re.compile(
-    r"(?:test/)?(?P<file>\S+\.py)::(?P<cls>\w+)::(?P<method>\S+?)\s+PASSED"
+# PyTorch's run_test.py prints an authoritative summary at the end of a shard
+# listing the exact tests that failed and then passed on rerun-in-new-process,
+# e.g.
+#   The following tests failed and then succeeded when run in a new process
+#   ['test/inductor/test_compiled_autograd.py::Cls::test_jacfwd', ...]
+# We parse this directly instead of guessing from nearby PASSED lines, which
+# misattributes flakiness to whatever test happened to pass most recently.
+RE_FLAKY_SUMMARY = re.compile(
+    r"The following tests failed and then succeeded when run in a new process\s*\[(?P<tests>[^\]]*)\]"
+)
+RE_FLAKY_ENTRY = re.compile(
+    r"['\"](?:test/)?(?P<file>\S+?\.py)::(?P<cls>\w+)::(?P<method>[^'\"]+?)['\"]"
 )
-RE_NEW_PROCESS_SUCCESS = re.compile(r"Test succeeded in new process")
 
 CRASH_PATTERNS = [
     (re.compile(r"Segmentation fault", re.IGNORECASE), "SEGFAULT"),
@@ -69,13 +78,23 @@
 
 
 def classify_log_file(filename):
-    """Return (platform, test_config, shard_num) from a log filename like rocm3.txt."""
+    """Return (platform, test_config, shard_num) from a log filename like rocm3.txt.
+
+    Commit-vs-commit parity prefixes log files with the short commit SHA
+    (for example, 09e0c59b_rocm3.txt). In that mode the SHA label is the
+    platform name used by generate_summary.py, so preserve it here.
+    """
     stem = Path(filename).stem
+    label = None
+    m = re.match(r"(?P<label>[0-9a-f]{8,40})_(?P<stem>.+)", stem)
+    if m:
+        label = m.group("label")[:8]
+        stem = m.group("stem")
     for prefix, (platform, test_config) in sorted(LOG_FILE_MAP.items(), key=lambda x: -len(x[0])):
         if stem.startswith(prefix):
             remainder = stem[len(prefix):]
             if remainder.isdigit():
-                return platform, test_config, int(remainder)
+                return label or platform, test_config, int(remainder)
     return None, None, None
 
 
@@ -101,16 +120,16 @@ def parse_log_file(filepath):
     and flaky tests.
 
     A flaky test is one that failed in its normal-process run but PASSED when the
-    CI harness re-ran it alone in a new subprocess (indicated by a PASSED line
-    for the specific test::class::method, followed by 'Test succeeded in new
-    process, continuing with the rest of the tests').
+    CI harness re-ran it alone in a new subprocess. These are read from the
+    authoritative summary run_test.py prints for each shard ('The following
+    tests failed and then succeeded when run in a new process[...]'), which
+    names the exact tests.
     """
     results = {}
     current_test = None
     last_failed_test = None
     consistent_failures = []
     flaky_tests = []
-    last_passed_individual = None
     first_ts = None
     last_ts = None
 
@@ -244,34 +263,27 @@ def parse_log_file(filepath):
                     shard_str = f"{info['shard']}/{info['total']}"
                 consistent_failures.append((m.group("test_path"), shard_str))
 
-            # Detect individual PASSED lines for flaky-rerun tracking.
-            m = RE_INDIV_PASSED.search(stripped)
+            # A test is flaky when it failed in its normal-process run but
+            # PASSED on rerun-in-a-new-process. run_test.py prints an
+            # authoritative summary naming exactly those tests; parse it
+            # directly rather than guessing from the most recent PASSED line
+            # (which misattributes flakiness to an unrelated test that merely
+            # happened to pass just before the rerun marker).
+            m = RE_FLAKY_SUMMARY.search(stripped)
             if m:
-                last_passed_individual = {
-                    "file": m.group("file"),
-                    "cls": m.group("cls"),
-                    "method": m.group("method"),
-                    "active": active,
-                }
-
-            # When we see 'Test succeeded in new process' after a PASSED
-            # individual test, that test was originally failing in the main
-            # process (CI only falls back to rerun-in-new-process for tests
-            # that crashed or failed) but passed on retry -> flaky.
-            if RE_NEW_PROCESS_SUCCESS.search(stripped) and last_passed_individual:
-                lp = last_passed_individual
-                lp_active = lp.get("active")
-                test_shard = ""
-                if lp_active and lp_active in results:
-                    info = results[lp_active]
-                    test_shard = f"{info['shard']}/{info['total']}"
-                flaky_tests.append({
-                    "file": lp["file"],
-                    "cls": lp["cls"],
-                    "method": lp["method"],
-                    "test_shard": test_shard,
-                })
-                last_passed_individual = None
+                for em in RE_FLAKY_ENTRY.finditer(m.group("tests")):
+                    f_file = em.group("file")
+                    test_shard = ""
+                    for info in results.values():
+                        if info["test_file"] and f_file == info["test_file"] + ".py":
+                            test_shard = f"{info['shard']}/{info['total']}"
+                            break
+                    flaky_tests.append({
+                        "file": f_file,
+                        "cls": em.group("cls"),
+                        "method": em.group("method"),
+                        "test_shard": test_shard,
+                    })
 
             if active and active in results:
                 for pattern, label in CRASH_PATTERNS:
diff --git a/.automation_scripts/pytorch-unit-test-scripts/download_testlogs b/.automation_scripts/pytorch-unit-test-scripts/download_testlogs
index 22371a1c9e9a1..8feaeaf839aa0 100755
--- a/.automation_scripts/pytorch-unit-test-scripts/download_testlogs
+++ b/.automation_scripts/pytorch-unit-test-scripts/download_testlogs
@@ -31,6 +31,22 @@ if missing_vars:
 # global variables
 error_msgs = []
 
+def _skip_missing_config(config_name, sha_val, wf_names):
+    """Record a missing ROCm test config and let the run continue.
+
+    mi300's default/distributed/inductor come from three separately-scheduled
+    workflows (rocm-mi300 / periodic-rocm-mi300 / inductor-rocm-mi300) that do
+    not always land on the same SHA, so a config can be absent for a given SHA
+    even though the arch "ran". Rather than aborting the whole download (which
+    loses the configs that *are* present), skip just this config: the report is
+    still built from the available configs, and appending to error_msgs keeps
+    the job flagged failed at exit so the gap stays visible."""
+    msg = (f"WARNING: no ROCm {config_name} workflow run found for sha {sha_val} "
+           f"({wf_names}); skipping the {config_name} config. Report will omit it "
+           f"and the job will be flagged failed.")
+    print(msg)
+    error_msgs.append(msg)
+
 # Job-name matching config (workflow names, job-name prefixes, shard counts and
 # check-run regexes) lives in a single JSON file next to this script so
 # download_testlogs and parity-auto.yml share one source of truth. See
@@ -43,6 +59,19 @@ with open(PARITY_CONFIG_PATH) as _f:
 # downloader always filters on it to avoid pulling foreign platforms' reports.
 ROCM_ARTIFACT_SUBSTRINGS = ["rocm.gpu"]
 
+# CI "variant" jobs reuse the normal shard's test-reports-<config>-<shard>-<total>
+# artifact prefix and the same "(config, shard, total" name, but only contain a
+# tiny, unrepresentative slice: rerun_disabled_tests re-runs just the currently
+# disabled/flaky tests, mem_leak_check runs a leak-checking pass. Matching one of
+# these instead of the normal job makes almost the whole suite look MISSED
+# (e.g. mi350 inductor: ~10 flaky reruns vs the full ~50k), so we track their job
+# IDs and drop both the jobs (name matching) and their artifacts.
+VARIANT_JOB_MARKERS = ("rerun_disabled_tests", "mem_leak_check")
+variant_job_ids = set()
+
+def _is_variant_job(name):
+    return any(m in (name or "") for m in VARIANT_JOB_MARKERS)
+
 # Test configs whose sources are listed in the per-arch config (in order).
 _TEST_CONFIGS = ("default", "distributed", "inductor")
 
@@ -106,7 +135,7 @@ def get_latest_commit_sha(pr_id, token):
         sys.exit(1)
 
 def write_test_log_to_file(filename, test_key, jobs, sha):
-    js = [j for j in jobs if test_key in j['name']]
+    js = [j for j in jobs if test_key in j['name'] and not _is_variant_job(j['name'])]
     if len(js) > 0:
         if len(js) > 1:
             print(f"WARNING: Found multiple jobs with key: '{test_key}', selecting first one")
@@ -183,8 +212,45 @@ def get_workflow_jobs(wf, all_attempts=False):
         for i in range(2, math.ceil(response_json['total_count']/page_size) + 1):
             page_json = _fetch_jobs_page(wf['jobs_url'], {**base_params, 'page': i})
             jobs += page_json["jobs"]
+    # Record variant (rerun_disabled_tests / mem_leak_check) job IDs so their
+    # near-empty artifacts can be excluded from the S3 download below.
+    for j in jobs:
+        if _is_variant_job(j.get('name', '')):
+            variant_job_ids.add(str(j['id']))
     return jobs
 
+def derive_shard_count(wf, job_prefix, test_config, fallback):
+    """Return the shard total upstream actually used for this (job_prefix,
+    test_config), parsed from job names like
+    '<job_prefix> / test (<config>, <idx>, <total>, ...)' -> <total>.
+
+    Shard counts differ between an arch's primary workflow and its fallback
+    (e.g. mi200 default is 6 shards in rocm-mi200 but 10 in the
+    trunk-rocm-sandbox fallback), so we read the real total from the resolved
+    run instead of assuming the config value - otherwise the constructed
+    "(default, i, 6)" keys miss the actual "(default, i, 10)" jobs. Falls back
+    to `fallback` when the run has no matching jobs.
+    """
+    try:
+        jobs = get_workflow_jobs(wf)
+    except Exception:
+        return fallback
+    pat = re.compile(
+        re.escape(job_prefix) + r" / \S+ \(" + re.escape(test_config) + r", \d+, (\d+)"
+    )
+    totals = {}
+    for job in jobs:
+        if _is_variant_job(job.get("name", "")):
+            continue
+        m = pat.search(job.get("name", ""))
+        if m:
+            total = int(m.group(1))
+            totals[total] = totals.get(total, 0) + 1
+    if not totals:
+        return fallback
+    # Most common total wins (guards against a stray reshaped/duplicate job).
+    return sorted(totals.items(), key=lambda kv: (-kv[1], -kv[0]))[0][0]
+
 def get_check_runs_for_commit(sha, prefix):
     """Get check runs for a commit filtered by name prefix.
 
@@ -385,11 +451,19 @@ def download_xml_files(workflow_run_id, workflow_run_attempts, prefixes=[], allo
                 artifact_paths += found
                 break
 
-    # Filter out rerun_disabled_tests artifacts (same prefix, different job)
+    # Filter out variant (rerun_disabled_tests / mem_leak_check) artifacts. They
+    # share the normal shard's prefix, so match by name AND by the variant job
+    # IDs recorded in get_workflow_jobs (the S3 artifact name embeds "_<jobid>"
+    # but does not always contain the variant marker itself).
     before = len(artifact_paths)
-    artifact_paths = [p for p in artifact_paths if "rerun_disabled" not in p.name]
+    artifact_paths = [
+        p for p in artifact_paths
+        if "rerun_disabled" not in p.name
+        and "mem_leak_check" not in p.name
+        and not any(f"_{vid}" in p.name for vid in variant_job_ids)
+    ]
     if before != len(artifact_paths):
-        print(f"  Filtered out {before - len(artifact_paths)} rerun_disabled artifacts")
+        print(f"  Filtered out {before - len(artifact_paths)} variant (rerun_disabled/mem_leak_check) artifacts")
 
     # Fall back to GHA artifacts if S3 returned nothing
     if len(artifact_paths) == 0:
@@ -526,7 +600,7 @@ def parse_args():
     parser.add_argument('--no_rocm', action='store_true')
     parser.add_argument('--no_cuda', action='store_true')
     parser.add_argument('--pr_id', type=int, help='The pull request ID')
-    parser.add_argument('--arch', type=str, choices=['mi200', 'mi300', 'mi350', 'navi31', 'nightly'], default='mi350', help='ROCm GPU architecture (mi200, mi300, mi350, navi31, or nightly, default: mi350)')
+    parser.add_argument('--arch', type=str, choices=['mi200', 'mi300', 'mi350', 'navi31', 'nightly', 'preview'], default='mi350', help='ROCm GPU architecture (mi200, mi300, mi350, navi31, nightly, or preview, default: mi350)')
     parser.add_argument('--include_inductor_periodic', action='store_true', help='Also download inductor-periodic benchmark artifacts (into a separate directory, not included in parity CSV)')
     parser.add_argument('--baseline_sha', type=str, help='Baseline commit SHA to compare against. Downloads the same ROCm workflows for this commit into baseline_xml/.')
     return parser.parse_args()
@@ -654,17 +728,37 @@ def main():
     token = os.getenv('GITHUB_TOKEN', '...')
     global authentication_headers
     authentication_headers = {'Authorization': f'token {token}'}
-    if (args.pr_id and args.sha1) or (not args.pr_id and not args.sha1):
-        error_msg = "Error: Please provide either pr_id or sha!"
+    status = "success"
+    # An empty --sha1 (or the literal "latest") means "use the latest green
+    # run on main" rather than a specific commit. download_workflow_run()
+    # already resolves that when given no head_sha, so treat these as unset.
+    sha_is_latest = (not args.sha1) or (str(args.sha1).strip().lower() == "latest")
+    if args.pr_id and not sha_is_latest:
+        error_msg = "Error: Please provide either pr_id or sha (not both)!"
         print(error_msg)
         sys.exit(1)
     if args.pr_id:
         pr_id = args.pr_id
-        sha = get_latest_commit_sha(pr_id, token)        
-    else:
+        sha = get_latest_commit_sha(pr_id, token)
+    elif not sha_is_latest:
         sha = args.sha1
         pr_id = None
-    status = "success"
+    else:
+        # No pr_id and no explicit sha: resolve the latest green ROCm run on
+        # main and use its head commit as the target sha.
+        pr_id = None
+        print("No sha or pr_id given; resolving latest green ROCm run on main...")
+        latest_wf = download_workflow_run(
+            created=args.created,
+            max_pages=args.max_pages,
+            workflow=ROCmWorkflowNames["default"],
+            sha=None,
+            ignore_status=args.ignore_status,
+            status=status,
+            error_msg="Error: could not find a latest green ROCm run on main",
+        )
+        sha = latest_wf["head_sha"]
+        print(f"Resolved 'latest' to sha: {sha}")
     print(f"sha: {sha}")
 
     # When comparing two commits, prefix log filenames with short SHAs
@@ -726,47 +820,46 @@ def main():
                 periodic_wf = download_workflow_run(created=args.created, max_pages=args.max_pages, workflow=fallback_wf, sha=periodic_sha, ignore_status=args.ignore_status, status=status, error_msg=error_msg)
             periodic_fallback_used = True
         if periodic_wf is None:
-            raise Exception(error_msg)
-        dist_wf_name = ROCmWorkflowNames['distributed'] if not periodic_fallback_used else periodic_fallbacks[arch][0]
-        print(f"Using workflow '{dist_wf_name}' with id:{periodic_wf['id']} for ROCm distributed")
-
-        if periodic_fallback_used and arch in periodic_fallbacks:
-            dist_job_prefix = periodic_fallbacks[arch][1]
+            _skip_missing_config("distributed", periodic_sha,
+                                 ROCmWorkflowNames['distributed'] + " and fallbacks")
         else:
-            dist_job_prefix = rocm_job_prefix['distributed']
+            dist_wf_name = ROCmWorkflowNames['distributed'] if not periodic_fallback_used else periodic_fallbacks[arch][0]
+            print(f"Using workflow '{dist_wf_name}' with id:{periodic_wf['id']} for ROCm distributed")
 
-        folder_list = get_or_create_test_folder(periodic_wf)
+            if periodic_fallback_used and arch in periodic_fallbacks:
+                dist_job_prefix = periodic_fallbacks[arch][1]
+            else:
+                dist_job_prefix = rocm_job_prefix['distributed']
 
-        # Download logs
-        # If the ROCm distributed logs aren't found you might want to check the HUD for the correct tags
-        # HUD link: https://hud.pytorch.org/hud/pytorch/pytorch/main/1?per_page=50&name_filter=rocm
-        # Make sure "Hide unstable jobs" is unselected, in case ROCm jobs are marked as unstable
-    
-        if arch == "mi350":
-            dist_shards = 3 if not periodic_fallback_used else rocm_shards["distributed"]
-        else:
-            dist_shards = rocm_shards["distributed"]
-        print(f"Using final ROCm shard count {dist_shards} for distributed")
+            folder_list = get_or_create_test_folder(periodic_wf)
 
-        if not args.artifacts_only:
-            test_log_list_rocm_distributed = [
-                [f"{current_prefix}rocm_dist{i}.txt", f"{dist_job_prefix} / test (distributed, {i}, {dist_shards}"]
+            # Download logs
+            # If the ROCm distributed logs aren't found you might want to check the HUD for the correct tags
+            # HUD link: https://hud.pytorch.org/hud/pytorch/pytorch/main/1?per_page=50&name_filter=rocm
+            # Make sure "Hide unstable jobs" is unselected, in case ROCm jobs are marked as unstable
+
+            dist_shards = derive_shard_count(periodic_wf, dist_job_prefix, "distributed", rocm_shards["distributed"])
+            print(f"Using final ROCm shard count {dist_shards} for distributed")
+
+            if not args.artifacts_only:
+                test_log_list_rocm_distributed = [
+                    [f"{current_prefix}rocm_dist{i}.txt", f"{dist_job_prefix} / test (distributed, {i}, {dist_shards}"]
+                    for i in range(1, dist_shards + 1)
+                ]
+                download_logs(periodic_wf, test_log_list_rocm_distributed, folder_list[0])
+
+            # Download artifacts
+            test_artifacts_list_rocm_distributed = [
+                f"test-reports-test-distributed-{i}-{dist_shards}"
                 for i in range(1, dist_shards + 1)
             ]
-            download_logs(periodic_wf, test_log_list_rocm_distributed, folder_list[0])
-
-        # Download artifacts
-        test_artifacts_list_rocm_distributed = [
-            f"test-reports-test-distributed-{i}-{dist_shards}"
-            for i in range(1, dist_shards + 1)
-        ]
-        download_artifacts(
-            periodic_wf,
-            test_artifacts_list_rocm_distributed,
-            folder_list[2],
-            allowed_substrings=rocm_artifact_substrings,
-        )
-        os.chdir("..")
+            download_artifacts(
+                periodic_wf,
+                test_artifacts_list_rocm_distributed,
+                folder_list[2],
+                allowed_substrings=rocm_artifact_substrings,
+            )
+            os.chdir("..")
 
     # Download ROCm default rocm_wf when ROCm is enabled
     if not args.no_rocm and not args.exclude_default:
@@ -796,41 +889,40 @@ def main():
             default_fallback_used = True
             rocm_job_prefix['default'] = fallback_prefix
         if rocm_wf is None:
-            raise Exception(error_msg)
-        default_wf_name = ROCmWorkflowNames['default'] if not default_fallback_used else default_fallbacks[arch][0]
-        print(f"Using workflow '{default_wf_name}' with id:{rocm_wf['id']} for ROCm default{' (fallback)' if default_fallback_used else ''}")
+            _skip_missing_config("default", rocm_sha,
+                                 ROCmWorkflowNames['default'] + " and fallbacks")
+        else:
+            default_wf_name = ROCmWorkflowNames['default'] if not default_fallback_used else default_fallbacks[arch][0]
+            print(f"Using workflow '{default_wf_name}' with id:{rocm_wf['id']} for ROCm default{' (fallback)' if default_fallback_used else ''}")
 
-        folder_list = get_or_create_test_folder(rocm_wf)
+            folder_list = get_or_create_test_folder(rocm_wf)
 
-        # Download logs
-        # If logs aren't found you might want to check the HUD for the correct tags
-        # HUD link: https://hud.pytorch.org/hud/pytorch/pytorch/main/1?per_page=50&name_filter=rocm
-        if arch == "mi350":
-            default_shards = 6 if default_fallback_used else rocm_shards["default"]
-        else:
-            default_shards = rocm_shards["default"]
-        print(f"Using final ROCm shard count {default_shards} for default")
+            # Download logs
+            # If logs aren't found you might want to check the HUD for the correct tags
+            # HUD link: https://hud.pytorch.org/hud/pytorch/pytorch/main/1?per_page=50&name_filter=rocm
+            default_shards = derive_shard_count(rocm_wf, rocm_job_prefix['default'], "default", rocm_shards["default"])
+            print(f"Using final ROCm shard count {default_shards} for default")
 
-        if not args.artifacts_only:
-            test_log_list_rocm_default = [
-              [f"{current_prefix}rocm{i}.txt", f"{rocm_job_prefix['default']} / test (default, {i}, {default_shards}"]
+            if not args.artifacts_only:
+                test_log_list_rocm_default = [
+                  [f"{current_prefix}rocm{i}.txt", f"{rocm_job_prefix['default']} / test (default, {i}, {default_shards}"]
+                  for i in range(1, default_shards + 1)
+                ]
+                download_logs(rocm_wf, test_log_list_rocm_default, folder_list[0])
+
+            # Download artifacts
+            test_artifacts_list_rocm_default = [
+              f"test-reports-test-default-{i}-{default_shards}"
               for i in range(1, default_shards + 1)
             ]
-            download_logs(rocm_wf, test_log_list_rocm_default, folder_list[0])
-
-        # Download artifacts
-        test_artifacts_list_rocm_default = [
-          f"test-reports-test-default-{i}-{default_shards}"
-          for i in range(1, default_shards + 1)
-        ]
-        if not args.exclude_default:
-            download_artifacts(
-                rocm_wf,
-                test_artifacts_list_rocm_default,
-                test_folder=folder_list[2],
-                allowed_substrings=rocm_artifact_substrings,
-            )
-        os.chdir("..")
+            if not args.exclude_default:
+                download_artifacts(
+                    rocm_wf,
+                    test_artifacts_list_rocm_default,
+                    test_folder=folder_list[2],
+                    allowed_substrings=rocm_artifact_substrings,
+                )
+            os.chdir("..")
 
     # add new inductor workflow downloading for ROCm
     if not args.no_rocm and not args.exclude_inductor:
@@ -842,10 +934,18 @@ def main():
         print("===========================================")
         error_msg="Error: inductor workflow not found in scanned workflow runs. Try increasing max_pages."
         inductor_fallback_used = False
-        try:
-            inductor_wf_rocm = download_workflow_run(created=args.created, max_pages=args.max_pages, workflow=ROCmWorkflowNames["inductor"], sha=inductor_rocm_sha, ignore_status=args.ignore_status, status=status, error_msg=error_msg)
-        except (IndexError, Exception):
-            inductor_wf_rocm = None
+        if ROCmWorkflowNames["inductor"] == "trunk" and trunk_full_wf is not None:
+            # Read ROCm inductor from the canonical trunk push run (same as
+            # default/distributed above). Resolving a fresh trunk run here can
+            # land on the periodic rerun_disabled_tests/mem_leak_check trunk run,
+            # which only has the variant inductor jobs (a handful of flaky
+            # reruns) and makes the whole inductor suite look MISSED.
+            inductor_wf_rocm = trunk_full_wf
+        else:
+            try:
+                inductor_wf_rocm = download_workflow_run(created=args.created, max_pages=args.max_pages, workflow=ROCmWorkflowNames["inductor"], sha=inductor_rocm_sha, ignore_status=args.ignore_status, status=status, error_msg=error_msg)
+            except (IndexError, Exception):
+                inductor_wf_rocm = None
         inductor_fallbacks = _fallbacks_for("inductor")
         if inductor_wf_rocm is None and arch in inductor_fallbacks:
             fallback_wf, fallback_prefix = inductor_fallbacks[arch]
@@ -853,22 +953,23 @@ def main():
             inductor_wf_rocm = download_workflow_run(created=args.created, max_pages=args.max_pages, workflow=fallback_wf, sha=inductor_rocm_sha, ignore_status=args.ignore_status, status=status, error_msg=error_msg)
             inductor_fallback_used = True
         if inductor_wf_rocm is None:
-            # Inductor is optional and does not run for every SHA. Skip it
-            # instead of aborting the whole run, which would also drop the
-            # CUDA download below and publish an empty report.
-            print(f"WARNING: {error_msg} Skipping ROCm inductor for this SHA.")
+            # Inductor does not run for every SHA. Skip it instead of aborting
+            # the whole run (which would also drop the CUDA download below and
+            # publish an empty report); record it so the job is flagged failed.
+            _skip_missing_config("inductor", inductor_rocm_sha,
+                                 ROCmWorkflowNames['inductor'] + " and fallbacks")
         else:
             inductor_wf_name = ROCmWorkflowNames['inductor'] if not inductor_fallback_used else inductor_fallbacks[arch][0]
             print(f"Using workflow '{inductor_wf_name}' with id:{inductor_wf_rocm['id']} for ROCm inductor")
 
             folder_list = get_or_create_test_folder(inductor_wf_rocm)
 
-            inductor_shards = rocm_shards["inductor"]
-            print(f"Using final ROCm shard count {inductor_shards} for inductor")
             if inductor_fallback_used and arch in inductor_fallbacks:
                 inductor_job_prefix = inductor_fallbacks[arch][1]
             else:
                 inductor_job_prefix = rocm_job_prefix['inductor']
+            inductor_shards = derive_shard_count(inductor_wf_rocm, inductor_job_prefix, "inductor", rocm_shards["inductor"])
+            print(f"Using final ROCm shard count {inductor_shards} for inductor")
     
             # Download logs
             if not args.artifacts_only:
@@ -974,48 +1075,58 @@ def main():
             # find tests in inductor workflow with given sha and success status
             #https://docs.github.com/en/rest/actions/workflow-runs#list-workflow-runs-for-a-repository
             error_msg="Error: inductor workflow not found in scanned workflow runs. Try increasing max_pages."
-            inductor_wf_cuda = download_workflow_run(created=args.created, max_pages=args.max_pages, workflow=CUDAWorkflowNames["inductor"], sha=inductor_sha, ignore_status=args.ignore_status, status=status, error_msg=error_msg)
-            print(f"Using workflow '{CUDAWorkflowNames['inductor']}' with id:{inductor_wf_cuda['id']} for CUDA inductor")
-
-            inductor_cuda_jobs = get_workflow_jobs(inductor_wf_cuda)
-            cuda_inductor_test_job_kind, cuda_inductor_test_jobs = get_cuda_inductor_test_jobs(inductor_cuda_jobs)
-            # Same carried-over-on-rerun reasoning as CUDA default above: match
-            # artifacts by job ID across all attempts of the inductor run.
-            all_attempt_inductor_jobs = get_workflow_jobs(inductor_wf_cuda, all_attempts=True)
-            _, all_attempt_inductor_test_jobs = get_cuda_inductor_test_jobs(all_attempt_inductor_jobs)
-            cuda_inductor_job_ids = sorted({str(j['id']) for j in (cuda_inductor_test_jobs + all_attempt_inductor_test_jobs)})
-            cuda_inductor_artifact_substrings = (
-                [f"_{jid}" for jid in cuda_inductor_job_ids]
-                if cuda_inductor_job_ids
-                else None
-            )
-            print(f"Using CUDA inductor test job kind: {cuda_inductor_test_job_kind}")
-            print(f"Found {len(cuda_inductor_test_jobs)} CUDA inductor test jobs")
-
-            folder_list = get_or_create_test_folder(inductor_wf_cuda)
+            try:
+                inductor_wf_cuda = download_workflow_run(created=args.created, max_pages=args.max_pages, workflow=CUDAWorkflowNames["inductor"], sha=inductor_sha, ignore_status=args.ignore_status, status=status, error_msg=error_msg)
+            except (IndexError, Exception):
+                inductor_wf_cuda = None
+            if inductor_wf_cuda is None:
+                # CUDA inductor baseline does not run for every SHA (e.g. a PR with
+                # ciflow/trunk but not ciflow/inductor). Skip it rather than abort,
+                # and record it so the job is flagged failed.
+                _skip_missing_config("inductor (CUDA baseline)", inductor_sha,
+                                     CUDAWorkflowNames['inductor'])
+            else:
+                print(f"Using workflow '{CUDAWorkflowNames['inductor']}' with id:{inductor_wf_cuda['id']} for CUDA inductor")
+
+                inductor_cuda_jobs = get_workflow_jobs(inductor_wf_cuda)
+                cuda_inductor_test_job_kind, cuda_inductor_test_jobs = get_cuda_inductor_test_jobs(inductor_cuda_jobs)
+                # Same carried-over-on-rerun reasoning as CUDA default above: match
+                # artifacts by job ID across all attempts of the inductor run.
+                all_attempt_inductor_jobs = get_workflow_jobs(inductor_wf_cuda, all_attempts=True)
+                _, all_attempt_inductor_test_jobs = get_cuda_inductor_test_jobs(all_attempt_inductor_jobs)
+                cuda_inductor_job_ids = sorted({str(j['id']) for j in (cuda_inductor_test_jobs + all_attempt_inductor_test_jobs)})
+                cuda_inductor_artifact_substrings = (
+                    [f"_{jid}" for jid in cuda_inductor_job_ids]
+                    if cuda_inductor_job_ids
+                    else None
+                )
+                print(f"Using CUDA inductor test job kind: {cuda_inductor_test_job_kind}")
+                print(f"Found {len(cuda_inductor_test_jobs)} CUDA inductor test jobs")
 
-            cuda_inductor_prefix = CUDA_JOB_PREFIXES["inductor"]
-            cuda_inductor_shards = cuda_shards["inductor"]
+                folder_list = get_or_create_test_folder(inductor_wf_cuda)
 
-            # Download logs
-            if not args.artifacts_only:
-              test_log_list_cuda_inductor = [
-                [f"cuda_inductor{i}.txt", f"{cuda_inductor_prefix} / {cuda_inductor_test_job_kind} (inductor, {i}, {cuda_inductor_shards}"]
-                for i in range(1, cuda_inductor_shards + 1)
-              ]
-              download_logs(inductor_wf_cuda, test_log_list_cuda_inductor, folder_list[0], jobs=inductor_cuda_jobs)
+                cuda_inductor_prefix = CUDA_JOB_PREFIXES["inductor"]
+                cuda_inductor_shards = cuda_shards["inductor"]
 
-            test_artifacts_list_cuda_inductor = [
-              f"test-reports-{cuda_inductor_test_job_kind}-inductor-{i}-{cuda_inductor_shards}"
-              for i in range(1, cuda_inductor_shards + 1)
-            ]
-            download_artifacts(
-                inductor_wf_cuda,
-                test_artifacts_list_cuda_inductor,
-                test_folder=folder_list[1],
-                allowed_substrings=cuda_inductor_artifact_substrings,
-            )
-            os.chdir("..")
+                # Download logs
+                if not args.artifacts_only:
+                  test_log_list_cuda_inductor = [
+                    [f"cuda_inductor{i}.txt", f"{cuda_inductor_prefix} / {cuda_inductor_test_job_kind} (inductor, {i}, {cuda_inductor_shards}"]
+                    for i in range(1, cuda_inductor_shards + 1)
+                  ]
+                  download_logs(inductor_wf_cuda, test_log_list_cuda_inductor, folder_list[0], jobs=inductor_cuda_jobs)
+
+                test_artifacts_list_cuda_inductor = [
+                  f"test-reports-{cuda_inductor_test_job_kind}-inductor-{i}-{cuda_inductor_shards}"
+                  for i in range(1, cuda_inductor_shards + 1)
+                ]
+                download_artifacts(
+                    inductor_wf_cuda,
+                    test_artifacts_list_cuda_inductor,
+                    test_folder=folder_list[1],
+                    allowed_substrings=cuda_inductor_artifact_substrings,
+                )
+                os.chdir("..")
 
     # Download baseline commit artifacts for commit-vs-commit comparison
     if args.baseline_sha and not args.no_rocm:
diff --git a/.automation_scripts/pytorch-unit-test-scripts/generate_summary.py b/.automation_scripts/pytorch-unit-test-scripts/generate_summary.py
index 5c78317197b43..50a89674c6c43 100644
--- a/.automation_scripts/pytorch-unit-test-scripts/generate_summary.py
+++ b/.automation_scripts/pytorch-unit-test-scripts/generate_summary.py
@@ -114,10 +114,10 @@ def test_config_stats_keys(s1_name, s2_name, has_set2=True):
             f'TOTAL {s1}',
         ]
     return [
-        f'SKIPPED (on {s1_name}, but not on {s2_name})',
+        f'SKIPPED (on {s1_name}, PASSED on {s2_name})',
         f'SKIPPED (on {s1_name})',
         f'SKIPPED (on {s2_name})',
-        f'MISSED (MISSED on {s1_name}, NOT SKIPPED on {s2_name})',
+        f'MISSED (on {s1_name}, PASSED on {s2_name})',
         f'{s1}ONLY (PASSED on {s1}, NOT PASSED on {s2})',
         s2,
         s1,
@@ -141,15 +141,18 @@ def compute_test_config_stats(rows, s1_col, s2_col, s1_name, s2_name, has_set2=T
         vals[keys[4]] = sum(1 for r in rows if r[s1_col].strip())
         return vals
 
+    # Count a ROCm SKIPPED/MISSED test as a disagreement only when CUDA actually
+    # PASSED it. This excludes parametrization variants CUDA never runs (CUDA
+    # MISSED) or also skips, which aren't real ROCm-vs-CUDA coverage gaps.
     s1_skip_not_s2 = sum(
         1 for r in rows
-        if r[s1_col] == 'SKIPPED' and r[s2_col] != 'SKIPPED'
+        if r[s1_col] == 'SKIPPED' and r[s2_col] == 'PASSED'
     )
     s1_skip = sum(1 for r in rows if r[s1_col] == 'SKIPPED')
     s2_skip = sum(1 for r in rows if r[s2_col] == 'SKIPPED')
     s1_miss_not_s2_skip = sum(
         1 for r in rows
-        if r[s1_col] == 'MISSED' and r[s2_col] != 'SKIPPED'
+        if r[s1_col] == 'MISSED' and r[s2_col] == 'PASSED'
     )
     only_s1 = sum(
         1 for r in rows
@@ -233,11 +236,11 @@ def safe_float(v):
         wf_rows = [r for r in rows if r['test_config'] == wf]
         s1_skip_not_s2 = sum(
             1 for r in wf_rows
-            if r[s1_col] == 'SKIPPED' and r[s2_col] != 'SKIPPED'
+            if r[s1_col] == 'SKIPPED' and r[s2_col] == 'PASSED'
         )
         s1_miss_not_s2_skip = sum(
             1 for r in wf_rows
-            if r[s1_col] == 'MISSED' and r[s2_col] != 'SKIPPED'
+            if r[s1_col] == 'MISSED' and r[s2_col] == 'PASSED'
         )
         total_disagree += s1_skip_not_s2 + s1_miss_not_s2_skip
         total_s2 += sum(1 for r in wf_rows if r[s2_col].strip() and r[s2_col].strip() != 'MISSED')
@@ -639,6 +642,18 @@ def build_rows(args, archs, arch_data):
 
     if args.sha:
         out.append(('__header__', f'Commit SHA: {args.sha}'))
+        # Link straight to the upstream HUD page filtered (regex) to the trunk
+        # CUDA / inductor / rocm test jobs this report is built from, so a
+        # reviewer can jump from the summary to the matching CI jobs. Parens and
+        # pipes are percent-encoded to keep the markdown link valid.
+        hud_url = (
+            f'https://hud.pytorch.org/hud/pytorch/pytorch/{args.sha}/1'
+            '?per_page=50'
+            '&name_filter=%28trunk.*cuda%7Cinductor%7Crocm%29.*test.*'
+            '%28default%7Cdistributed%7Cinductor%29%2C'
+            '&useRegexFilter=true'
+        )
+        out.append(('__header__', f'HUD: [parity jobs for this commit]({hud_url})'))
     if args.pr_id:
         out.append(('__header__', f'PR ID: {args.pr_id}'))
 
diff --git a/.automation_scripts/pytorch-unit-test-scripts/parity_job_config.json b/.automation_scripts/pytorch-unit-test-scripts/parity_job_config.json
index 976fb2e8ae27f..3c8bb62621743 100644
--- a/.automation_scripts/pytorch-unit-test-scripts/parity_job_config.json
+++ b/.automation_scripts/pytorch-unit-test-scripts/parity_job_config.json
@@ -70,6 +70,13 @@
       "inductor": [{ "workflow": "rocm-nightly", "job_prefix": "linux-noble-rocm-nightly-py3.12-gfx942" }],
       "shard_counts": { "default": 6, "distributed": 3, "inductor": 2 },
       "checkrun_regex": "rocm-nightly.*/ test [(](default|distributed|inductor),"
+    },
+    "preview": {
+      "default": [{ "workflow": "rocm-preview", "job_prefix": "linux-noble-rocm-preview-py3.12-gfx942" }],
+      "distributed": [{ "workflow": "rocm-preview", "job_prefix": "linux-noble-rocm-preview-py3.12-gfx942" }],
+      "inductor": [{ "workflow": "rocm-preview", "job_prefix": "linux-noble-rocm-preview-py3.12-gfx942" }],
+      "shard_counts": { "default": 6, "distributed": 3, "inductor": 2 },
+      "checkrun_regex": "linux-noble-rocm-preview-py3.12-gfx942 / test [(](default|distributed|inductor),"
     }
   }
 }
diff --git a/.automation_scripts/pytorch-unit-test-scripts/summarize_xml_testreports.py b/.automation_scripts/pytorch-unit-test-scripts/summarize_xml_testreports.py
index 4349a90f21f80..c1a3fa3a37288 100755
--- a/.automation_scripts/pytorch-unit-test-scripts/summarize_xml_testreports.py
+++ b/.automation_scripts/pytorch-unit-test-scripts/summarize_xml_testreports.py
@@ -69,6 +69,41 @@ def _extract_shard(dirname):
         return f"{m.group(1)}/{m.group(2)}"
     return ""
 
+def _canonical_distributed_config(test_file, test_config):
+    """Fold the CUDA-only distributed test split introduced by pytorch#189232.
+
+    #189232 hived the single-GPU (non-multigpu) distributed tests out of the
+    'distributed' config and onto CUDA's cheaper 1-GPU 'default' runner, while
+    ROCm (and CPU) still run the whole distributed suite under 'distributed'.
+    Because we key on (test_file, ..., test_config), a CUDA-'default' result
+    then can't line up with the matching ROCm-'distributed' result, so the same
+    passing test double-counts as MISSED on both sides.
+
+    Map default -> distributed for distributed.* test files so the two stacks
+    match again. Scoped to distributed.* files only, so genuine default/inductor
+    results are left untouched (no risk of masking a real per-config
+    disagreement)."""
+    if test_file.startswith("distributed") and test_config == TestConfigName.default.name:
+        return TestConfigName.distributed.name
+    return test_config
+
+def _fold_distributed_config(test_cases):
+    """Apply _canonical_distributed_config to a parsed testcase dict.
+
+    The config lives in the last element of the (test_file, ..., test_config)
+    key. On collision (the same test seen under both configs) keep the
+    highest-priority status so a PASSED result isn't lost to a MISSED one."""
+    folded = {}
+    for key, case in test_cases.items():
+        new_config = _canonical_distributed_config(key[0], key[-1])
+        if new_config != key[-1]:
+            key = key[:-1] + (new_config,)
+            case["test_config"] = new_config
+        existing = folded.get(key)
+        if existing is None or _status_priority(case) > _status_priority(existing):
+            folded[key] = case
+    return folded
+
 def parse_xml_reports_as_dict(workflow_run_id, workflow_run_attempt, tag, path="."):
     test_config = ""
     test_cases = {}
@@ -242,6 +277,7 @@ def summarize_xml_files(args):
     # TODO: Does it matter what the workflow_run_attempt is set to below??
     # test_cases is dict of dicts, with keys as tuple of test_file, test_class, test_name and test_config
     test_cases_set1 = parse_xml_reports_as_dict(-1, -1, 'testcase', set1_path)
+    test_cases_set1 = _fold_distributed_config(test_cases_set1)
     for (k,v) in list(test_cases_set1.items()):
         if v['test_config'] == TestConfigName.default.name:
             ROCM_DEFAULT += 1
@@ -275,6 +311,7 @@ def summarize_xml_files(args):
               "set2 path not specified correctly, should be different from set1 path"
       test_cases_set2_running_time = parse_xml_reports_as_dict(-1, -1, 'testsuite', set2_path)
       test_cases_set2 = parse_xml_reports_as_dict(-1, -1, 'testcase', set2_path)
+      test_cases_set2 = _fold_distributed_config(test_cases_set2)
       for (k,v) in list(test_cases_set2.items()):
           if v['test_config'] == TestConfigName.default.name:
               CUDA_DEFAULT += 1
@@ -318,18 +355,29 @@ def summarize_xml_files(args):
     # test file level running time: ROCm and CUDA
     test_file_level_ROCm: Dict[Tuple[str], float] = {}
     test_file_level_CUDA: Dict[Tuple[str], float] = {}
+    test_file_shards_ROCm: Dict[Tuple[str], set] = {}
+    test_file_shards_CUDA: Dict[Tuple[str], set] = {}
     for (k,v) in list(test_cases_set1_running_time.items()):
           test_file_name = k[0]
-          test_config_name = k[2]
+          # Fold the distributed split (pytorch#189232) so per-file times line
+          # up with the already-folded per-file counts; otherwise distributed.*
+          # files show a phantom 'default' row with import time but 0 tests.
+          test_config_name = _canonical_distributed_config(test_file_name, k[2])
           tar_tup_rocm = (test_file_name, test_config_name,)
+          test_file_shards_ROCm.setdefault(tar_tup_rocm, set())
+          if v.get("shard"):
+              test_file_shards_ROCm[tar_tup_rocm].add(v["shard"])
           if test_file_level_ROCm.get(tar_tup_rocm) == None:
               test_file_level_ROCm[ ( test_file_name, test_config_name ) ] = v["running_time_xml"]
           else:
               test_file_level_ROCm[ ( test_file_name, test_config_name ) ] += v["running_time_xml"]
     for (k,v) in list(test_cases_set2_running_time.items()):
           test_file_name = k[0]
-          test_config_name = k[2]
+          test_config_name = _canonical_distributed_config(test_file_name, k[2])
           tar_tup_cuda = (test_file_name, test_config_name)
+          test_file_shards_CUDA.setdefault(tar_tup_cuda, set())
+          if v.get("shard"):
+              test_file_shards_CUDA[tar_tup_cuda].add(v["shard"])
           if test_file_level_CUDA.get(tar_tup_cuda) == None:
               test_file_level_CUDA[ ( test_file_name, test_config_name ) ] = v["running_time_xml"]
           else:
@@ -615,24 +663,35 @@ def sorting_key(e):
 
     # write test file running time to file
     test_file_running_time_for_csv = {}
+    set1_running_time_col = f"{set1_name}_running_time"
+    set2_running_time_col = f"{set2_name}_running_time"
+    set1_tests_run_col = f"{set1_name}_tests_run"
+    set2_tests_run_col = f"{set2_name}_tests_run"
+    set1_test_shards_col = f"{set1_name}_test_shards"
+    set2_test_shards_col = f"{set2_name}_test_shards"
+    set1_passed_col = f"{set1_name}_passed"
+    set1_skipped_col = f"{set1_name}_skipped"
+    set1_missed_col = f"{set1_name}_missed"
     for key_rocm in test_file_level_ROCm.keys():
         item_values = {}
         item_values["test_file"] = key_rocm[0]
         item_values["test_config"] = key_rocm[1]
-        item_values["rocm_running_time"] = test_file_level_ROCm[key_rocm]
-        item_values["cuda_running_time"] = 0.0
+        item_values[set1_running_time_col] = test_file_level_ROCm[key_rocm]
+        item_values[set2_running_time_col] = 0.0
         if key_rocm in test_file_level_CUDA.keys():
-            item_values["cuda_running_time"] = test_file_level_CUDA[key_rocm]
-        item_values["abs_time_diff"] = item_values["rocm_running_time"] - item_values["cuda_running_time"]
+            item_values[set2_running_time_col] = test_file_level_CUDA[key_rocm]
+        item_values["abs_time_diff"] = item_values[set1_running_time_col] - item_values[set2_running_time_col]
         item_values["relative_time_diff"] = 0.0
-        if item_values["cuda_running_time"] != 0.0:
-            item_values["relative_time_diff"] = 100 * (item_values["rocm_running_time"] - item_values["cuda_running_time"]) / item_values["cuda_running_time"]
+        if item_values[set2_running_time_col] != 0.0:
+            item_values["relative_time_diff"] = 100 * (item_values[set1_running_time_col] - item_values[set2_running_time_col]) / item_values[set2_running_time_col]
         # Add test counts
-        item_values["rocm_tests_run"] = test_file_counts_ROCm.get(key_rocm, {}).get('tests_run', 0)
-        item_values["cuda_tests_run"] = test_file_counts_CUDA.get(key_rocm, 0)
-        item_values["rocm_passed"] = test_file_counts_ROCm.get(key_rocm, {}).get('passed', 0)
-        item_values["rocm_skipped"] = test_file_counts_ROCm.get(key_rocm, {}).get('skipped', 0)
-        item_values["rocm_missed"] = test_file_counts_ROCm.get(key_rocm, {}).get('missed', 0)
+        item_values[set1_tests_run_col] = test_file_counts_ROCm.get(key_rocm, {}).get('tests_run', 0)
+        item_values[set2_tests_run_col] = test_file_counts_CUDA.get(key_rocm, 0)
+        item_values[set1_test_shards_col] = len(test_file_shards_ROCm.get(key_rocm, set()))
+        item_values[set2_test_shards_col] = len(test_file_shards_CUDA.get(key_rocm, set()))
+        item_values[set1_passed_col] = test_file_counts_ROCm.get(key_rocm, {}).get('passed', 0)
+        item_values[set1_skipped_col] = test_file_counts_ROCm.get(key_rocm, {}).get('skipped', 0)
+        item_values[set1_missed_col] = test_file_counts_ROCm.get(key_rocm, {}).get('missed', 0)
         test_file_running_time_for_csv[key_rocm] = item_values
 
     for key_cuda in test_file_level_CUDA.keys():
@@ -640,18 +699,20 @@ def sorting_key(e):
             item_values = {}
             item_values["test_file"] = key_cuda[0]
             item_values["test_config"] = key_cuda[1]
-            item_values["rocm_running_time"] = 0.0
-            item_values["cuda_running_time"] = test_file_level_CUDA[key_cuda]
-            item_values["abs_time_diff"] = item_values["rocm_running_time"] - item_values["cuda_running_time"]
+            item_values[set1_running_time_col] = 0.0
+            item_values[set2_running_time_col] = test_file_level_CUDA[key_cuda]
+            item_values["abs_time_diff"] = item_values[set1_running_time_col] - item_values[set2_running_time_col]
             item_values["relative_time_diff"] = 0.0
-            if item_values["cuda_running_time"] != 0.0:
-                item_values["relative_time_diff"] = 100 * (item_values["rocm_running_time"] - item_values["cuda_running_time"]) / item_values["cuda_running_time"]
+            if item_values[set2_running_time_col] != 0.0:
+                item_values["relative_time_diff"] = 100 * (item_values[set1_running_time_col] - item_values[set2_running_time_col]) / item_values[set2_running_time_col]
             # Add test counts
-            item_values["rocm_tests_run"] = test_file_counts_ROCm.get(key_cuda, {}).get('tests_run', 0)
-            item_values["cuda_tests_run"] = test_file_counts_CUDA.get(key_cuda, 0)
-            item_values["rocm_passed"] = test_file_counts_ROCm.get(key_cuda, {}).get('passed', 0)
-            item_values["rocm_skipped"] = test_file_counts_ROCm.get(key_cuda, {}).get('skipped', 0)
-            item_values["rocm_missed"] = test_file_counts_ROCm.get(key_cuda, {}).get('missed', 0)
+            item_values[set1_tests_run_col] = test_file_counts_ROCm.get(key_cuda, {}).get('tests_run', 0)
+            item_values[set2_tests_run_col] = test_file_counts_CUDA.get(key_cuda, 0)
+            item_values[set1_test_shards_col] = len(test_file_shards_ROCm.get(key_cuda, set()))
+            item_values[set2_test_shards_col] = len(test_file_shards_CUDA.get(key_cuda, set()))
+            item_values[set1_passed_col] = test_file_counts_ROCm.get(key_cuda, {}).get('passed', 0)
+            item_values[set1_skipped_col] = test_file_counts_ROCm.get(key_cuda, {}).get('skipped', 0)
+            item_values[set1_missed_col] = test_file_counts_ROCm.get(key_cuda, {}).get('missed', 0)
             test_file_running_time_for_csv[key_cuda] = item_values
 
     test_file_running_time_for_csv = dict(sorted(test_file_running_time_for_csv.items()))
@@ -661,24 +722,28 @@ def sorting_key_running_time(e):
           return 0
         elif e == "test_config":
           return 1
-        elif e == "rocm_running_time":
+        elif e == set1_running_time_col:
           return 2
-        elif e == "cuda_running_time":
+        elif e == set2_running_time_col:
           return 3
         elif e == "abs_time_diff":
           return 4
         elif e == "relative_time_diff":
           return 5
-        elif e == "rocm_tests_run":
+        elif e == set1_tests_run_col:
           return 6
-        elif e == "cuda_tests_run":
+        elif e == set2_tests_run_col:
           return 7
-        elif e == "rocm_passed":
+        elif e == set1_test_shards_col:
           return 8
-        elif e == "rocm_skipped":
+        elif e == set2_test_shards_col:
           return 9
-        elif e == "rocm_missed":
+        elif e == set1_passed_col:
           return 10
+        elif e == set1_skipped_col:
+          return 11
+        elif e == set1_missed_col:
+          return 12
         else:
           return 100
 
@@ -690,20 +755,25 @@ def sorting_key_running_time(e):
          writer.writerows(test_file_running_time_for_csv.values())
 
     # print summary
+    # User-facing labels follow the same set1/set2 naming as the CSV columns
+    # (set1_name/set2_name default to rocm/cuda, or are commit SHAs in
+    # commit-vs-commit mode) so the printed summary stays consistent with them.
+    set1_disp = set1_name.upper()
+    set2_disp = set2_name.upper()
     print( " " )
     print( "_____________________________________" )
     print( "Test-results" )
     print( " " )
     print( "=====Single GPU Number=====" )
-    print( "SKIPPED_DEFAULT, MISSED_DEFAULT, ROCMONLY_DEFAULT, CUDA_DEFAULT, ROCM_DEFAULT" )
+    print( f"SKIPPED_DEFAULT, MISSED_DEFAULT, {set1_disp}ONLY_DEFAULT, {set2_disp}_DEFAULT, {set1_disp}_DEFAULT" )
     print( str(SKIPPED_DEFAULT) + ", " + str(MISSED_DEFAULT) + ", " + str(ROCMONLY_DEFAULT) + ", " + str(CUDA_DEFAULT) + ", " + str(ROCM_DEFAULT) )
     print( " " )
     print( "=====Distributed GPU Number=====" )
-    print( "SKIPPED_DISTRIBUTED, MISSED_DISTRIBUTED, ROCMONLY_DISTRIBUTED, CUDA_DISTRIBUTED, ROCM_DISTRIBUTED" )
+    print( f"SKIPPED_DISTRIBUTED, MISSED_DISTRIBUTED, {set1_disp}ONLY_DISTRIBUTED, {set2_disp}_DISTRIBUTED, {set1_disp}_DISTRIBUTED" )
     print( str(SKIPPED_DISTRIBUTED) + ", " + str(MISSED_DISTRIBUTED) + ", " + str(ROCMONLY_DISTRIBUTED) + ", " + str(CUDA_DISTRIBUTED) + ", " + str(ROCM_DISTRIBUTED) )
     print( " " )
     print( "=====Inductor GPU Number=====" )
-    print( "SKIPPED_INDUCTOR, MISSED_INDUCTOR, ROCMONLY_INDUCTOR, CUDA_INDUCTOR, ROCM_INDUCTOR" )
+    print( f"SKIPPED_INDUCTOR, MISSED_INDUCTOR, {set1_disp}ONLY_INDUCTOR, {set2_disp}_INDUCTOR, {set1_disp}_INDUCTOR" )
     print( str(SKIPPED_INDUCTOR) + ", " + str(MISSED_INDUCTOR) + ", " + str(ROCMONLY_INDUCTOR) + ", " + str(CUDA_INDUCTOR) + ", " + str(ROCM_INDUCTOR) )
     print( " " )
     print( "SELECTED CAUSES SUMMARY" )
@@ -728,7 +798,7 @@ def sorting_key_running_time(e):
     print( " " )
     print( "=====================" )
     print( "Time statistics" )
-    print( "ROCM_RUNNING_TIME, CUDA_RUNNING_TIME" )
+    print( f"{set1_disp}_RUNNING_TIME, {set2_disp}_RUNNING_TIME" )
     print( str(TOTAL_ROCM_RUNNING_TIME) + ", " + str(TOTAL_CUDA_RUNNING_TIME) )
     #print( "ROCm test file level time statistics" )
     #for (k,v) in list(test_file_level_ROCm.items()):
diff --git a/.github/workflows/parity-auto.yml b/.github/workflows/parity-auto.yml
index 06eb450f9bdbf..129c0a5e01f5f 100644
--- a/.github/workflows/parity-auto.yml
+++ b/.github/workflows/parity-auto.yml
@@ -1,10 +1,16 @@
 name: Parity Auto Trigger
 run-name: "Parity auto-trigger · pytorch/pytorch main"
 
-# Every 10 min, dispatch parity.yml once per completed upstream trunk.yml push
-# whose parity inputs have all finished. Scope is trunk only: the mi350 ROCm
-# shards that ride along in trunk.yml vs that run's CUDA shards. Other arches
-# have their own periodic workflows - run parity.yml manually for those.
+# Every 10 min, dispatch parity.yml once per upstream SHA whose parity inputs
+# have all finished. Candidate SHAs come from two sources: completed trunk.yml
+# pushes (mi350 rides along there) and the scheduled per-arch workflows for
+# mi300/mi200/navi31 (which run on their own SHAs at their own cadence). A SHA
+# that carries several arches yields ONE combined parity report (parity.yml's
+# matrix emits a per-arch artifact plus a merged summary).
+#
+# Because the scheduled arches lag trunk, we hold back SHAs newer than the
+# newest scheduled run so a late mi300/mi200 batch can still join that SHA's
+# report instead of producing a premature trunk-only one.
 #
 # Readiness is gated per check-run, not on workflow_run conclusion: one failed
 # shard flips the parent run to failure while siblings are still going, so we
@@ -76,9 +82,11 @@ jobs:
           MAX_COMMITS: ${{ github.event_name == 'pull_request' && '20' || inputs.max_commits || '200' }}
           MAX_DISPATCHES: ${{ github.event_name == 'pull_request' && '5' || inputs.max_dispatches || '50' }}
           MAX_AGE_HOURS: ${{ inputs.max_age_hours || '72' }}
-          # Auto-parity is trunk-scoped: mi350 is the only ROCm arch that rides
-          # along in trunk.yml. Other arches have their own periodic workflows.
-          ARCHS_IN: mi350
+          # mi350 rides along in trunk.yml (per push); mi300/mi200/navi31 run in
+          # their own scheduled workflows on their own SHAs. We scan both sources
+          # (see fetch_candidate_commits) so a SHA that carries several arches
+          # yields one combined parity report once every arch's run has finished.
+          ARCHS_IN: mi350 mi300 mi200 navi31
           # Optional manual overrides; blank means "derive from parity_job_config.json".
           ARCH_JOBNAME_REGEX_OVERRIDE: ${{ inputs.arch_jobname_regex_map || '' }}
           ARCH_WORKFLOW_REGEX_OVERRIDE: ${{ inputs.arch_workflow_regex_map || '' }}
@@ -117,6 +125,15 @@ jobs:
                 | "(^|/)(" + join("|") + ")[.]yml$"
               )')}
             CUDA_JOBNAME_REGEX=$(echo "$config_json" | jq -r '.cuda.checkrun_regex')
+            # Upstream workflow file names for the non-trunk arches (mi350 rides
+            # trunk; nightly is not auto-scanned). Their scheduled runs land on
+            # their own SHAs, so fetch_scheduled_commits mines those SHAs as
+            # extra candidates on top of the trunk pushes.
+            SCHEDULED_WORKFLOWS=$(echo "$config_json" | jq -r '
+              [ .rocm | to_entries[]
+                | select(.key != "mi350" and .key != "nightly")
+                | (.value.default[]?, .value.distributed[]?, .value.inductor[]?).workflow ]
+              | unique | .[]')
           }
 
           print_run_config() {
@@ -124,6 +141,7 @@ jobs:
               "Upstream:        $UPSTREAM@$BRANCH" \
               "Target ref:      $TARGET_REF" \
               "Scope archs:     $ARCHS" \
+              "Scheduled wfs:   $(echo "$SCHEDULED_WORKFLOWS" | tr '\n' ' ')" \
               "Max trunk runs:  $MAX_COMMITS" \
               "Max dispatches:  $MAX_DISPATCHES" \
               "Max age:         ${MAX_AGE_HOURS}h" \
@@ -166,6 +184,33 @@ jobs:
             echo "$commits_json" | jq -r '.[] | "\(.head_sha) \(.created_at)"'
           }
 
+          # Echo "<sha> <created_at>" lines for recent completed runs of the
+          # non-trunk arch workflows (mi300/mi200/navi31). These arches run on a
+          # schedule against whatever main HEAD was current then, so their runs
+          # surface SHAs that the trunk-push scan alone would rank too low to
+          # reach. 404/stale workflows simply yield nothing.
+          fetch_scheduled_commits() {
+            local wf rows
+            for wf in $SCHEDULED_WORKFLOWS; do
+              rows=$(gh api \
+                "repos/$UPSTREAM/actions/workflows/$wf.yml/runs?branch=$BRANCH&status=completed&per_page=30" \
+                --jq '.workflow_runs[] | "\(.head_sha) \(.created_at)"' 2>/dev/null)
+              [ -n "$rows" ] && printf '%s\n' "$rows"
+            done
+          }
+
+          # Merge trunk pushes + the already-fetched scheduled candidates (global
+          # SCHEDULED_COMMITS) into one newest-first "<sha> <created_at>" list, one
+          # row per SHA, capped at MAX_COMMITS. Called in a subshell via $(), so it
+          # only reads globals - main computes NEWEST_SCHEDULED_EPOCH itself.
+          fetch_candidate_commits() {
+            # $1 ~ 40-hex guard drops any stray non-row output (e.g. a 404 body
+            # from a missing scheduled workflow) that slipped past the API calls.
+            { fetch_trunk_commits; printf '%s\n' "$SCHEDULED_COMMITS"; } \
+              | awk 'NF>=2 && $1 ~ /^[0-9a-f]{40}$/' \
+              | sort -k2,2r | awk '!seen[$1]++' | head -n "$MAX_COMMITS"
+          }
+
           # Echo a JSON array of recent auto-parity workflow_dispatch runs in our
           # repo, used to skip SHAs we already dispatched. Auto runs come from
           # github-actions[bot] and carry an "autoparity-" run-name prefix; we
@@ -261,6 +306,18 @@ jobs:
               return 1
             fi
 
+            # Hold back SHAs newer than the newest scheduled arch run: the schedule
+            # has not reached them yet, so a mi300/mi200/navi batch may still land
+            # on this SHA. Waiting lets us emit ONE combined report per SHA once
+            # every arch that will run has finished, rather than a premature
+            # trunk-only report that a later batch can no longer join. (When no
+            # scheduled runs exist NEWEST_SCHEDULED_EPOCH == now, so nothing is
+            # held and mi350/trunk behaves exactly as before.)
+            if [ "$commit_epoch" -ne 0 ] && [ "$commit_epoch" -gt "$NEWEST_SCHEDULED_EPOCH" ]; then
+              echo "[$short] $date  newer than newest scheduled arch run - waiting for scheduled arches to catch up"
+              return 0
+            fi
+
             if sha_already_dispatched "$sha"; then
               echo "[$short] parity report already exists for this SHA - skip"
               return 0
@@ -300,6 +357,31 @@ jobs:
               return 0
             fi
 
+            # Per-config CUDA baseline presence. A config whose CUDA test jobs did
+            # not run on this SHA (e.g. a failed trunk run that never launched CUDA
+            # default) has no baseline to compare against, so exclude it from the
+            # dispatch instead of emitting a bogus all-MISSED column. The length
+            # check above guarantees at least one config still has a baseline.
+            EXCLUDE_FLAGS=""
+            local cfg cuda_missing=""
+            for cfg in default distributed inductor; do
+              if ! echo "$cuda_check_runs" | jq -e --arg c "$cfg" \
+                   'any(.[]; .name | test("[(]" + $c + ","))' >/dev/null; then
+                EXCLUDE_FLAGS="$EXCLUDE_FLAGS -f exclude_${cfg}=true"
+                cuda_missing="$cuda_missing $cfg"
+              fi
+            done
+            cuda_missing=$(echo "$cuda_missing" | xargs)
+            if [ -n "$cuda_missing" ]; then
+              # Drop the excluded configs' ROCm check-runs from the completion gate
+              # so a missing-CUDA config never makes us wait on its ROCm shards.
+              local miss_rx
+              miss_rx=$(echo "$cuda_missing" | tr ' ' '|')
+              ROCM_CHECK_RUNS=$(echo "$ROCM_CHECK_RUNS" | jq --arg rx "[(]($miss_rx)," \
+                '[.[] | select((.name | test($rx)) | not)]')
+              echo "[$short] $date  no CUDA baseline for: $cuda_missing - excluding from report"
+            fi
+
             # Gate 1: every check-run the report consumes (ROCm shards for arches
             # that ran + CUDA tests) must be status=completed - we author the
             # SHA's report on dispatch, so dispatching early = partial data.
@@ -328,22 +410,24 @@ jobs:
             fi
 
             arch_dispatch=$(echo "$READY" | sed 's/ /, /g')
-            echo "[$short] READY archs: '$(echo "$READY" | tr ' ' ',')' (committed $date; not-run: ${NOT_RUN_NOTES:-none})"
-            echo "[$short] dispatching for: '$(echo "$READY" | tr ' ' ',')'"
+            echo "[$short] READY archs: '$(echo "$READY" | tr ' ' ',')' (committed $date; not-run: ${NOT_RUN_NOTES:-none}; excluded-configs: ${cuda_missing:-none})"
+            echo "[$short] dispatching for: '$(echo "$READY" | tr ' ' ',')'${EXCLUDE_FLAGS:+ (excluding${cuda_missing:+ }$cuda_missing)}"
 
             if [ "$DRY_RUN" = "true" ]; then
               echo "[$short] DRY_RUN=true - not dispatching"
             else
+              # $EXCLUDE_FLAGS is intentionally unquoted so its -f pairs word-split.
               gh workflow run parity.yml \
                 --repo "$GITHUB_REPOSITORY" \
                 --ref  "$TARGET_REF" \
                 -f sha="$sha" \
                 -f arch="$arch_dispatch" \
-                -f auto_triggered=true
+                -f auto_triggered=true \
+                $EXCLUDE_FLAGS
             fi
 
             DISPATCHED_COUNT=$((DISPATCHED_COUNT + 1))
-            DISPATCHED_SUMMARY="${DISPATCHED_SUMMARY}${short}:${arch_dispatch}"$'\n'
+            DISPATCHED_SUMMARY="${DISPATCHED_SUMMARY}${short}:${arch_dispatch}${cuda_missing:+ (no-cuda:$(echo "$cuda_missing" | tr ' ' ','))}"$'\n'
             if [ "$DISPATCHED_COUNT" -ge "$MAX_DISPATCHES" ]; then
               echo "Reached max dispatches for this scan ($MAX_DISPATCHES); stopping"
               return 1
@@ -388,10 +472,23 @@ jobs:
             load_matching_config
             print_run_config
 
+            # Fetch scheduled-arch candidates once (global, read by
+            # fetch_candidate_commits) and record the newest scheduled run time so
+            # process_commit can hold back not-yet-reached SHAs. No scheduled runs
+            # => hold nothing (NEWEST_SCHEDULED_EPOCH = now).
+            SCHEDULED_COMMITS=$(fetch_scheduled_commits)
+            local newest
+            newest=$(printf '%s\n' "$SCHEDULED_COMMITS" | awk 'NF>=2 && $1 ~ /^[0-9a-f]{40}$/{print $2}' | sort -r | head -1)
+            if [ -n "$newest" ]; then
+              NEWEST_SCHEDULED_EPOCH=$(date -u -d "$newest" +%s 2>/dev/null || echo "$NOW_EPOCH")
+            else
+              NEWEST_SCHEDULED_EPOCH=$NOW_EPOCH
+            fi
+
             local commits
-            commits=$(fetch_trunk_commits)
+            commits=$(fetch_candidate_commits)
             if [ -z "$commits" ]; then
-              echo "::warning::No completed trunk.yml push runs returned from $UPSTREAM@$BRANCH"
+              echo "::warning::No completed trunk/scheduled workflow runs returned from $UPSTREAM@$BRANCH"
               exit 0
             fi
 
diff --git a/.github/workflows/parity.yml b/.github/workflows/parity.yml
index fc53323a25ee0..e85ec3e4dc067 100644
--- a/.github/workflows/parity.yml
+++ b/.github/workflows/parity.yml
@@ -31,7 +31,7 @@ on:
         required: false
         type: string
       arch:
-        description: 'ROCm architectures, comma or space separated. Options: mi350, mi300, mi200, nightly, navi31. Example: "nightly, mi350" or "mi300"'
+        description: 'ROCm architectures, comma or space separated. Options: mi350, mi300, mi200, nightly, navi31, preview. Example: "nightly, mi350" or "preview"'
         required: false
         default: 'mi350, mi300, mi200'
         type: string
@@ -98,7 +98,7 @@ on:
       auto_classify:
         description: 'Auto-classify skip reasons for SKIPPED/MISSED tests in the output CSV'
         required: false
-        default: false
+        default: true
         type: boolean
 
 jobs:
@@ -154,6 +154,7 @@ jobs:
         run: pip install -r requirements.txt
 
       - name: Download artifacts
+        id: download
         working-directory: .automation_scripts/pytorch-unit-test-scripts
         env:
           GITHUB_TOKEN: ${{ secrets.PARITY_GITHUB_TOKEN || github.token }}
@@ -200,7 +201,20 @@ jobs:
           fi
 
           echo "Running: python3 ./download_testlogs $ARGS"
+          # Don't abort the job here on a partial download (e.g. a config whose
+          # upstream job was skipped -- ROCm distributed during GPU capacity
+          # crunches). Capture download_testlogs' real exit code, keep going so
+          # the report still generates and uploads from whatever downloaded,
+          # then fail the job at the end (see "Flag incomplete download") so the
+          # run is still clearly marked failed.
+          set +e
           python3 ./download_testlogs $ARGS 2>&1 | tee download_${{ matrix.arch }}.log
+          DL_RC=${PIPESTATUS[0]}
+          set -e
+          echo "download_rc=$DL_RC" >> "$GITHUB_OUTPUT"
+          if [ "$DL_RC" -ne 0 ]; then
+            echo "::warning::download_testlogs exited $DL_RC for ${{ matrix.arch }}; building a partial report, the job will be failed at the end."
+          fi
 
       - name: Identify output folder
         id: folder
@@ -284,6 +298,7 @@ jobs:
 
       - name: Collect upload paths
         id: upload-paths
+        if: ${{ always() && steps.folder.outputs.folder != '' }}
         run: |
           FOLDER=".automation_scripts/pytorch-unit-test-scripts/${{ steps.folder.outputs.folder }}"
           PATHS="${FOLDER}/*.csv
@@ -302,11 +317,22 @@ jobs:
           echo "EOF" >> "$GITHUB_OUTPUT"
 
       - name: Upload artifacts
+        if: ${{ always() && steps.upload-paths.outputs.paths != '' }}
         uses: actions/upload-artifact@v4
         with:
           name: ${{ needs.setup-matrix.outputs.prefix }}-results-${{ matrix.arch }}
           path: ${{ steps.upload-paths.outputs.paths }}
 
+      - name: Flag incomplete download
+        # A partial download (some config's upstream jobs missing) still produces
+        # and uploads a report above; fail the job here so the run is clearly
+        # marked failed for that arch without losing the report.
+        if: ${{ always() && steps.download.outputs.download_rc != '0' && steps.download.outputs.download_rc != '' }}
+        run: |
+          echo "download_testlogs reported an incomplete download (rc=${{ steps.download.outputs.download_rc }}) for ${{ matrix.arch }}."
+          echo "A partial report was still generated and uploaded; failing the job to flag the incomplete download."
+          exit 1
+
   summarize:
     needs: [setup-matrix, generate-parity]
     if: ${{ !cancelled() }}