make the checker/runtime divergence harness ci-robust

- provision the interpreter through uv (3.13) so blocks using modern syntax
  run everywhere, not only where the system python is already new enough
- run the blocks through a bounded worker pool instead of serially, so the
  ~200 transpile/python subprocess pairs don't dominate the linux job
- skip on windows, where subprocess discovery and stdout encoding differ and
  the checker/runtime contract it validates is platform-independent
- exempt it from the ci profile's 60s deadlock cutoff, which it still exceeds
  on a cold 2-core runner even parallelised

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: KotlinIsland

.config/nextest.toml

@@ -22,3 +22,11 @@ slow-timeout =  { period = "1s", terminate-after = 60 }
 
 # Show slow jobs in the final summary
 final-status-level = "slow"
+
+# The divergence harness runs its blocks in parallel, but on a 2-core CI runner
+# it still spends up to a minute or two spawning ~200 transpile/python subprocess
+# pairs (plus a possible one-off uv python download), so exempt it from the
+# aggressive 60s deadlock cutoff that would otherwise kill it mid-run.
+[[profile.ci.overrides]]
+filter = 'test(clean_mdtest_blocks_run)'
+slow-timeout = { period = "60s", terminate-after = 3 }

crates/ty/tests/mdtest_divergence.rs

@@ -14,35 +14,52 @@ use std::fs;
 use std::io::Write;
 use std::path::{Path, PathBuf};
 use std::process::{Command, Stdio};
+use std::sync::Mutex;
+use std::sync::atomic::{AtomicUsize, Ordering};
 
 fn mdtest_dir() -> PathBuf {
     Path::new(env!("CARGO_MANIFEST_DIR")).join("../ty_python_semantic/resources/mdtest")
 }
 
-fn python() -> String {
+/// A CPython 3.13 interpreter, provisioned through uv so the harness runs the
+/// same interpreter everywhere instead of riding on whatever `python3` the host
+/// happens to ship. The transpiler emits modern syntax — PEP 695 generics, PEP
+/// 696 type-parameter defaults (`class C[T = int]`), PEP 646 unpacking — whose
+/// runtime floor is 3.13; CI runners range from 3.10 upward, so a checker-clean
+/// block can fail to even parse on an older interpreter. Returns `None` (the
+/// test then skips) when uv or the interpreter can't be obtained.
+#[cfg(not(windows))]
+fn python() -> Option<String> {
     if let Ok(p) = std::env::var("PYTHON") {
-        return p;
+        return Some(p);
     }
-    // newest available interpreter: blocks exercise modern syntax (`match`,
-    // subclassable `Any`) that the system `python3` may predate
-    for candidate in [
-        "python3.14",
-        "python3.13",
-        "python3.12",
-        "python3.11",
-        "python3.10",
-        "python3",
-    ] {
-        if Command::new(candidate)
-            .arg("-c")
-            .arg("pass")
+    let find = || {
+        let out = Command::new("uv")
+            .args(["python", "find", "3.13"])
             .output()
-            .is_ok_and(|o| o.status.success())
-        {
-            return candidate.to_owned();
-        }
+            .ok()?;
+        out.status
+            .success()
+            .then(|| String::from_utf8_lossy(&out.stdout).trim().to_owned())
+    };
+    if let Some(path) = find() {
+        return Some(path);
     }
-    "python3".to_owned()
+    // not discoverable yet — let uv download a managed build, then locate it
+    Command::new("uv")
+        .args(["python", "install", "3.13"])
+        .output()
+        .ok()?;
+    find()
+}
+
+/// On windows the harness is skipped unless `PYTHON` is set explicitly: it drives
+/// a python subprocess whose interpreter discovery and stdout encoding differ
+/// from unix, and the checker/runtime contract it validates is platform
+/// independent, so unix coverage suffices.
+#[cfg(windows)]
+fn python() -> Option<String> {
+    std::env::var("PYTHON").ok()
 }
 
 /// `major.minor` of the interpreter the blocks will execute on, so the
@@ -59,7 +76,7 @@ fn python_version(python: &str) -> Option<String> {
         .then(|| String::from_utf8_lossy(&output.stdout).trim().to_owned())
 }
 
-/// Extract the ```by fenced blocks of a markdown file, in order, with a flag
+/// Extract the `by` fenced code blocks of a markdown file, in order, with a flag
 /// for blocks living in a multi-file section (one that declares companion
 /// modules via a `` `name.py`: `` marker) — those import section-local modules
 /// and cannot run standalone.
@@ -133,8 +150,15 @@ fn with_reveal_stub(transpiled: &str) -> String {
 }
 
 #[test]
+#[expect(
+    clippy::print_stderr,
+    reason = "skip diagnostic when python is unavailable"
+)]
 fn clean_mdtest_blocks_run() {
-    let python = python();
+    let Some(python) = python() else {
+        eprintln!("skipping: no python 3.13 interpreter available (uv on unix, or set PYTHON)");
+        return;
+    };
     let Some(version) = python_version(&python) else {
         eprintln!("skipping: `{python}` not runnable");
         return;
@@ -147,65 +171,97 @@ fn clean_mdtest_blocks_run() {
         .output()
         .is_ok_and(|o| o.status.success());
 
-    let mut failures: Vec<String> = Vec::new();
-    let mut total = 0usize;
     let dir = mdtest_dir();
     let mut files: Vec<PathBuf> = fs::read_dir(&dir)
         .expect("mdtest dir")
         .filter_map(|e| e.ok().map(|e| e.path()))
         .filter(|p| {
-            p.file_name()
-                .and_then(|n| n.to_str())
-                .is_some_and(|n| n.starts_with("basedpython_") && n.ends_with(".md"))
+            p.extension()
+                .is_some_and(|ext| ext.eq_ignore_ascii_case("md"))
+                && p.file_name()
+                    .and_then(|n| n.to_str())
+                    .is_some_and(|n| n.starts_with("basedpython_"))
         })
         .collect();
     files.sort();
     assert!(!files.is_empty(), "no basedpython mdtests found in {dir:?}");
 
-    let tmp = tempfile::tempdir().expect("tempdir");
-    for file in files {
+    // gather every runnable block up front — skipping ones with expected
+    // diagnostics (only checker-clean blocks carry the contract) or that import
+    // section-local companion modules this harness doesn't materialize — so the
+    // work can be spread across a pool of workers rather than run serially. each
+    // block is an independent `by transpile` + python subprocess pair, so the
+    // harness is dominated by process spawn latency and parallelises cleanly.
+    let mut items: Vec<(String, usize, String)> = Vec::new();
+    for file in &files {
         let name = file.file_name().unwrap().to_string_lossy().into_owned();
-        let markdown = fs::read_to_string(&file).expect("read mdtest");
-        for (i, (block, multi_file)) in by_blocks(&markdown).iter().enumerate() {
-            // a block with expected diagnostics is allowed to misbehave at
-            // runtime (only checker-clean blocks carry the contract), and a
-            // block in a multi-file section imports section-local modules this
-            // harness doesn't materialize
-            if block.contains("# error:") || *multi_file {
+        let markdown = fs::read_to_string(file).expect("read mdtest");
+        for (i, (block, multi_file)) in by_blocks(&markdown).into_iter().enumerate() {
+            if block.contains("# error:") || multi_file {
                 continue;
             }
-            total += 1;
-            let transpiled = match transpile(block, &version) {
-                Ok(t) => t,
-                Err(e) => {
-                    failures.push(format!(
-                        "{name} block {i}: transpile failed:\n{e}\n--- block ---\n{block}"
+            items.push((name.clone(), i, block));
+        }
+    }
+    let total = items.len();
+    assert!(total > 0, "no checker-clean by blocks found");
+
+    let tmp = tempfile::tempdir().expect("tempdir");
+    let failures: Mutex<Vec<String>> = Mutex::new(Vec::new());
+    let next = AtomicUsize::new(0);
+    // each block alternates a cpu-bound `by transpile` with a wait on the python
+    // subprocess, and `by` startup also reads typeshed from disk — so a little
+    // oversubscription past the core count overlaps that i/o and keeps the cores
+    // busy, while the cap bounds how many processes are ever live at once
+    let workers = std::thread::available_parallelism()
+        .map_or(4, |n| n.get() * 2)
+        .clamp(1, 16)
+        .min(total);
+
+    // work-stealing pool: each worker claims the next index and processes it
+    // until the list is drained. the closures borrow the shared state (the
+    // work list, the cursor, the failure sink) so `failures` stays owned for
+    // the drain below. temp files are keyed by `name_i`, unique per block, so
+    // concurrent writes never collide.
+    std::thread::scope(|scope| {
+        for _ in 0..workers {
+            scope.spawn(|| {
+                while let Some((name, i, block)) =
+                    items.get(next.fetch_add(1, Ordering::Relaxed))
+                {
+                    let transpiled = match transpile(block, &version) {
+                        Ok(t) => t,
+                        Err(e) => {
+                            failures.lock().unwrap().push(format!(
+                                "{name} block {i}: transpile failed:\n{e}\n--- block ---\n{block}"
+                            ));
+                            continue;
+                        }
+                    };
+                    if !has_typing_extensions && transpiled.contains("typing_extensions") {
+                        continue;
+                    }
+                    let py = tmp.path().join(format!(
+                        "{}_{i}.py",
+                        name.trim_end_matches(".md").replace('-', "_")
                     ));
-                    continue;
+                    fs::write(&py, with_reveal_stub(&transpiled)).unwrap();
+                    let run = Command::new(&python)
+                        .arg(&py)
+                        .output()
+                        .expect("failed to run python");
+                    if !run.status.success() {
+                        failures.lock().unwrap().push(format!(
+                            "{name} block {i}: checker-clean block crashed at runtime:\n{}\n--- block ---\n{block}",
+                            String::from_utf8_lossy(&run.stderr)
+                        ));
+                    }
                 }
-            };
-            if !has_typing_extensions && transpiled.contains("typing_extensions") {
-                continue;
-            }
-            let py = tmp.path().join(format!(
-                "{}_{i}.py",
-                name.trim_end_matches(".md").replace('-', "_")
-            ));
-            fs::write(&py, with_reveal_stub(&transpiled)).unwrap();
-            let run = Command::new(&python)
-                .arg(&py)
-                .output()
-                .expect("failed to run python");
-            if !run.status.success() {
-                failures.push(format!(
-                    "{name} block {i}: checker-clean block crashed at runtime:\n{}\n--- block ---\n{block}",
-                    String::from_utf8_lossy(&run.stderr)
-                ));
-            }
+            });
         }
-    }
+    });
 
-    assert!(total > 0, "no checker-clean by blocks found");
+    let failures = failures.into_inner().expect("failures mutex poisoned");
     assert!(
         failures.is_empty(),
         "{} of {} checker-clean blocks diverge at runtime:\n\n{}",