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..6c6b6f62561f6 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",
@@ -110,6 +119,31 @@
      "file": r"^test_nn$",
      "msg": r"skipIfRocm.*doesn't currently work"},
 
+    # --- Linalg: hipSOLVER xgeev requires ROCm >= 7.14 (linalg.eig via xgeev, pytorch#188720) ---
+    {"reason": "Linalg",
+     "msg": r"hipSOLVER xgeev"},
+    # --- Linalg: ROCm 6.4 regression on linalg ops (householder/decomp) ---
+    {"reason": "Linalg",
+     "msg": r"regression in ROCm 6\.4"},
+
+    # --- Profiler: CUPTI-dependent profiler tests (libcupti / cupti-python) ---
+    {"reason": "Profiler",
+     "msg": r"(?i)cupti"},
+
+    # --- block_table: ROCm does not support paged-KV block_table (varlen attention) ---
+    {"reason": "block_table",
+     "msg": r"ROCm does not support block_table"},
+
+    # --- explicit NVIDIA test: CUDA / SM-gated tests skipped on ROCm ---
+    {"reason": "explicit NVIDIA test",
+     "msg": r"only supported on NVIDIA CUDA"},
+    {"reason": "explicit NVIDIA test",
+     "msg": r"^CUDA-only$"},
+    {"reason": "explicit NVIDIA test",
+     "msg": r"requires CUDA SM80"},
+    {"reason": "explicit NVIDIA test",
+     "msg": r"CUTLASS.*CUDA-only"},
+
     # --- hipSolver/Magma: skipCUDAIfRocm in test_ops for ldl_solve, scaled_dot_product, conv_transpose3d ---
     {"reason": "hipSolver/Magma",
      "msg": r"skipCUDAIfRocm.*doesn't currently work",
@@ -433,6 +467,11 @@
     {"reason": "Misc",
      "msg": r"test skipped on \('gfx"},
 
+    # Misc: generic skipIfRocm pointing at a tracked GitHub issue (no specific
+    # category; heterogeneous inductor/varlen tests)
+    {"reason": "Misc",
+     "msg": r"skipIfRocm: https?://"},
+
     # cuFFT-specific
     {"reason": "Misc",
      "msg": r"cuFFT-specific"},
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..42d69152e243f 100755
--- a/.automation_scripts/pytorch-unit-test-scripts/summarize_xml_testreports.py
+++ b/.automation_scripts/pytorch-unit-test-scripts/summarize_xml_testreports.py
@@ -318,10 +318,15 @@ 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]
           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:
@@ -330,6 +335,9 @@ def summarize_xml_files(args):
           test_file_name = k[0]
           test_config_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 +623,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 +659,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 +682,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 +715,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 +758,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/.automation_scripts/rocm_ci_debug_env/cli_common.py b/.automation_scripts/rocm_ci_debug_env/cli_common.py
new file mode 100644
index 0000000000000..412cecba6d671
--- /dev/null
+++ b/.automation_scripts/rocm_ci_debug_env/cli_common.py
@@ -0,0 +1,14 @@
+"""Shared CLI helpers for the ROCm CI debug-env scripts."""
+
+from __future__ import annotations
+
+import sys
+
+
+class CliError(Exception):
+    """Expected failure with a user-facing message."""
+
+
+def cli_fail(message: str) -> int:
+    print(f"error: {message}", file=sys.stderr)
+    return 1
diff --git a/.automation_scripts/rocm_ci_debug_env/pytorch_rocm_ci_debug_env.py b/.automation_scripts/rocm_ci_debug_env/pytorch_rocm_ci_debug_env.py
new file mode 100644
index 0000000000000..44d1eaffcc888
--- /dev/null
+++ b/.automation_scripts/rocm_ci_debug_env/pytorch_rocm_ci_debug_env.py
@@ -0,0 +1,1416 @@
+#!/usr/bin/env python3
+"""
+Set up a local ROCm CI debugging environment: clone PyTorch at a commit, parse ROCm workflow
+YAML, resolve GHCR image refs, download GitHub Actions build artifacts, ``docker pull`` the CI
+image, then ``docker run`` + ``docker exec`` to install the wheel inside the container.
+
+The container is named ``<user>_pytorch_rocm_ci_debug_env_<short-sha>_<workflow>``, where
+``<user>`` is the launching user's login name (``getpass.getuser()``), ``<short-sha>`` is the
+first 12 chars of the resolved head sha, and ``<workflow>`` is the workflow stem. Because the
+name is deterministic, re-running for the same commit+workflow fails fast if that container
+still exists (remove it with ``docker rm -f <name>``).
+
+Forking: change ``REPO_OWNER`` / ``REPO_NAME`` below (clone URL, GitHub API, S3 paths follow).
+
+Repo root is ``/var/lib/jenkins/pytorch`` in every mode (wheel at ``dist/``, tests at ``test/``).
+This is a subdir of the image's jenkins HOME (``/var/lib/jenkins``), so the image workspace (pip/
+sccache caches, dotfiles, etc.) is never shadowed. A ``/workspace/pytorch`` symlink pointing at the
+repo root is also created inside the container for convenience.
+
+Acquisition modes (how the PyTorch checkout reaches the container; pick at most one):
+* Default (no flag): do NO host clone and NO bind mount. The workflow metadata that the
+  host-mapped modes derive from a local checkout is instead fetched from GitHub for the commit
+  (workflow YAML, ``.ci/docker`` tree sha, and full head sha) via the contents/commits APIs. The
+  image is pulled/detected as usual, then the container is started without a bind mount and ALL
+  remaining work happens inside it via ``docker exec``: ``git clone`` PyTorch to
+  ``/var/lib/jenkins/pytorch`` and ``git checkout --detach`` the commit, ``curl`` the public S3
+  ``artifacts.zip``, unzip, ``pip install`` the wheel, and end at the test dir.
+* ``--host-mapped``: clone PyTorch to a host temp directory outside this repo and bind-mount that
+  directory to ``/var/lib/jenkins/pytorch`` in the debug container; resolve metadata from the local
+  checkout, stage artifacts on the host, and install the wheel via ``docker exec``.
+* ``--repo-dir DIR``: host-mapped mode reusing an existing host clone (fetch/checkout the commit),
+  bind-mounted to ``/var/lib/jenkins/pytorch``.
+
+Dry run (inspect only):
+* ``--workflow-parse-only``: applies on top of any acquisition mode. It resolves the
+  ``(commit, workflow)`` to the ROCm build job label, docker-image stem, and ``.ci/docker`` tree
+  sha, prints them, and exits — WITHOUT downloading CI artifacts, pulling or ``docker run``-ing the
+  image, or installing the wheel. The default mode resolves via the GitHub API (no clone), while
+  ``--host-mapped``/``--repo-dir`` resolve from a local checkout (and additionally print
+  ``clone-path``).
+
+  Dry-run examples (no artifacts/image/container; see the GH_TOKEN note under Examples):
+
+    # Default mode: resolve via the GitHub API, no clone.
+    python3 scripts/pytorch_rocm_ci_debug_env.py <commit> trunk --workflow-parse-only
+
+    # Via a host temp clone (also prints clone-path).
+    python3 scripts/pytorch_rocm_ci_debug_env.py <commit> trunk --host-mapped --workflow-parse-only
+
+    # Via an existing host clone (also prints clone-path).
+    python3 scripts/pytorch_rocm_ci_debug_env.py <commit> trunk \
+        --repo-dir /path/to/pytorch --workflow-parse-only
+
+Auth: optional ``GITHUB_TOKEN`` or ``GH_TOKEN`` (required in practice for the GitHub API calls
+used by the default in-container mode).
+
+Requires: ``pip install pyyaml`` (or toolkit ``requirements.txt``).
+
+Examples — real runs that build the debug container and install the wheel (a ``GITHUB_TOKEN``/
+``GH_TOKEN`` is recommended, and required in practice for the default in-container mode's GitHub API
+calls; e.g. ``export GH_TOKEN=$(gh auth token)``). For inspect-only invocations see the Dry-run
+examples above.
+
+    # Default: clone + install inside the container (no host clone, no bind mount).
+    python3 scripts/pytorch_rocm_ci_debug_env.py <commit> trunk
+
+    # Host-mapped: clone to a host temp dir and bind-mount it into the container.
+    python3 scripts/pytorch_rocm_ci_debug_env.py <commit> trunk --host-mapped
+
+    # Host-mapped reusing an existing clone (fetch/checkout the commit, then bind-mount it).
+    python3 scripts/pytorch_rocm_ci_debug_env.py <commit> trunk --repo-dir /path/to/pytorch
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import getpass
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+try:
+    import yaml
+except ImportError:
+    print("error: PyYAML is required. Install with: pip install pyyaml", file=sys.stderr)
+    sys.exit(1)
+
+from cli_common import CliError, cli_fail
+
+
+# ---------------------------------------------------------------------------
+# Repository identity (edit for a fork)
+# ---------------------------------------------------------------------------
+
+REPO_OWNER = "pytorch"
+REPO_NAME = "pytorch"
+PYTORCH_CLONE_URL = f"https://github.com/{REPO_OWNER}/{REPO_NAME}.git"
+
+
+# ---------------------------------------------------------------------------
+# Git
+# ---------------------------------------------------------------------------
+
+
+def git_stdout(repo: Path, *args: str) -> str:
+    completed = subprocess.run(
+        ["git", "-C", str(repo), *args],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    return completed.stdout.strip()
+
+
+def clone_shallow_then_checkout(repo_url: str, dest: Path, commit: str) -> None:
+    subprocess.run(
+        ["git", "clone", "--filter=blob:none", "--no-checkout", repo_url, str(dest)],
+        check=True,
+    )
+    shallow = subprocess.run(
+        ["git", "-C", str(dest), "fetch", "--depth", "1", "origin", commit],
+        capture_output=True,
+        text=True,
+    )
+    if shallow.returncode != 0:
+        full = subprocess.run(
+            ["git", "-C", str(dest), "fetch", "origin", commit],
+            capture_output=True,
+            text=True,
+        )
+        if full.returncode != 0:
+            raise RuntimeError(
+                f"git fetch failed for {commit!r}:\n{shallow.stderr}\n{full.stderr}"
+            )
+    subprocess.run(["git", "-C", str(dest), "checkout", "--detach", commit], check=True)
+
+
+def verify_detached_head_matches(repo: Path, commit: str) -> None:
+    head = git_stdout(repo, "rev-parse", "HEAD")
+    target = git_stdout(repo, "rev-parse", commit)
+    if head != target:
+        raise RuntimeError(f"HEAD {head!r} does not match expected {target!r}")
+
+
+def branch_label_for_container_env(repo: Path) -> str:
+    """Value for docker ``BRANCH=``; detached HEAD uses env ``BRANCH`` or ``master``."""
+    try:
+        abbrev = git_stdout(repo, "rev-parse", "--abbrev-ref", "HEAD")
+    except subprocess.CalledProcessError:
+        return os.environ.get("BRANCH", "master")
+    if abbrev == "HEAD":
+        return os.environ.get("BRANCH", "master")
+    return abbrev
+
+
+def checkout_user_repo(repo: Path, commit: str) -> None:
+    fetch = subprocess.run(
+        ["git", "-C", str(repo), "fetch", "--quiet", "origin", commit],
+        capture_output=True,
+        text=True,
+    )
+    if fetch.returncode != 0:
+        raise RuntimeError(f"git fetch failed for {commit!r}:\n{fetch.stderr}")
+    subprocess.run(
+        ["git", "-C", str(repo), "checkout", "--detach", "--quiet", commit],
+        check=True,
+    )
+
+
+def _safe_name_segment(value: str, *, fallback: str) -> str:
+    """One segment restricted to ``[A-Za-z0-9._-]`` (unsafe chars -> ``_``).
+
+    This character set is safe for both a single filesystem path segment and a Docker
+    container-name segment, so path and container-name helpers share it.
+    """
+    cleaned = "".join(c if (c.isalnum() or c in "._-") else "_" for c in value.strip())
+    return cleaned or fallback
+
+
+def filesystem_safe_commit_segment(ref: str) -> str:
+    """Commit-ish string safe for a single path segment (no slashes)."""
+    return _safe_name_segment(ref, fallback="unknown")
+
+
+def acquire_repository(
+    commit: str,
+    repo_dir: Path | None,
+) -> tuple[Path, Path | None]:
+    """
+    Return ``(repo_root, managed_clone_dir_or_none)`` for the host-clone modes.
+    ``managed_clone_dir_or_none`` is set when this script created the clone (temp dir prefix
+    ``pytorch-ghcr-``); that directory is kept on disk after exit. The default in-container mode
+    does not use this function (it performs no host clone).
+    """
+    if repo_dir is not None:
+        root = repo_dir.resolve()
+        if not root.is_dir():
+            raise CliError(f"--repo-dir is not a directory: {root}")
+        checkout_user_repo(root, commit)
+        return root, None
+
+    tmp = Path(tempfile.mkdtemp(prefix="pytorch-ghcr-"))
+    try:
+        clone_shallow_then_checkout(PYTORCH_CLONE_URL, tmp, commit)
+    except (RuntimeError, subprocess.CalledProcessError) as exc:
+        raise CliError(f"clone/checkout failed: {exc}") from exc
+    return tmp, tmp
+
+
+# ---------------------------------------------------------------------------
+# Workflow YAML (docker-image-name + build job labels)
+# ---------------------------------------------------------------------------
+
+
+def is_static_github_actions_string(value: str) -> bool:
+    """False for empty strings and ``${{ }}`` expressions we cannot evaluate locally."""
+    s = value.strip()
+    if not s:
+        return False
+    return not (s.startswith("${{") and s.endswith("}}"))
+
+
+def _job_text_has_rocm_not_cuda(*parts: str) -> bool:
+    combined = " ".join(p.lower() for p in parts if p)
+    return "rocm" in combined and "cuda" not in combined
+
+
+def find_rocm_build_jobs(workflow: dict[str, Any]) -> list[tuple[str, dict[str, Any]]]:
+    """Jobs whose id and display name match ROCm build (``rocm``, not ``cuda``)."""
+    jobs = workflow.get("jobs")
+    if not isinstance(jobs, dict):
+        return []
+    matched: list[tuple[str, dict[str, Any]]] = []
+    for job_id, job_def in jobs.items():
+        if not isinstance(job_def, dict):
+            continue
+        jid = str(job_id).strip()
+        if not jid or "build" not in jid.lower():
+            continue
+        raw_name = job_def.get("name")
+        name = raw_name.strip() if isinstance(raw_name, str) else ""
+        if _job_text_has_rocm_not_cuda(jid, name):
+            matched.append((jid, job_def))
+    return matched
+
+
+def rocm_build_job_label(job_id: str, job_def: dict[str, Any]) -> str:
+    """Artifact / Actions label: static ``job.name`` if set, else the YAML job key."""
+    raw_name = job_def.get("name")
+    if isinstance(raw_name, str):
+        n = raw_name.strip()
+        if n and is_static_github_actions_string(n):
+            return n
+    return job_id.strip()
+
+
+def _require_single_rocm_build_job(
+    workflow: dict[str, Any], *, workflow_path: Path
+) -> tuple[str, dict[str, Any]]:
+    matched = find_rocm_build_jobs(workflow)
+    if not matched:
+        raise CliError(
+            f"no ROCm build job in {workflow_path} "
+            "(job id/name must contain 'rocm', not 'cuda', and id must contain 'build')."
+        )
+    if len(matched) > 1:
+        ids = [job_id for job_id, _ in matched]
+        raise CliError(
+            f"expected exactly one ROCm build job in {workflow_path}, found {len(matched)}: {ids}"
+        )
+    return matched[0]
+
+
+def collect_build_job_label(workflow: dict[str, Any], *, workflow_path: Path) -> str:
+    """Exactly one ROCm build job; excludes CUDA and other build jobs."""
+    job_id, job_def = _require_single_rocm_build_job(workflow, workflow_path=workflow_path)
+    return rocm_build_job_label(job_id, job_def)
+
+
+def collect_docker_image_stem(workflow: dict[str, Any], *, workflow_path: Path) -> str:
+    """``docker-image-name`` from the single ROCm build job ``with:`` block."""
+    _, job_def = _require_single_rocm_build_job(workflow, workflow_path=workflow_path)
+    with_block = job_def.get("with")
+    if not isinstance(with_block, dict):
+        raise CliError(f"ROCm build job has no 'with:' mapping in {workflow_path}.")
+    raw = with_block.get("docker-image-name")
+    if not isinstance(raw, str):
+        raise CliError(f"ROCm build job missing docker-image-name in {workflow_path}.")
+    if not is_static_github_actions_string(raw):
+        raise CliError(
+            f"ROCm build job docker-image-name is not a static string in {workflow_path}: {raw!r}"
+        )
+    return raw.strip()
+
+
+def load_workflow_yaml(path: Path) -> dict[str, Any]:
+    data = yaml.safe_load(path.read_text(encoding="utf-8"))
+    if not isinstance(data, dict):
+        raise ValueError(f"workflow root must be a mapping: {path}")
+    return data
+
+
+def workflow_file_path(repo: Path, workflow_stem: str) -> Path:
+    return repo / ".github" / "workflows" / f"{workflow_stem}.yml"
+
+
+@dataclass(frozen=True)
+class WorkflowResolution:
+    """Checkout-derived workflow metadata used for artifacts, pulls, and ``docker run``."""
+
+    workflow_path: Path
+    docker_directory_sha: str  # ``git rev-parse HEAD:.ci/docker``
+    image_stem: str
+    build_label: str
+    head_sha: str
+
+
+def resolve_workflow(repo: Path, commit: str, workflow_stem: str) -> WorkflowResolution:
+    verify_detached_head_matches(repo, commit)
+
+    path = workflow_file_path(repo, workflow_stem)
+    if not path.is_file():
+        raise CliError(f"workflow file not found: {path}")
+
+    try:
+        workflow = load_workflow_yaml(path)
+    except (OSError, yaml.YAMLError, ValueError) as exc:
+        raise CliError(f"failed to read workflow YAML: {exc}") from exc
+
+    build_label = collect_build_job_label(workflow, workflow_path=path)
+    image_stem = collect_docker_image_stem(workflow, workflow_path=path)
+
+    try:
+        docker_directory_sha = git_stdout(repo, "rev-parse", "HEAD:.ci/docker")
+    except subprocess.CalledProcessError as exc:
+        raise CliError(f"git rev-parse HEAD:.ci/docker failed:\n{exc.stderr}") from exc
+
+    try:
+        head_sha = git_stdout(repo, "rev-parse", commit)
+    except subprocess.CalledProcessError as exc:
+        raise CliError(f"git rev-parse failed for {commit!r}:\n{exc.stderr}") from exc
+
+    return WorkflowResolution(
+        workflow_path=path,
+        docker_directory_sha=docker_directory_sha,
+        image_stem=image_stem,
+        build_label=build_label,
+        head_sha=head_sha,
+    )
+
+
+def resolve_workflow_via_github(
+    commit: str, workflow_stem: str, token: str | None
+) -> WorkflowResolution:
+    """Resolve workflow metadata for the default in-container mode with no host checkout.
+
+    Reuses the same workflow-parsing functions as ``resolve_workflow`` on a dict parsed from the
+    raw workflow YAML fetched from GitHub, and derives the ``.ci/docker`` tree sha and full head
+    sha from the contents/commits APIs. ``workflow_path`` is a non-filesystem stand-in used only
+    for ``.name`` (the run lookup) and error messages.
+    """
+    workflow_repo_path = f".github/workflows/{workflow_stem}.yml"
+    path = Path(workflow_repo_path)
+
+    text = github_file_text(workflow_repo_path, commit, token)
+    try:
+        workflow = yaml.safe_load(text)
+    except yaml.YAMLError as exc:
+        raise CliError(f"failed to parse workflow YAML from GitHub: {exc}") from exc
+    if not isinstance(workflow, dict):
+        raise CliError(f"workflow root must be a mapping: {workflow_repo_path}")
+
+    build_label = collect_build_job_label(workflow, workflow_path=path)
+    image_stem = collect_docker_image_stem(workflow, workflow_path=path)
+    docker_directory_sha = github_tree_entry_sha(".ci", "docker", commit, token)
+    head_sha = github_commit_full_sha(commit, token)
+
+    return WorkflowResolution(
+        workflow_path=path,
+        docker_directory_sha=docker_directory_sha,
+        image_stem=image_stem,
+        build_label=build_label,
+        head_sha=head_sha,
+    )
+
+
+# ---------------------------------------------------------------------------
+# GitHub Actions API
+# ---------------------------------------------------------------------------
+
+
+def github_api_token() -> str | None:
+    return os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN")
+
+
+def github_get_json(url: str, token: str | None) -> Any:
+    headers = {
+        "Accept": "application/vnd.github+json",
+        "User-Agent": "pytorch-rocm-ci-debug-env",
+        "X-GitHub-Api-Version": "2022-11-28",
+    }
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    request = urllib.request.Request(url, headers=headers)
+    with urllib.request.urlopen(request, timeout=120) as response:
+        return json.loads(response.read().decode("utf-8"))
+
+
+def github_contents_url(path: str, commit: str) -> str:
+    encoded_path = urllib.parse.quote(path, safe="/")
+    query = urllib.parse.urlencode({"ref": commit})
+    return f"https://api.github.com/repos/{REPO_OWNER}/{REPO_NAME}/contents/{encoded_path}?{query}"
+
+
+def github_get_json_or_fail(url: str, token: str | None, *, what: str) -> Any:
+    """Wrap ``github_get_json`` and convert network/HTTP errors into ``CliError``."""
+    try:
+        return github_get_json(url, token)
+    except urllib.error.HTTPError as exc:
+        body = exc.read().decode("utf-8", errors="replace") if exc.fp else ""
+        raise CliError(f"GitHub API HTTP {exc.code} fetching {what}: {body.strip()}") from exc
+    except urllib.error.URLError as exc:
+        raise CliError(f"GitHub API request failed fetching {what}: {exc}") from exc
+
+
+def github_file_text(path: str, commit: str, token: str | None) -> str:
+    """Decoded text of a repo file at ``commit`` via the contents API (base64 ``content``)."""
+    payload = github_get_json_or_fail(github_contents_url(path, commit), token, what=path)
+    if not isinstance(payload, dict):
+        raise CliError(f"unexpected contents response for {path} (not an object)")
+    content = payload.get("content")
+    encoding = payload.get("encoding")
+    if encoding != "base64" or not isinstance(content, str):
+        raise CliError(f"contents response for {path} is not base64-encoded")
+    return base64.b64decode(content).decode("utf-8")
+
+
+def github_tree_entry_sha(parent_path: str, name: str, commit: str, token: str | None) -> str:
+    """Git tree sha of entry ``name`` under directory ``parent_path`` at ``commit``.
+
+    For a directory entry this equals ``git rev-parse <commit>:<parent_path>/<name>``.
+    """
+    payload = github_get_json_or_fail(
+        github_contents_url(parent_path, commit), token, what=parent_path
+    )
+    if not isinstance(payload, list):
+        raise CliError(f"contents response for {parent_path} is not a directory listing")
+    for entry in payload:
+        if isinstance(entry, dict) and entry.get("name") == name:
+            sha = entry.get("sha")
+            if isinstance(sha, str) and sha:
+                return sha
+            raise CliError(f"entry {name!r} under {parent_path} has no sha")
+    raise CliError(f"no entry named {name!r} under {parent_path} at {commit}")
+
+
+def github_commit_full_sha(commit: str, token: str | None) -> str:
+    url = f"https://api.github.com/repos/{REPO_OWNER}/{REPO_NAME}/commits/{commit}"
+    payload = github_get_json_or_fail(url, token, what=f"commit {commit}")
+    if not isinstance(payload, dict):
+        raise CliError(f"unexpected commit response for {commit} (not an object)")
+    sha = payload.get("sha")
+    if not isinstance(sha, str) or not sha:
+        raise CliError(f"commit response for {commit} missing sha")
+    return sha
+
+
+def parse_github_datetime(raw: str) -> datetime:
+    if raw.endswith("Z"):
+        raw = raw[:-1] + "+00:00"
+    return datetime.fromisoformat(raw)
+
+
+def latest_workflow_run_id(workflow_filename: str, head_sha: str, token: str | None) -> int:
+    encoded_wf = urllib.parse.quote(workflow_filename, safe="")
+    query = urllib.parse.urlencode({"head_sha": head_sha, "per_page": "100"})
+    url = (
+        f"https://api.github.com/repos/{REPO_OWNER}/{REPO_NAME}/"
+        f"actions/workflows/{encoded_wf}/runs?{query}"
+    )
+    payload = github_get_json_or_fail(url, token, what=f"runs for {workflow_filename}")
+
+    runs = payload.get("workflow_runs") if isinstance(payload, dict) else None
+    if not isinstance(runs, list) or not runs:
+        raise CliError("GitHub API returned no workflow_runs for this workflow and head_sha")
+
+    def sort_key(run: Any) -> datetime:
+        created = run.get("created_at") if isinstance(run, dict) else None
+        if not isinstance(created, str):
+            return datetime.min.replace(tzinfo=timezone.utc)
+        try:
+            return parse_github_datetime(created)
+        except ValueError:
+            return datetime.min.replace(tzinfo=timezone.utc)
+
+    newest = max(runs, key=sort_key)
+    if not isinstance(newest, dict):
+        raise CliError("GitHub API workflow run entry is not an object")
+    run_id = newest.get("id")
+    if not isinstance(run_id, int):
+        raise CliError("GitHub API workflow run entry missing integer id")
+    return run_id
+
+
+# ---------------------------------------------------------------------------
+# Artifact downloads (gha-artifacts S3)
+# ---------------------------------------------------------------------------
+
+_FS_UNSAFE = str.maketrans({c: "_" for c in '<>:"/\\|?*\x00'})
+
+
+def filesystem_safe_label(label: str) -> str:
+    return label.translate(_FS_UNSAFE).strip() or "job"
+
+
+def artifact_zip_url(workflow_run_id: int, build_job_label: str) -> str:
+    o = urllib.parse.quote(REPO_OWNER, safe="")
+    r = urllib.parse.quote(REPO_NAME, safe="")
+    segment = urllib.parse.quote(build_job_label, safe="")
+    return (
+        f"https://gha-artifacts.s3.amazonaws.com/{o}/{r}/"
+        f"{workflow_run_id}/{segment}/artifacts.zip"
+    )
+
+
+def format_byte_size(num_bytes: int | float) -> str:
+    value = float(max(num_bytes, 0))
+    for scale, suffix in ((1 << 30, "GiB"), (1 << 20, "MiB"), (1 << 10, "KiB")):
+        if value >= scale:
+            return f"{value / scale:.2f} {suffix}"
+    return f"{int(num_bytes)} B"
+
+
+def download_with_progress(url: str, destination: Path, *, stderr_prefix: str = "") -> None:
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    request = urllib.request.Request(url, headers={"User-Agent": "pytorch-rocm-ci-debug-env"})
+    chunk_bytes = 256 * 1024
+    prefix = f"{stderr_prefix} " if stderr_prefix else ""
+
+    try:
+        with urllib.request.urlopen(request, timeout=600) as response:
+            status = getattr(response, "status", None) or response.getcode()
+            if status != 200:
+                raise RuntimeError(f"unexpected HTTP status {status}")
+
+            length_header = response.headers.get("Content-Length")
+            try:
+                total_bytes = int(length_header) if length_header else None
+            except (TypeError, ValueError):
+                total_bytes = None
+
+            downloaded = 0
+            with destination.open("wb") as outfile:
+                while True:
+                    chunk = response.read(chunk_bytes)
+                    if not chunk:
+                        break
+                    outfile.write(chunk)
+                    downloaded += len(chunk)
+                    if total_bytes and total_bytes > 0:
+                        pct = 100.0 * downloaded / total_bytes
+                        line = (
+                            f"{prefix}artifacts.zip {format_byte_size(downloaded)} / "
+                            f"{format_byte_size(total_bytes)} ({pct:.1f}%)"
+                        )
+                    else:
+                        line = f"{prefix}artifacts.zip {format_byte_size(downloaded)}"
+                    sys.stderr.write(f"\r{line}\x1b[K")
+                    sys.stderr.flush()
+
+            sys.stderr.write("\r\x1b[2K")
+            sys.stderr.flush()
+            print(file=sys.stderr)
+
+    except urllib.error.HTTPError as exc:
+        raise RuntimeError(f"HTTP {exc.code} when downloading {url}") from exc
+
+
+def download_build_artifact(repo: Path, build_label: str, workflow_run_id: int) -> None:
+    url = artifact_zip_url(workflow_run_id, build_label)
+    dest = repo / "gha-artifacts" / filesystem_safe_label(build_label) / "artifacts.zip"
+    print(url)
+    download_with_progress(url, dest, stderr_prefix=f"[{build_label}]")
+    print(f"artifact-zip: {dest}", file=sys.stderr)
+
+
+def copy_build_artifact_to_repo_root(repo: Path, build_label: str) -> None:
+    src = repo / "gha-artifacts" / filesystem_safe_label(build_label) / "artifacts.zip"
+    if not src.is_file():
+        raise RuntimeError(f"missing artifact zip for {build_label!r}: {src}")
+    shutil.copy2(src, repo / "artifacts.zip")
+    print(f"staged-artifacts-zip-from: {build_label}", file=sys.stderr)
+
+
+# ---------------------------------------------------------------------------
+# Docker
+# ---------------------------------------------------------------------------
+
+DOCKER_RUN_ENV_LITERAL: tuple[tuple[str, str], ...] = (
+    ("GITHUB_ACTIONS", "true"),
+    ("PYTORCH_RETRY_TEST_CASES", "1"),
+    ("PYTORCH_OVERRIDE_FLAKY_SIGNAL", "1"),
+    ("SCCACHE_BUCKET", "ossci-compiler-cache-circleci-v2"),
+    ("XLA_CLANG_CACHE_S3_BUCKET_NAME", "ossci-compiler-clang-cache-circleci-xla"),
+    ("PYTORCH_TEST_CUDA_MEM_LEAK_CHECK", "0"),
+    ("PYTORCH_TEST_RERUN_DISABLED_TESTS", "0"),
+    ("CI", "True"),
+)
+
+DOCKER_RUN_ENV_FROM_HOST = (
+    "AWS_DEFAULT_REGION",
+    "IN_WHEEL_TEST",
+    "PR_BODY",
+    "COMMIT_MESSAGES",
+)
+
+# Jenkins HOME shipped by the CI image (pip/sccache caches, dotfiles, etc.). It is never used as
+# a bind-mount target, so the image's own contents are preserved (not shadowed).
+CONTAINER_HOME_DIR = "/var/lib/jenkins"
+
+# Repo root for ALL modes: a child dir of the jenkins HOME. The host-mapped modes bind-mount the
+# checkout here; the default in-container mode clones here. Using a subdir (rather than the HOME
+# itself) avoids shadowing the image workspace and keeps the chown scope small.
+CONTAINER_REPO_DIR = f"{CONTAINER_HOME_DIR}/pytorch"
+
+
+def container_setup_script(
+    work_dir: str,
+    *,
+    clone_url: str | None = None,
+    commit: str | None = None,
+    artifact_url: str | None = None,
+) -> str:
+    """Bash that ends at the test dir: (optional clone + artifact download), unzip, pip install.
+
+    ``work_dir`` is the repo root (``CONTAINER_REPO_DIR`` for every mode). With ``clone_url``/
+    ``commit`` it first ``git clone``s + ``checkout --detach``s into ``work_dir`` (used by the
+    default in-container mode). With ``artifact_url`` it downloads ``artifacts.zip`` into
+    ``work_dir`` (the in-container path; the host-mapped modes stage the zip via the bind mount
+    instead). The unzip/``pip install``/``cd test`` tail is shared by all modes, as is a
+    ``/workspace/pytorch`` -> ``work_dir`` convenience symlink.
+    """
+    lines = ["set -e"]
+    if clone_url and commit:
+        lines += [
+            f"rm -rf {work_dir}",
+            f"git clone --filter=blob:none --no-checkout {clone_url} {work_dir}",
+            f"cd {work_dir}",
+            f"git fetch --depth 1 origin {commit} || git fetch origin {commit}",
+            f"git checkout --detach {commit}",
+        ]
+    lines += [
+        f"sudo chown -R jenkins:jenkins {work_dir}",
+        # Convenience symlink so /workspace/pytorch resolves to the unified repo root.
+        "sudo mkdir -p /workspace",
+        f"sudo ln -sfn {work_dir} /workspace/pytorch",
+        f"cd {work_dir}",
+        "if ! command -v unzip >/dev/null 2>&1; then",
+        "  export DEBIAN_FRONTEND=noninteractive",
+        "  apt-get update -qq && apt-get install -y -qq unzip",
+        "fi",
+    ]
+    if artifact_url:
+        lines += [
+            f'if command -v curl >/dev/null 2>&1; then curl -fSL -o artifacts.zip "{artifact_url}";',
+            f'else wget -O artifacts.zip "{artifact_url}"; fi',
+        ]
+    lines += [
+        "test -f artifacts.zip",
+        "unzip -o -q artifacts.zip",
+        "cd dist",
+        "pip install *.whl",
+        f"cd {work_dir}/test",
+        "pwd >&2",
+    ]
+    return "\n".join(lines)
+
+
+DEBUG_ENV_NAME_INFIX = "pytorch_rocm_ci_debug_env"
+
+
+def launching_user_id() -> str:
+    """Login name of the user running the script, for the container name.
+
+    Prefers ``getpass.getuser()`` (honors ``LOGNAME``/``USER``/``LNAME``/``USERNAME``,
+    then the password database for the current uid); falls back to the numeric uid,
+    then the literal ``user`` if neither is resolvable.
+    """
+    try:
+        name = getpass.getuser().strip()
+    except (OSError, KeyError):
+        name = ""
+    if name:
+        return name
+    getuid = getattr(os, "getuid", None)
+    if getuid is not None:
+        return str(getuid())
+    return "user"
+
+
+def debug_env_container_name(user: str, head_sha: str, workflow: str) -> str:
+    """``<user>_pytorch_rocm_ci_debug_env_<short-sha>_<workflow>``, Docker-name safe.
+
+    Each segment is restricted to ``[A-Za-z0-9._-]``; the commit uses the first 12
+    chars of the resolved head sha. Docker requires the first char to be alphanumeric,
+    so a leading ``x_`` is prepended if the assembled name would not start with one.
+    """
+    user_seg = _safe_name_segment(user, fallback="user")
+    sha_seg = _safe_name_segment(head_sha[:12], fallback="commit")
+    workflow_seg = _safe_name_segment(workflow, fallback="workflow")
+    name = f"{user_seg}_{DEBUG_ENV_NAME_INFIX}_{sha_seg}_{workflow_seg}"
+    if not name[:1].isalnum():
+        name = f"x_{name}"
+    return name
+
+
+def docker_container_name_in_use(name: str) -> bool:
+    """True if a container (running or stopped) with exactly ``name`` already exists."""
+    completed = subprocess.run(
+        ["docker", "ps", "-aq", "--filter", f"name=^{name}$"],
+        capture_output=True,
+        text=True,
+    )
+    return completed.returncode == 0 and bool(completed.stdout.strip())
+
+
+def host_render_gid() -> str | None:
+    """GID of the host ``render`` group (setup-rocm action.yml pattern), if present."""
+    try:
+        for line in Path("/etc/group").read_text(encoding="utf-8").splitlines():
+            if line.startswith("render:"):
+                parts = line.split(":")
+                if len(parts) >= 3 and parts[2].isdigit():
+                    return parts[2]
+    except OSError:
+        pass
+    return None
+
+
+def docker_run_pytorch_container(
+    image_ref: str,
+    repo: Path | None,
+    head_sha: str,
+    branch: str,
+    container_name: str,
+) -> str:
+    """Start a detached container; ``-i`` without ``-t`` so stdin is not required.
+
+    ``repo`` is bind-mounted to ``CONTAINER_REPO_DIR`` for the host-clone modes (docker auto-creates
+    the target dir, so the image's jenkins HOME is preserved, not shadowed). When ``repo`` is
+    ``None`` (the default in-container mode) no bind mount is added and the clone happens inside the
+    container. Either way the repo root is ``CONTAINER_REPO_DIR``.
+
+    ``container_name`` is the required ``docker run --name`` (see ``debug_env_container_name``).
+    """
+    command: list[str] = ["docker", "run", "-i", "--name", container_name]
+    for device in ("/dev/mem", "/dev/kfd", "/dev/dri"):
+        command.extend(["--device", device])
+    command.extend(
+        ["--user", "jenkins", "--group-add", "bin", "--group-add", "video", "--group-add", "daemon"]
+    )
+    render_gid = host_render_gid()
+    if render_gid:
+        command.extend(["--group-add", render_gid])
+    else:
+        print("note: host has no render group; skipping --group-add for render", file=sys.stderr)
+    command.extend(["-e", f"BRANCH={branch}", "-e", f"SHA1={head_sha}"])
+    for key, val in DOCKER_RUN_ENV_LITERAL:
+        command.extend(["-e", f"{key}={val}"])
+    for key in DOCKER_RUN_ENV_FROM_HOST:
+        command.extend(["-e", key])
+    command.extend(
+        [
+            "--ulimit",
+            "stack=10485760:83886080",
+            "--security-opt",
+            "seccomp=unconfined",
+            "--cap-add=SYS_PTRACE",
+            "--shm-size=8g",
+            "--detach",
+            "--network=host",
+        ]
+    )
+    if repo is not None:
+        command.extend(["-v", f"{repo.resolve()}:{CONTAINER_REPO_DIR}"])
+        work_dir = CONTAINER_REPO_DIR
+    else:
+        work_dir = CONTAINER_HOME_DIR
+    command.extend(["-w", work_dir, image_ref])
+    completed = subprocess.run(command, capture_output=True, text=True)
+    if completed.returncode != 0:
+        raise RuntimeError(
+            f"docker run failed ({completed.returncode}): "
+            f"{(completed.stderr or completed.stdout).strip()}"
+        )
+    container_id = (completed.stdout or "").strip()
+    if not container_id:
+        raise RuntimeError("docker run produced no container id on stdout")
+    return container_id
+
+
+def docker_inspect_container_name(container_id: str) -> str:
+    completed = subprocess.run(
+        ["docker", "inspect", "-f", "{{.Name}}", container_id],
+        capture_output=True,
+        text=True,
+    )
+    if completed.returncode != 0:
+        raise RuntimeError((completed.stderr or completed.stdout).strip())
+    return completed.stdout.strip().lstrip("/")
+
+
+def docker_exec_chown_workspace(container_id: str, target_dir: str) -> None:
+    """Root chown of ``target_dir`` so the jenkins user can write to it."""
+    completed = subprocess.run(
+        [
+            "docker",
+            "exec",
+            "-u",
+            "root",
+            container_id,
+            "chown",
+            "-R",
+            "jenkins:jenkins",
+            target_dir,
+        ],
+        capture_output=True,
+        text=True,
+    )
+    if completed.returncode != 0:
+        detail = (completed.stderr or completed.stdout).strip()
+        raise RuntimeError(
+            f"docker exec -u root chown failed ({completed.returncode}): {detail}"
+        )
+
+
+def _docker_exec_bash(container_id: str, script: str) -> None:
+    completed = subprocess.run(["docker", "exec", container_id, "bash", "-lc", script])
+    if completed.returncode != 0:
+        raise RuntimeError(f"docker exec failed with exit code {completed.returncode}")
+
+
+def docker_exec_install_wheel(container_id: str) -> None:
+    """Host-clone modes: the bind mount at ``CONTAINER_REPO_DIR`` already has ``artifacts.zip``."""
+    docker_exec_chown_workspace(container_id, CONTAINER_REPO_DIR)
+    _docker_exec_bash(container_id, container_setup_script(CONTAINER_REPO_DIR))
+
+
+def docker_exec_clone_and_install_wheel(
+    container_id: str, commit: str, artifact_url: str
+) -> None:
+    """Default in-container mode: clone, download artifacts, and install the wheel in-container."""
+    # The repo subdir does not exist yet, so chown the jenkins HOME so the clone can create it.
+    docker_exec_chown_workspace(container_id, CONTAINER_HOME_DIR)
+    script = container_setup_script(
+        CONTAINER_REPO_DIR,
+        clone_url=PYTORCH_CLONE_URL,
+        commit=commit,
+        artifact_url=artifact_url,
+    )
+    _docker_exec_bash(container_id, script)
+
+
+def docker_pull_with_progress(image_reference: str) -> int:
+    """
+    Run ``docker pull`` and print JSON progress as layer lines on stderr.
+
+    Docker's plain progress mode lacks per-layer percentages; we disable hints and parse JSON lines.
+    """
+    process = subprocess.Popen(
+        ["docker", "pull", image_reference],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        env={**os.environ, "DOCKER_CLI_HINTS": "false"},
+    )
+
+    layers: dict[str, str] = {}
+    lines_on_screen = 0
+
+    def clear_layers() -> None:
+        nonlocal lines_on_screen
+        if lines_on_screen:
+            sys.stderr.write(f"\x1b[{lines_on_screen}A\x1b[0J")
+            lines_on_screen = 0
+
+    def redraw_layers() -> None:
+        nonlocal lines_on_screen
+        clear_layers()
+        if layers:
+            block = "\n".join(f"  {layer_id[:12]}  {status}" for layer_id, status in layers.items())
+            sys.stderr.write(block + "\n")
+            sys.stderr.flush()
+            lines_on_screen = len(layers)
+
+    assert process.stdout is not None
+    for raw_line in process.stdout:
+        line = raw_line.rstrip("\n")
+        try:
+            event = json.loads(line)
+        except json.JSONDecodeError:
+            clear_layers()
+            print(line, file=sys.stderr)
+            redraw_layers()
+            continue
+
+        error_message = event.get("error")
+        if error_message:
+            clear_layers()
+            print(f"error: {error_message}", file=sys.stderr)
+            continue
+
+        status = event.get("status")
+        if not status:
+            continue
+
+        layer_id = event.get("id")
+        detail = event.get("progressDetail") or {}
+        current = detail.get("current")
+        total = detail.get("total")
+
+        if layer_id:
+            if current is not None and total and total > 0:
+                pct = 100.0 * current / total
+                layers[layer_id] = (
+                    f"{status}: {format_byte_size(current)} / "
+                    f"{format_byte_size(total)} ({pct:.1f}%)"
+                )
+            elif current is not None:
+                layers[layer_id] = f"{status}: {format_byte_size(current)}"
+            else:
+                layers[layer_id] = status
+            redraw_layers()
+        else:
+            clear_layers()
+            print(status, file=sys.stderr)
+            redraw_layers()
+
+    clear_layers()
+    process.wait()
+    return process.returncode
+
+
+def ghcr_image_ref(image_stem: str, docker_directory_sha: str) -> str:
+    return f"ghcr.io/pytorch/{image_stem}-{docker_directory_sha}"
+
+
+def docker_image_present_locally(image_reference: str) -> bool:
+    completed = subprocess.run(
+        ["docker", "image", "inspect", image_reference],
+        capture_output=True,
+        text=True,
+    )
+    return completed.returncode == 0
+
+
+def pull_workflow_image(image_stem: str, docker_directory_sha: str) -> None:
+    reference = ghcr_image_ref(image_stem, docker_directory_sha)
+    print(reference)
+    if docker_image_present_locally(reference):
+        print(f"docker-image-present: {reference}", file=sys.stderr)
+        return
+    if docker_pull_with_progress(reference) != 0:
+        raise RuntimeError(f"docker pull failed for {reference}")
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+def parse_arguments() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Clone PyTorch at a commit, resolve a ROCm workflow, download CI artifacts, "
+            "pull GHCR images, start a container, and install the wheel inside it."
+        ),
+    )
+    parser.add_argument("commit", help="Git commit hash to check out")
+    parser.add_argument(
+        "workflow",
+        help=(
+            "Workflow file stem without .yml (e.g. trunk → .github/workflows/trunk.yml); "
+            "ROCm build job is selected from jobs in that file."
+        ),
+    )
+    modes = parser.add_argument_group(
+        "acquisition modes",
+        "How the PyTorch checkout reaches the container (pick at most one; default is in-container).",
+    )
+    modes.add_argument(
+        "--host-mapped",
+        action="store_true",
+        help=(
+            "Host-mapped mode: clone PyTorch to a host temp dir, bind-mount it to "
+            "/var/lib/jenkins/pytorch, resolve metadata from the local checkout, stage artifacts "
+            "on the host, and install the wheel via docker exec. Prints clone-path. Default "
+            "(without this flag or --repo-dir): everything happens inside the container."
+        ),
+    )
+    modes.add_argument(
+        "--repo-dir",
+        type=Path,
+        help=(
+            "Host-mapped mode using this existing clone (fetch/checkout the commit, bind-mount it) "
+            "instead of creating a temporary one"
+        ),
+    )
+    dry_run = parser.add_argument_group("dry run")
+    dry_run.add_argument(
+        "--workflow-parse-only",
+        action="store_true",
+        help=(
+            "Dry run: resolve (commit, workflow) to the ROCm build job label, docker-image stem, "
+            "and .ci/docker tree sha, print them, and exit. No artifact download, no image pull, "
+            "no docker run, no wheel install. Works with any mode: the default resolves via the "
+            "GitHub API (no clone); --host-mapped/--repo-dir resolve from a local checkout."
+        ),
+    )
+    return parser.parse_args()
+
+
+def use_in_container_mode(args: argparse.Namespace) -> bool:
+    """True for the default in-container mode (no host clone / no bind mount).
+
+    Host-mapped mode is selected by ``--host-mapped`` (temp clone) or ``--repo-dir`` (existing
+    clone); anything else defaults to cloning + installing inside the container.
+    """
+    return not args.host_mapped and args.repo_dir is None
+
+
+def resolve_for_mode(
+    args: argparse.Namespace, token: str | None
+) -> tuple[Path | None, Path | None, WorkflowResolution]:
+    """Resolve workflow metadata for the selected mode.
+
+    Returns ``(repo_root, temporary_clone_dir, resolution)``. In the default in-container mode both
+    paths are ``None`` (no host clone) and metadata comes from the GitHub API; for the host-mapped
+    modes the repo is acquired (temp clone or ``--repo-dir``) and parsed from the local checkout.
+    """
+    if use_in_container_mode(args):
+        return None, None, resolve_workflow_via_github(args.commit, args.workflow, token)
+    repo_root, temporary_clone_dir = acquire_repository(args.commit, args.repo_dir)
+    return repo_root, temporary_clone_dir, resolve_workflow(repo_root, args.commit, args.workflow)
+
+
+def main() -> int:
+    args = parse_arguments()
+
+    if args.repo_dir is not None and args.host_mapped:
+        return cli_fail("--repo-dir and --host-mapped cannot be used together")
+
+    in_container = use_in_container_mode(args)
+    token = github_api_token()
+
+    try:
+        repo_root, temporary_clone_dir, resolved = resolve_for_mode(args, token)
+    except CliError as exc:
+        return cli_fail(str(exc))
+
+    print(f"build-job-name: {resolved.build_label}", file=sys.stderr)
+    print(f"docker-image-stem: {resolved.image_stem}", file=sys.stderr)
+    print(f"docker-directory-sha: {resolved.docker_directory_sha}", file=sys.stderr)
+
+    if args.workflow_parse_only:
+        if temporary_clone_dir is not None:
+            print(f"clone-path: {temporary_clone_dir}", file=sys.stderr)
+        return 0
+
+    try:
+        run_id = latest_workflow_run_id(resolved.workflow_path.name, resolved.head_sha, token)
+        print(f"workflow-run-id: {run_id}", file=sys.stderr)
+        if not in_container:
+            download_build_artifact(repo_root, resolved.build_label, run_id)
+        pull_workflow_image(resolved.image_stem, resolved.docker_directory_sha)
+    except (CliError, RuntimeError) as exc:
+        return cli_fail(str(exc))
+
+    primary_image = ghcr_image_ref(resolved.image_stem, resolved.docker_directory_sha)
+    print(f"docker-run-image: {primary_image}", file=sys.stderr)
+
+    container_name = debug_env_container_name(
+        launching_user_id(), resolved.head_sha, args.workflow
+    )
+    if docker_container_name_in_use(container_name):
+        return cli_fail(
+            f"a container named {container_name!r} already exists; remove it with "
+            f"`docker rm -f {container_name}` (or rename it) and re-run"
+        )
+
+    container_id: str | None = None
+    try:
+        if in_container:
+            container_id = docker_run_pytorch_container(
+                primary_image,
+                None,
+                resolved.head_sha,
+                os.environ.get("BRANCH", "master"),
+                container_name,
+            )
+            artifact_url = artifact_zip_url(run_id, resolved.build_label)
+            print(artifact_url, file=sys.stderr)
+            docker_exec_clone_and_install_wheel(container_id, args.commit, artifact_url)
+        else:
+            copy_build_artifact_to_repo_root(repo_root, resolved.build_label)
+            container_id = docker_run_pytorch_container(
+                primary_image,
+                repo_root,
+                resolved.head_sha,
+                branch_label_for_container_env(repo_root),
+                container_name,
+            )
+            docker_exec_install_wheel(container_id)
+    except (CliError, RuntimeError) as exc:
+        if container_id:
+            subprocess.run(["docker", "rm", "-f", container_id], capture_output=True)
+        return cli_fail(str(exc))
+
+    print(f"docker-container-id: {container_id}", file=sys.stderr)
+    print("docker exec: unzip, pip install wheel, cd test - done.", file=sys.stderr)
+
+    if temporary_clone_dir is not None:
+        print(f"clone-path: {temporary_clone_dir}", file=sys.stderr)
+    if in_container:
+        print(f"container-clone-path: {CONTAINER_REPO_DIR}", file=sys.stderr)
+
+    try:
+        print(f"docker-container-name: {docker_inspect_container_name(container_id)}", file=sys.stderr)
+    except (CliError, RuntimeError) as exc:
+        return cli_fail(str(exc))
+
+    return 0
+
+
+def artifact_url_exists(url: str) -> bool:
+    request = urllib.request.Request(
+        url,
+        method="HEAD",
+        headers={"User-Agent": "pytorch-rocm-ci-debug-env"},
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=60) as response:
+            return 200 <= response.status < 300
+    except urllib.error.HTTPError:
+        return False
+
+
+def workflow_run_candidates(
+    workflow_filename: str,
+    *,
+    branch: str,
+    commit: str | None,
+    token: str | None,
+) -> list[tuple[str, int]]:
+    encoded_wf = urllib.parse.quote(workflow_filename, safe="")
+    if commit:
+        head_sha = github_commit_full_sha(commit, token)
+        query = urllib.parse.urlencode({"head_sha": head_sha, "per_page": "30"})
+    else:
+        query = urllib.parse.urlencode(
+            {"branch": branch, "status": "success", "per_page": "30"}
+        )
+    url = (
+        f"https://api.github.com/repos/{REPO_OWNER}/{REPO_NAME}/"
+        f"actions/workflows/{encoded_wf}/runs?{query}"
+    )
+    payload = github_get_json_or_fail(url, token, what=f"runs for {workflow_filename}")
+    runs = payload.get("workflow_runs") if isinstance(payload, dict) else None
+    if not isinstance(runs, list) or not runs:
+        raise CliError("GitHub API returned no workflow_runs for this workflow")
+
+    def sort_key(run: Any) -> datetime:
+        created = run.get("created_at") if isinstance(run, dict) else None
+        if not isinstance(created, str):
+            return datetime.min.replace(tzinfo=timezone.utc)
+        try:
+            return parse_github_datetime(created)
+        except ValueError:
+            return datetime.min.replace(tzinfo=timezone.utc)
+
+    candidates: list[tuple[str, int]] = []
+    for run in sorted(runs, key=sort_key, reverse=True):
+        if not isinstance(run, dict):
+            continue
+        run_id = run.get("id")
+        head_sha = run.get("head_sha")
+        if isinstance(run_id, int) and isinstance(head_sha, str) and head_sha:
+            candidates.append((head_sha, run_id))
+    return candidates
+
+
+def resolve_publish_workflow(
+    workflow: str,
+    *,
+    branch: str,
+    commit: str | None,
+    token: str | None,
+) -> tuple[WorkflowResolution, int, str]:
+    workflow_filename = workflow if workflow.endswith(".yml") else f"{workflow}.yml"
+    last_error: Exception | None = None
+    for head_sha, run_id in workflow_run_candidates(
+        workflow_filename,
+        branch=branch,
+        commit=commit,
+        token=token,
+    ):
+        try:
+            resolved = resolve_workflow_via_github(head_sha, workflow_filename[:-4], token)
+            artifact_url = artifact_zip_url(run_id, resolved.build_label)
+            if not artifact_url_exists(artifact_url):
+                print(
+                    f"artifact-missing: run={run_id} label={resolved.build_label}",
+                    file=sys.stderr,
+                )
+                continue
+            return resolved, run_id, artifact_url
+        except CliError as exc:
+            last_error = exc
+            continue
+    if last_error:
+        raise CliError(f"no usable workflow run found; last error: {last_error}") from last_error
+    raise CliError("no usable workflow run found with a public artifacts.zip")
+
+
+def run_checked(command: list[str]) -> str:
+    completed = subprocess.run(command, capture_output=True, text=True)
+    if completed.returncode != 0:
+        detail = (completed.stderr or completed.stdout).strip()
+        raise RuntimeError(f"{command[0]} failed ({completed.returncode}): {detail}")
+    return completed.stdout.strip()
+
+
+def docker_run_publish_container(image_ref: str, head_sha: str, branch: str) -> str:
+    command: list[str] = [
+        "docker",
+        "run",
+        "-i",
+        "-e",
+        f"BRANCH={branch}",
+        "-e",
+        f"SHA1={head_sha}",
+    ]
+    for key, val in DOCKER_RUN_ENV_LITERAL:
+        command.extend(["-e", f"{key}={val}"])
+    command.extend(
+        [
+            "--ulimit",
+            "stack=10485760:83886080",
+            "--security-opt",
+            "seccomp=unconfined",
+            "--cap-add=SYS_PTRACE",
+            "--shm-size=8g",
+            "--detach",
+            "--network=host",
+            "-w",
+            CONTAINER_HOME_DIR,
+            image_ref,
+            "sleep",
+            "infinity",
+        ]
+    )
+    return run_checked(command)
+
+
+def docker_exec_root_bash(container_id: str, script: str) -> None:
+    completed = subprocess.run(["docker", "exec", "-u", "root", container_id, "bash", "-lc", script])
+    if completed.returncode != 0:
+        raise RuntimeError(f"docker exec failed with exit code {completed.returncode}")
+
+
+def publish_container_setup_script(commit: str, artifact_url: str) -> str:
+    return f"""
+set -euo pipefail
+command -v unzip >/dev/null 2>&1 || {{ apt-get update -qq && apt-get install -y -qq unzip; }}
+command -v curl >/dev/null 2>&1 || {{ apt-get update -qq && apt-get install -y -qq curl; }}
+WORK={CONTAINER_REPO_DIR}
+rm -rf "$WORK"
+git clone --filter=blob:none --no-checkout {PYTORCH_CLONE_URL} "$WORK"
+cd "$WORK"
+git fetch --depth 1 origin {commit} || git fetch origin {commit}
+git checkout --detach {commit}
+mkdir -p /workspace
+ln -sfn "$WORK" /workspace/pytorch
+curl -fSL -o artifacts.zip "{artifact_url}"
+unzip -o -q artifacts.zip
+pip install dist/*.whl
+# Verify from outside the repo so the source tree's torch/ package does not shadow the wheel.
+( cd /tmp && python -c "import torch; print('torch', torch.__version__)" )
+"""
+
+
+def write_github_output(values: dict[str, str]) -> None:
+    output_path = os.environ.get("GITHUB_OUTPUT")
+    if not output_path:
+        return
+    with Path(output_path).open("a", encoding="utf-8") as output:
+        for key, value in values.items():
+            print(f"{key}={value}", file=output)
+
+
+def publish_prepared_container(
+    container_id: str,
+    resolved: WorkflowResolution,
+    *,
+    image: str,
+    tag: str,
+    push: bool,
+) -> tuple[str, str]:
+    date = datetime.now(timezone.utc).strftime("%Y%m%d")
+    dated_tag = f"nightly-{date}-{resolved.head_sha[:7]}"
+    primary = f"{image}:{tag}"
+    dated = f"{image}:{dated_tag}"
+    source_image = ghcr_image_ref(resolved.image_stem, resolved.docker_directory_sha)
+    run_checked(
+        [
+            "docker",
+            "commit",
+            "--change",
+            f"WORKDIR {CONTAINER_REPO_DIR}",
+            "--change",
+            f"LABEL pytorch.repo={REPO_OWNER}/{REPO_NAME}",
+            "--change",
+            f"LABEL pytorch.commit={resolved.head_sha}",
+            "--change",
+            f"LABEL pytorch.source_image={source_image}",
+            container_id,
+            primary,
+        ]
+    )
+    run_checked(["docker", "tag", primary, dated])
+    if push:
+        subprocess.run(["docker", "push", primary], check=True)
+        subprocess.run(["docker", "push", dated], check=True)
+    return tag, dated_tag
+
+
+def parse_publish_arguments(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Prepare a PyTorch ROCm CI container, commit it to an image, and optionally push tags."
+    )
+    parser.add_argument("--workflow", default="trunk.yml")
+    parser.add_argument("--branch", default="main")
+    parser.add_argument("--commit", default="")
+    parser.add_argument("--image", required=True, help="Image without tag, e.g. docker.io/rocm/pytorch-nightly")
+    parser.add_argument("--tag", default="nightly", help="Primary tag to publish")
+    parser.add_argument("--push", action="store_true", help="Push primary and dated tags")
+    return parser.parse_args(argv)
+
+
+def publish_main(argv: list[str]) -> int:
+    args = parse_publish_arguments(argv)
+    token = github_api_token()
+    try:
+        resolved, run_id, artifact_url = resolve_publish_workflow(
+            args.workflow,
+            branch=args.branch,
+            commit=args.commit.strip() or None,
+            token=token,
+        )
+        image_ref = ghcr_image_ref(resolved.image_stem, resolved.docker_directory_sha)
+        print(f"commit       = {resolved.head_sha}")
+        print(f"run_id       = {run_id}")
+        print(f"image_ref    = {image_ref}")
+        print(f"build_label  = {resolved.build_label}")
+        print(f"artifact_url = {artifact_url}")
+
+        pull_workflow_image(resolved.image_stem, resolved.docker_directory_sha)
+        container_id: str | None = None
+        try:
+            container_id = docker_run_publish_container(image_ref, resolved.head_sha, args.branch)
+            docker_exec_root_bash(
+                container_id,
+                publish_container_setup_script(resolved.head_sha, artifact_url),
+            )
+            primary_tag, dated_tag = publish_prepared_container(
+                container_id,
+                resolved,
+                image=args.image,
+                tag=args.tag,
+                push=args.push,
+            )
+            write_github_output(
+                {
+                    "commit": resolved.head_sha,
+                    "short_sha": resolved.head_sha[:7],
+                    "run_id": str(run_id),
+                    "image_ref": image_ref,
+                    "artifact_url": artifact_url,
+                    "build_label": resolved.build_label,
+                    "primary_tag": primary_tag,
+                    "dated_tag": dated_tag,
+                }
+            )
+            return 0
+        finally:
+            if container_id:
+                subprocess.run(["docker", "rm", "-f", container_id], capture_output=True, text=True)
+    except (CliError, RuntimeError, subprocess.CalledProcessError) as exc:
+        return cli_fail(str(exc))
+
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1 and sys.argv[1] == "publish":
+        raise SystemExit(publish_main(sys.argv[2:]))
+    raise SystemExit(main())
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() }}
diff --git a/.github/workflows/publish-pytorch-private-nightly.yml b/.github/workflows/publish-pytorch-private-nightly.yml
new file mode 100644
index 0000000000000..c0e6899802274
--- /dev/null
+++ b/.github/workflows/publish-pytorch-private-nightly.yml
@@ -0,0 +1,82 @@
+name: Publish pytorch-nightly (upstream CI image + built torch)
+
+# Nightly: take the pytorch/pytorch ROCm CI image at the latest trunk commit,
+# install the CI-built wheel and matching source into that image, then publish
+# it to Docker Hub.
+
+on:
+  schedule:
+    - cron: "0 7 * * *"   # daily at 07:00 UTC
+  workflow_dispatch:
+    inputs:
+      source_workflow:
+        description: "Upstream pytorch/pytorch workflow file that builds the ROCm CI wheel"
+        type: string
+        default: "trunk.yml"
+      commit:
+        description: "pytorch/pytorch commit to use (default: latest main with a successful build + artifact)"
+        type: string
+        default: ""
+      push_tag:
+        description: "Primary Docker Hub tag"
+        type: string
+        default: "nightly"
+
+permissions:
+  contents: read
+
+env:
+  REGISTRY: docker.io
+  IMAGE_NAME: rocm/pytorch-nightly
+  UPSTREAM_BRANCH: main
+  SOURCE_WORKFLOW: ${{ inputs.source_workflow || 'trunk.yml' }}
+  PUSH_TAG: ${{ inputs.push_tag || 'nightly' }}
+
+jobs:
+  publish:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout workflow scripts
+        uses: actions/checkout@v4
+
+      - name: Install script dependencies
+        run: python3 -m pip install --user pyyaml
+
+      - name: Log in to Docker Hub
+        uses: docker/login-action@v3
+        with:
+          username: ${{ secrets.DOCKERUSERNAME }}
+          password: ${{ secrets.DOCKERTOKEN || secrets.DCKRPAT }}
+
+      - name: Publish upstream PyTorch ROCm CI image
+        id: publish
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+          INPUT_COMMIT: ${{ inputs.commit }}
+        run: |
+          args=(
+            --workflow "${SOURCE_WORKFLOW}"
+            --branch "${UPSTREAM_BRANCH}"
+            --image "${REGISTRY}/${IMAGE_NAME}"
+            --tag "${PUSH_TAG}"
+            --push
+          )
+          if [[ -n "${INPUT_COMMIT}" ]]; then
+            args+=(--commit "${INPUT_COMMIT}")
+          fi
+
+          python3 .automation_scripts/rocm_ci_debug_env/pytorch_rocm_ci_debug_env.py publish "${args[@]}"
+
+      - name: Summary
+        if: always()
+        run: |
+          {
+            echo "## pytorch-nightly publish"
+            echo ""
+            echo "| field | value |"
+            echo "|---|---|"
+            echo "| upstream commit | \`${{ steps.publish.outputs.commit }}\` |"
+            echo "| source CI image | \`${{ steps.publish.outputs.image_ref }}\` |"
+            echo "| build label | \`${{ steps.publish.outputs.build_label }}\` |"
+            echo "| pushed tags | \`${{ steps.publish.outputs.primary_tag }}\`, \`${{ steps.publish.outputs.dated_tag }}\` |"
+          } >> "$GITHUB_STEP_SUMMARY"