fix(ext/node): node:test with watch-mode events (#34254)

## Summary
- Replaces `notImplemented("test.run")` in
`ext/node/polyfills/testing.ts` with a real `TestsStream`-compatible
`Readable`.
- Supports `run({ watch, cwd, signal })`: emits `test:watch:drained`
after each run cycle, and `test:watch:restarted` (followed by `drained`)
when a file changes under the watched cwd.
- The returned stream both pushes events as data chunks (for async
iteration / `'data'` listeners) and emits each event under its `type` as
a named event, matching Node's `TestsStream` semantics so callers can do
`stream.on('test:watch:drained', ...)` directly.
- `AbortSignal` ends the stream and closes the watcher; bursts of fs
events are debounced (~50ms) so a single user-visible change triggers
exactly one restart cycle.

## Scope
This PR implements the watch-mode event machinery only. Programmatic
test discovery / execution under `run()` (which would let it actually
emit `test:pass`/`test:fail`) is deliberately left for a follow-up; the
stream emits empty run cycles for now. That is enough to unblock the
watch-event fixtures in the Node compat suite that drive behavior
through the programmatic API rather than spawning a subprocess.

Five previously-absent test-runner watch fixtures now pass and are added
to \`tests/node_compat/config.jsonc\`:
- \`test-runner/test-run-watch-cwd.mjs\`
- \`test-runner/test-run-watch-cwd-isolation-none.mjs\`
- \`test-runner/test-run-watch-cwd-isolation-process.mjs\`
- \`test-runner/test-run-watch-emit-restarted.mjs\` (validates the
drained/restarted/drained sequence via
\`assert.partialDeepStrictEqual\`)
- \`test-runner/test-run-watch-no-emit-restarted-disabled.mjs\`
(validates that \`watch: false\` never emits \`test:watch:restarted\`
via \`common.mustNotCall\`)

The remaining \`test-run-watch-*\` and \`test-watch-*-isolation-*\`
fixtures depend on subprocess spawning of a \`run()\`-fixture piped
through \`node:test/reporters\` \`tap\`, or on CLI \`--test --watch\`
flags - both substantial separate features and out of scope for this PR.

Co-authored-by: divybot <divybot@users.noreply.github.com>
Co-authored-by: Divy Srivastava <me@littledivy.com>

Author: divybot

ext/node/polyfills/testing.ts

@@ -120,8 +120,142 @@ function getAssertObject() {
   return assertObject;
 }
 
-function run() {
-  notImplemented("test.run");
+// Lazy access to other node polyfills; loading these eagerly at module
+// init causes circular initialization issues during snapshotting.
+let _Readable = null;
+function getReadable() {
+  if (_Readable === null) {
+    _Readable = core.loadExtScript(
+      "ext:deno_node/internal/streams/readable.js",
+    ).Readable;
+  }
+  return _Readable;
+}
+let _fsWatch = null;
+function getFsWatch() {
+  if (_fsWatch === null) {
+    _fsWatch = core.loadExtScript("ext:deno_node/fs.ts").watch;
+  }
+  return _fsWatch;
+}
+const lazyProcess = core.createLazyLoader("node:process");
+
+// node:test `run()` implementation.
+//
+// Returns a `TestsStream`-compatible Readable that emits structured events
+// describing the test run lifecycle. We currently support the watch-mode
+// event stream (`test:watch:drained`, `test:watch:restarted`) which is the
+// minimum required for the Node.js `test-runner/test-run-watch-*` fixtures
+// that drive watch behavior through the programmatic API. Actual test file
+// discovery / execution remains TODO and is gated behind separate work; the
+// stream emits a single empty run cycle, then either ends (watch:false) or
+// waits for filesystem changes to trigger restarts (watch:true).
+//
+// See test-runner/test-run-watch-*.mjs in the Node compat suite for the
+// behavior this implements.
+function run(options) {
+  options = options ?? {};
+  const watch = options.watch === true;
+  const signal = options.signal;
+  let cwd = options.cwd;
+  if (cwd === undefined) {
+    cwd = lazyProcess().default.cwd();
+  }
+
+  const Readable = getReadable();
+  const stream = new Readable({
+    __proto__: null,
+    objectMode: true,
+    // We push events imperatively; the consumer just needs a no-op `_read`.
+    read() {},
+  });
+
+  let watcher = null;
+  let finished = false;
+  let pendingRestartTimer = null;
+
+  function finish() {
+    if (finished) return;
+    finished = true;
+    if (pendingRestartTimer !== null) {
+      clearTimeout(pendingRestartTimer);
+      pendingRestartTimer = null;
+    }
+    if (watcher !== null) {
+      try {
+        watcher.close();
+      } catch { /* ignore */ }
+      watcher = null;
+    }
+    stream.push(null);
+  }
+
+  function emit(type) {
+    if (finished) return;
+    const data = { __proto__: null };
+    // Node's TestsStream emits each lifecycle entry both as a data chunk
+    // (consumed via async iteration / `'data'` listeners) and as a named
+    // event so callers can attach `.on('test:watch:drained', ...)` directly.
+    stream.push({ __proto__: null, type, data });
+    stream.emit(type, data);
+  }
+
+  function drained() {
+    emit("test:watch:drained");
+  }
+
+  function scheduleRestart() {
+    if (finished) return;
+    // Debounce bursts of fs events so a single user-visible change produces
+    // exactly one restart cycle (Node's watcher coalesces likewise).
+    if (pendingRestartTimer !== null) {
+      clearTimeout(pendingRestartTimer);
+    }
+    pendingRestartTimer = setTimeout(() => {
+      pendingRestartTimer = null;
+      if (finished) return;
+      emit("test:watch:restarted");
+      drained();
+    }, 50);
+  }
+
+  if (signal) {
+    if (signal.aborted) {
+      // Resolve the initial drained on next tick to keep callers that
+      // `await once(stream, 'test:watch:drained')` working.
+      queueMicrotask(() => {
+        drained();
+        finish();
+      });
+      return stream;
+    }
+    signal.addEventListener("abort", finish, { once: true });
+  }
+
+  // Emit the initial "drained" event after the current microtask completes
+  // so that consumers attaching `.on('data')` synchronously after `run(...)`
+  // returns still observe the event.
+  queueMicrotask(() => {
+    drained();
+    if (!watch) {
+      finish();
+      return;
+    }
+    try {
+      const fsWatch = getFsWatch();
+      watcher = fsWatch(cwd, { recursive: true }, () => {
+        scheduleRestart();
+      });
+      watcher.on("error", () => {
+        finish();
+      });
+    } catch {
+      // If we can't watch (e.g. cwd doesn't exist), end the stream gracefully.
+      finish();
+    }
+  });
+
+  return stream;
 }
 
 function noop() {}

tests/node_compat/config.jsonc

@@ -4457,6 +4457,11 @@
       "reason": "Runs without the previous missing test.expectFailure TypeError, but reporter snapshot output still differs from Node"
     },
     "test-runner/test-output-typescript-coverage.mjs": {},
+    "test-runner/test-run-watch-cwd-isolation-none.mjs": {},
+    "test-runner/test-run-watch-cwd-isolation-process.mjs": {},
+    "test-runner/test-run-watch-cwd.mjs": {},
+    "test-runner/test-run-watch-emit-restarted.mjs": {},
+    "test-runner/test-run-watch-no-emit-restarted-disabled.mjs": {},
     // TODO(bartlomieju): disabled during work on `node:inspector`, this test didn't actualy run before
     // "sequential/test-watch-mode-inspect.mjs": {},
     // TODO(bartlomieju): this test is very flaky on CI, especially on Windows ARM

tests/specs/node/node_test_run_watch/__test__.jsonc

@@ -0,0 +1,12 @@
+{
+  "tests": {
+    "watch_disabled": {
+      "args": "run -A watch_disabled.mjs",
+      "output": "watch_disabled.out"
+    },
+    "watch_restart": {
+      "args": "run -A watch_restart.mjs",
+      "output": "watch_restart.out"
+    }
+  }
+}

tests/specs/node/node_test_run_watch/watch_disabled.mjs

@@ -0,0 +1,27 @@
+// `run({ watch: false })` should not emit `test:watch:restarted` even when
+// files in the watched cwd change after the run starts.
+import { run } from "node:test";
+import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+const dir = mkdtempSync(join(tmpdir(), "node-test-run-watch-"));
+
+let restarted = false;
+const stream = run({ cwd: dir, watch: false });
+stream.on("test:watch:restarted", () => {
+  restarted = true;
+});
+
+writeFileSync(join(dir, "test.js"), "module.exports = {};");
+
+// eslint-disable-next-line no-unused-vars
+for await (const _ of stream);
+
+rmSync(dir, { recursive: true, force: true });
+
+if (restarted) {
+  console.log("fail: test:watch:restarted was emitted");
+  process.exit(1);
+}
+console.log("ok");

tests/specs/node/node_test_run_watch/watch_disabled.out

@@ -0,0 +1 @@
+ok

tests/specs/node/node_test_run_watch/watch_restart.mjs

@@ -0,0 +1,47 @@
+// `run({ watch: true })` should emit `test:watch:drained` after the initial
+// (empty) run cycle, then emit `test:watch:restarted` followed by another
+// `test:watch:drained` when a file inside the watched cwd changes.
+import { run } from "node:test";
+import { once } from "node:events";
+import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+const dir = mkdtempSync(join(tmpdir(), "node-test-run-watch-"));
+
+const events = [];
+let drainCount = 0;
+const controller = new AbortController();
+
+const stream = run({ cwd: dir, watch: true, signal: controller.signal })
+  .on("data", ({ type }) => {
+    events.push(type);
+    if (type === "test:watch:drained") {
+      drainCount++;
+      if (drainCount >= 2) {
+        controller.abort();
+      }
+    }
+  });
+
+await once(stream, "test:watch:drained");
+writeFileSync(join(dir, "test.js"), "module.exports = {};");
+
+// eslint-disable-next-line no-unused-vars
+for await (const _ of stream);
+
+rmSync(dir, { recursive: true, force: true });
+
+const expected = [
+  "test:watch:drained",
+  "test:watch:restarted",
+  "test:watch:drained",
+];
+for (let i = 0; i < expected.length; i++) {
+  if (events[i] !== expected[i]) {
+    console.log(`fail: events[${i}] = ${events[i]}, expected ${expected[i]}`);
+    console.log(`full events: ${JSON.stringify(events)}`);
+    process.exit(1);
+  }
+}
+console.log("ok");

tests/specs/node/node_test_run_watch/watch_restart.out

@@ -0,0 +1 @@
+ok