feat: 监督者远程模式 TCP 传输与 --remote-supervise 旗标

Cargo.lock

@@ -1160,7 +1160,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_builtins"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "chrono",
  "fast-float",
@@ -1185,7 +1185,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_bytecode"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "num-bigint",
  "oxide_log",
@@ -1193,7 +1193,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_cfg"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "oxide_bytecode",
  "oxide_emit",
@@ -1204,7 +1204,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_cli"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "ansi_term",
  "bumpalo",
@@ -1228,7 +1228,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_code_cache"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "lru",
  "oxide_bytecode",
@@ -1237,7 +1237,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_compiler"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "oxide_bytecode",
  "oxide_cfg",
@@ -1257,7 +1257,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_dce"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "oxide_bytecode",
  "oxide_cfg",
@@ -1271,7 +1271,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_emit"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "num-bigint",
  "oxc_allocator",
@@ -1286,7 +1286,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_engine"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "oxide_bytecode",
  "oxide_compiler",
@@ -1298,7 +1298,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_ir"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "oxide_bytecode",
  "oxide_log",
@@ -1307,7 +1307,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_kernel"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "bumpalo",
  "dashmap",
@@ -1321,7 +1321,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_liveness"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "oxide_bytecode",
  "oxide_cfg",
@@ -1333,7 +1333,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_log"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "tracing",
  "tracing-appender",
@@ -1342,11 +1342,11 @@ dependencies = [
 
 [[package]]
 name = "oxide_net"
-version = "0.15.18"
+version = "0.15.19"
 
 [[package]]
 name = "oxide_parser"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "oxc_allocator",
  "oxc_ast",
@@ -1361,7 +1361,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_regalloc"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "oxide_bytecode",
  "oxide_cfg",
@@ -1377,7 +1377,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_runtime_api"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "num-bigint",
  "num-traits",
@@ -1390,7 +1390,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_test262"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "oxide_bytecode",
  "oxide_compiler",
@@ -1410,7 +1410,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_types"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "bumpalo",
  "dashmap",
@@ -1423,7 +1423,7 @@ dependencies = [
 
 [[package]]
 name = "oxide_vm"
-version = "0.15.18"
+version = "0.15.19"
 dependencies = [
  "bumpalo",
  "chrono",

Cargo.toml

@@ -24,7 +24,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.15.18"
+version = "0.15.19"
 edition = "2021"
 license = "MIT"
 rust-version = "1.80"

crates/oxide_test262/src/config.rs

@@ -12,6 +12,9 @@ pub(crate) struct RunConfig {
     pub(crate) supervise: bool,
     /// 远程分片 worker 模式：连监督者端点收窗口指派帧，in-process 执行并发帧。
     pub(crate) worker: bool,
+    /// 远程监督模式：（端点文本, worker 数）。监督者 bind 端点、每窗口 accept 一条
+    /// worker 连接，经 Tcp 传输执行窗口。
+    pub(crate) remote_supervise: Option<(String, usize)>,
     /// 关闭 liveness/精确 DCE/RegAlloc 链（开关对比：关优化链 vs 开优化链各跑一遍）。
     pub(crate) no_regalloc: bool,
     /// 逐测试打印 PASS/FAIL/SKIP（开关对比时对比两轮结果集合用）。
@@ -29,24 +32,28 @@ impl RunConfig {
             no_skip: false,
             supervise: false,
             worker: false,
+            remote_supervise: None,
             no_regalloc: false,
             verbose: false,
             no_fail_list: false,
         }
     }
 
-    /// 监督模式判定：`--supervise` 且非监督派生的子进程（子进程不重入监督）。
-    /// 路径 filter 不参与判定——filter 在子进程内逐测试应用，不构成监督屏障。
+    /// 监督模式判定：`--supervise` 或 `--remote-supervise` 且非监督派生的子进程
+    /// （子进程不重入监督）。路径 filter 不参与判定——filter 在子进程内逐测试
+    /// 应用，不构成监督屏障。
     pub(crate) fn supervised(&self, is_chunk_child: bool) -> bool {
-        self.supervise && !is_chunk_child
+        (self.supervise || self.remote_supervise.is_some()) && !is_chunk_child
     }
 
-    /// 解析命令行参数为运行配置；未知选项或参数过多返回错误。
+    /// 解析命令行参数为运行配置；未知选项、参数过多或 `--remote-supervise`
+    /// 缺实参 / worker 数非法返回错误。
     pub(crate) fn parse(args: &[String]) -> Result<Self, String> {
         let mut config = Self::new();
         let mut positional = Vec::new();
-
-        for arg in args.iter().skip(1) {
+        let mut i = 1;
+        while i < args.len() {
+            let arg = &args[i];
             match arg.as_str() {
                 "--no-skip" => config.no_skip = true,
                 "--no-regalloc" => config.no_regalloc = true,
@@ -54,10 +61,25 @@ impl RunConfig {
                 "--supervise" => config.supervise = true,
                 "--worker" => config.worker = true,
                 "--no-fail-list" => config.no_fail_list = true,
+                // --remote-supervise 消费后两个实参：端点文本与 worker 数。
+                "--remote-supervise" => {
+                    let endpoint = args
+                        .get(i + 1)
+                        .ok_or_else(|| format!("--remote-supervise requires an endpoint\n\n{}", Self::usage()))?;
+                    let workers_str = args
+                        .get(i + 2)
+                        .ok_or_else(|| format!("--remote-supervise requires a worker count\n\n{}", Self::usage()))?;
+                    let workers = workers_str
+                        .parse::<usize>()
+                        .map_err(|_| format!("invalid worker count: {workers_str}\n\n{}", Self::usage()))?;
+                    config.remote_supervise = Some((endpoint.clone(), workers));
+                    i += 2;
+                }
                 "--help" | "-h" => return Err(Self::usage()),
                 _ if arg.starts_with("--") => return Err(format!("unknown option: {arg}\n\n{}", Self::usage())),
                 _ => positional.push(arg.clone()),
             }
+            i += 1;
         }
 
         // worker 与 supervise 互斥：前者是远程分片的执行半区（连监督者端点），
@@ -65,6 +87,13 @@ impl RunConfig {
         if config.worker && config.supervise {
             return Err(format!("--worker and --supervise are mutually exclusive\n\n{}", Self::usage()));
         }
+        // remote-supervise 是远程监督半区，与 worker（执行半区）及本地 supervise 互斥。
+        if config.worker && config.remote_supervise.is_some() {
+            return Err(format!("--worker and --remote-supervise are mutually exclusive\n\n{}", Self::usage()));
+        }
+        if config.supervise && config.remote_supervise.is_some() {
+            return Err(format!("--supervise and --remote-supervise are mutually exclusive\n\n{}", Self::usage()));
+        }
 
         if let Some(root) = positional.first() {
             config.test262_root = Some(PathBuf::from(root));
@@ -81,7 +110,7 @@ impl RunConfig {
 
     /// 打印用法说明。
     pub(crate) fn usage() -> String {
-        "usage: test262-runner [--no-skip] [--no-regalloc] [--verbose] [--supervise] [--worker] [--version] [test262-root] [path-filter]\n\
+        "usage: test262-runner [--no-skip] [--no-regalloc] [--verbose] [--supervise] [--worker] [--remote-supervise <endpoint> <workers>] [--version] [test262-root] [path-filter]\n\
           \n\
           --no-skip    Run capability-excluded tests and count unsupported compile/runtime results as failures.\n\
           --no-regalloc  Disable the liveness/precise-DCE/RegAlloc compiler chain (vregs stay as physical numbers).\n\
@@ -94,6 +123,11 @@ impl RunConfig {
           --worker     Run as a remote shard worker: connect to the supervisor endpoint and execute assigned\n\
           \x20            test windows in-process, sending heartbeat/fail/scene/DONE frames over the connection.\n\
           \x20            Mutually exclusive with --supervise.\n\
+          --remote-supervise  Run as a remote supervisor: bind <endpoint> (host:port) and execute test\n\
+          \x20            windows on <workers> remote workers, sending assign frames and receiving\n\
+          \x20            heartbeat/fail/scene/DONE frames over the accepted connections. A worker\n\
+          \x20            disconnect is recovered as a child-process death. Mutually exclusive with\n\
+          \x20            --supervise and --worker.\n\
           \n\
           env tunables (all optional):\n\
           \x20  OXIDE_LOG                          per-subsystem log levels (default off)\n\
@@ -163,6 +197,47 @@ mod tests {
         assert!(err.contains("mutually exclusive"), "应提示互斥：{err}");
     }
 
+    /// --remote-supervise 正常解析端点与 worker 数；缺实参、worker 数非法返回错误。
+    #[test]
+    fn parse_remote_supervise_flag() {
+        let cfg = RunConfig::parse(&args(&["--remote-supervise", "127.0.0.1:8080", "4"]))
+            .expect("remote-supervise 应解析成功");
+        assert_eq!(cfg.remote_supervise, Some(("127.0.0.1:8080".to_string(), 4)));
+        assert!(!cfg.supervise, "remote-supervise 不应置位 supervise");
+        assert!(!cfg.worker, "remote-supervise 不应置位 worker");
+
+        let err = RunConfig::parse(&args(&["--remote-supervise"])).expect_err("缺端点应报错");
+        assert!(err.contains("--remote-supervise requires an endpoint"), "应提示缺端点：{err}");
+
+        let err = RunConfig::parse(&args(&["--remote-supervise", "127.0.0.1:8080"])).expect_err("缺 worker 数应报错");
+        assert!(err.contains("--remote-supervise requires a worker count"), "应提示缺 worker 数：{err}");
+
+        let err =
+            RunConfig::parse(&args(&["--remote-supervise", "127.0.0.1:8080", "abc"])).expect_err("worker 数非法应报错");
+        assert!(err.contains("invalid worker count"), "应提示 worker 数非法：{err}");
+    }
+
+    /// --remote-supervise 与 --worker、--supervise 互斥。
+    #[test]
+    fn parse_remote_supervise_exclusive() {
+        let err = RunConfig::parse(&args(&["--remote-supervise", "127.0.0.1:8080", "4", "--worker"]))
+            .expect_err("worker+remote-supervise 应互斥");
+        assert!(err.contains("mutually exclusive"), "应提示互斥：{err}");
+
+        let err = RunConfig::parse(&args(&["--remote-supervise", "127.0.0.1:8080", "4", "--supervise"]))
+            .expect_err("supervise+remote-supervise 应互斥");
+        assert!(err.contains("mutually exclusive"), "应提示互斥：{err}");
+    }
+
+    /// remote_supervise 置位也触发 supervised()（非子进程）；子进程不重入。
+    #[test]
+    fn supervised_predicate_with_remote_supervise() {
+        let mut cfg = RunConfig::new();
+        cfg.remote_supervise = Some(("127.0.0.1:8080".to_string(), 4));
+        assert!(cfg.supervised(false), "remote_supervise 应触发 supervised");
+        assert!(!cfg.supervised(true), "子进程不应重入 supervised");
+    }
+
     /// 未知选项返回含 "unknown option" 的错误；--help 的错误文案与 usage 逐字相等。
     #[test]
     fn parse_unknown_option_and_help_err_text() {

crates/oxide_test262/src/main.rs

@@ -16,6 +16,7 @@ mod runner;
 mod shard_protocol;
 mod stats;
 mod supervise;
+mod tcp_window_transport;
 mod test262_log;
 mod window_transport;
 mod worker;
@@ -311,7 +312,7 @@ fn run_tests() -> bool {
     if config.supervised(is_chunk_child) {
         test262_info!("supervised mode enabled");
         eprintln!("supervised mode enabled: child-window execution with per-test timeout + auto-resume");
-        return run_supervised(&args, skip_until, end_index, config.no_skip, &paths);
+        return run_supervised(&args, skip_until, end_index, config.no_skip, &paths, config.remote_supervise.clone());
     }
 
     if let Some(chunk_size) = chunk_size {

crates/oxide_test262/src/supervise.rs

@@ -9,9 +9,12 @@ use crate::heartbeat::merge_heartbeat;
 use crate::print_fail_categories;
 use crate::report::{format_fail_groupings, format_fail_list};
 use crate::stats::RunStats;
+use crate::tcp_window_transport::TcpWindowTransport;
 use crate::window_transport::{LocalFileWindowTransport, PollResult, WindowTransport};
+use oxide_net::Listener;
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 /// 超时/崩溃后把 sidecar 日志的末 ~20 行打到父进程 stderr，供现场重建。
@@ -186,6 +189,7 @@ fn supervise_window(
 /// 从不持有 `KernelCore`，因此按引用共享 `paths`/`args` 是安全的。
 pub(crate) fn run_supervised(
     args: &[String], skip_until: usize, end_index: usize, no_skip: bool, paths: &[PathBuf],
+    remote_supervise: Option<(String, usize)>,
 ) -> bool {
     let exe = match std::env::current_exe() {
         Ok(path) => path,
@@ -195,14 +199,19 @@ pub(crate) fn run_supervised(
         }
     };
 
-    // 路径清单文件：派生窗口前写一次（每行一路径，与父扫描同序），跨窗口只读
-    // 共享；子进程读文件代替全量重扫目录。运行结束删除，防临时目录残留。
-    let paths_file = std::env::temp_dir().join(format!("oxide_t262_paths_{}.txt", std::process::id()));
-    let list = paths.iter().map(|p| p.to_string_lossy()).collect::<Vec<_>>().join("\n");
-    if let Err(err) = std::fs::write(&paths_file, list) {
-        eprintln!("failed to write paths file {}: {err}", paths_file.display());
-        return false;
-    }
+    // 路径清单文件：仅本地模式需要（子进程读文件代替全量重扫目录）；远程模式
+    // 路径随 assign 帧携带，不写文件。运行结束删除，防临时目录残留。
+    let paths_file: Option<PathBuf> = if remote_supervise.is_none() {
+        let pf = std::env::temp_dir().join(format!("oxide_t262_paths_{}.txt", std::process::id()));
+        let list = paths.iter().map(|p| p.to_string_lossy()).collect::<Vec<_>>().join("\n");
+        if let Err(err) = std::fs::write(&pf, list) {
+            eprintln!("failed to write paths file {}: {err}", pf.display());
+            return false;
+        }
+        Some(pf)
+    } else {
+        None
+    };
 
     let window = std::env::var("OXIDE_TEST262_WINDOW")
         .ok()
@@ -234,68 +243,140 @@ pub(crate) fn run_supervised(
         id += 1;
     }
 
-    // 缺省 16：全量复测验证过的安全并发；未经复测不得抬高。
-    let supervisors = std::env::var("OXIDE_TEST262_SUPERVISORS")
-        .ok()
-        .and_then(|s| s.parse::<usize>().ok())
-        .filter(|&n| n > 0)
-        .unwrap_or(16)
-        .min(windows.len().max(1));
-
-    eprintln!(
-        "supervised mode: {} window(s) of up to {window} test(s), {supervisors} supervisor(s), per-test timeout {}s",
-        windows.len(),
-        timeout.as_secs()
-    );
+    // 缺省 16：全量复测验证过的安全并发；未经复测不得抬高。远程模式用旗标 worker 数。
+    let (supervisors, endpoint) = match &remote_supervise {
+        Some((endpoint, workers)) => ((*workers).min(windows.len().max(1)), Some(endpoint.clone())),
+        None => (
+            std::env::var("OXIDE_TEST262_SUPERVISORS")
+                .ok()
+                .and_then(|s| s.parse::<usize>().ok())
+                .filter(|&n| n > 0)
+                .unwrap_or(16)
+                .min(windows.len().max(1)),
+            None,
+        ),
+    };
 
     let next = AtomicUsize::new(0);
     let next = &next;
     let windows = &windows;
     let exe = &exe;
-    let paths_file = &paths_file;
 
-    let partials: Vec<RunStats> = std::thread::scope(|scope| {
-        let handles: Vec<_> = (0..supervisors)
-            .map(|_| {
-                scope.spawn(move || {
-                    let mut stats = RunStats::default();
-                    loop {
-                        let wi = next.fetch_add(1, Ordering::Relaxed);
-                        if wi >= windows.len() {
-                            break;
-                        }
-                        let (window_id, wstart, wend) = windows[wi];
-                        // 每窗口一个传输实例（跨该窗口内子进程重启存活），
-                        // 已消费偏移等窗口级状态随之保留。
-                        let window_stats = supervise_window(
-                            no_skip,
-                            wstart,
-                            wend,
-                            timeout,
-                            startup_grace,
-                            paths,
-                            window_id,
-                            &mut LocalFileWindowTransport::new(exe, args, paths_file, window_id),
-                        );
-                        stats.merge(window_stats);
-                    }
-                    stats
-                })
+    let partials: Vec<RunStats> = match &endpoint {
+        Some(endpoint) => {
+            // 远程分支：bind 监听端点，每窗口 accept 一条 worker 连接，worker 数控制
+            // 并发线程数。路径随 assign 帧携带，不写 paths_file。
+            let listener = match std::net::TcpListener::bind(endpoint) {
+                Ok(l) => Arc::new(Listener::Tcp(l)),
+                Err(err) => {
+                    eprintln!("failed to bind remote supervisor endpoint {endpoint}: {err}");
+                    return false;
+                }
+            };
+            let bound = listener.local_addr().unwrap_or_else(|_| endpoint.clone());
+            eprintln!(
+                "remote supervised mode: {} window(s) of up to {window} test(s), {supervisors} worker(s), endpoint {bound}, per-test timeout {}s",
+                windows.len(),
+                timeout.as_secs()
+            );
+            // 路径清单经 Arc 跨监督线程共享（每窗口克隆一份随 assign 帧携带）。
+            let paths_vec: Arc<Vec<String>> =
+                Arc::new(paths.iter().map(|p| p.to_string_lossy().into_owned()).collect());
+            let timeout_secs = timeout.as_secs();
+            let listener_ref = &listener;
+            let paths_vec_ref = &paths_vec;
+            std::thread::scope(|scope| {
+                let handles: Vec<_> = (0..supervisors)
+                    .map(|_| {
+                        scope.spawn(move || {
+                            let mut stats = RunStats::default();
+                            loop {
+                                let wi = next.fetch_add(1, Ordering::Relaxed);
+                                if wi >= windows.len() {
+                                    break;
+                                }
+                                let (window_id, wstart, wend) = windows[wi];
+                                // 每窗口一个传输实例（跨该窗口内 worker 重启存活）。
+                                let window_stats = supervise_window(
+                                    no_skip,
+                                    wstart,
+                                    wend,
+                                    timeout,
+                                    startup_grace,
+                                    paths,
+                                    window_id,
+                                    &mut TcpWindowTransport::new(
+                                        Arc::clone(listener_ref),
+                                        (**paths_vec_ref).clone(),
+                                        timeout_secs,
+                                    ),
+                                );
+                                stats.merge(window_stats);
+                            }
+                            stats
+                        })
+                    })
+                    .collect();
+                handles
+                    .into_iter()
+                    .map(|h| h.join().expect("supervisor thread panicked"))
+                    .collect()
             })
-            .collect();
-        handles
-            .into_iter()
-            .map(|h| h.join().expect("supervisor thread panicked"))
-            .collect()
-    });
+        }
+        None => {
+            // 本地分支：现状不变（写 paths_file、LocalFileWindowTransport、缺省 16）。
+            let paths_file = paths_file.as_ref().expect("本地模式已写 paths_file");
+            eprintln!(
+                "supervised mode: {} window(s) of up to {window} test(s), {supervisors} supervisor(s), per-test timeout {}s",
+                windows.len(),
+                timeout.as_secs()
+            );
+            std::thread::scope(|scope| {
+                let handles: Vec<_> = (0..supervisors)
+                    .map(|_| {
+                        scope.spawn(move || {
+                            let mut stats = RunStats::default();
+                            loop {
+                                let wi = next.fetch_add(1, Ordering::Relaxed);
+                                if wi >= windows.len() {
+                                    break;
+                                }
+                                let (window_id, wstart, wend) = windows[wi];
+                                // 每窗口一个传输实例（跨该窗口内子进程重启存活），
+                                // 已消费偏移等窗口级状态随之保留。
+                                let window_stats = supervise_window(
+                                    no_skip,
+                                    wstart,
+                                    wend,
+                                    timeout,
+                                    startup_grace,
+                                    paths,
+                                    window_id,
+                                    &mut LocalFileWindowTransport::new(exe, args, paths_file, window_id),
+                                );
+                                stats.merge(window_stats);
+                            }
+                            stats
+                        })
+                    })
+                    .collect();
+                handles
+                    .into_iter()
+                    .map(|h| h.join().expect("supervisor thread panicked"))
+                    .collect()
+            })
+        }
+    };
 
     let mut stats = RunStats::default();
     for partial_stats in partials {
         stats.merge(partial_stats);
     }
 
-    // 全部窗口结束、子进程已读完清单，删除路径清单文件防临时目录残留。
-    let _ = std::fs::remove_file(paths_file);
+    // 全部窗口结束、子进程已读完清单，删除路径清单文件防临时目录残留（远程模式无文件）。
+    if let Some(pf) = &paths_file {
+        let _ = std::fs::remove_file(pf);
+    }
 
     let total = stats.pass + stats.fail + stats.skip;
     println!();

crates/oxide_test262/src/tcp_window_transport.rs

@@ -0,0 +1,425 @@
+//! 远程监督窗口传输：`WindowTransport` 的 Tcp 臂。监督者 `accept` 一条 worker
+//! 连接，经该连接发 assign 帧、收事件帧；断连（EOF / 真实 io 错误 / 畸形帧）
+//! 映射为 `PollResult::Exited`（合成退出码），走与本地子进程死亡完全同构的恢复
+//! 路径。
+//!
+//! 写读侧分离：`writer` 是 `try_clone` 出的写侧句柄，`reader` 包住底层连接读侧
+//! （含余量缓冲）。`kill` 后须 drop 两侧句柄真正断连，否则连接半开。
+//!
+//! 计数合并契约：监督者每窗口只合并一次最终心跳（DONE，带全窗累计计数）；
+//! 中间 COMPLETED 心跳只用于超时跟踪（last_index），不合并计数。`Done` 帧把
+//! `last_heartbeat` 的 phase 改 DONE 并保留计数；worker 发 Done 前无 Heartbeat
+//! 时合成零计数 DONE。
+//!
+//! 非阻塞读陷阱：`set_read_timeout(0)` 后 `WouldBlock` 必须判「无数据、仍存活」
+//! 而非断连，否则正常窗口被误杀重启；仅真实 io 错误与 EOF 判断连。残行留在
+//! `FrameReader` 余量缓冲，跨 poll 不丢。
+
+use std::collections::HashMap;
+use std::io::BufReader;
+use std::io::ErrorKind;
+use std::sync::Arc;
+
+use crate::heartbeat::Heartbeat;
+use crate::shard_protocol::{encode_command, parse_event, FrameReader, ShardCommand, ShardEvent};
+use crate::stats::RunStats;
+use crate::window_transport::{PollResult, SpawnError, WindowTransport};
+use oxide_net::{Conn, Listener};
+
+/// 远程监督窗口传输：`WindowTransport` 的 Tcp 臂。`listener` 供跨子进程重启
+/// 重新 `accept`；`writer` / `reader` 是当前 worker 连接的写读两侧句柄。
+pub(crate) struct TcpWindowTransport {
+    listener: Arc<Listener>,
+    /// 写侧句柄（`try_clone` 出，与读侧共享底层套接字）。
+    writer: Option<Conn>,
+    /// 读侧（含余量缓冲，跨 poll 保留残行）。
+    reader: Option<FrameReader<BufReader<Conn>>>,
+    /// 最近一次心跳帧（中间 COMPLETED 只跟踪不合并；Done 帧改 phase 为 DONE）。
+    last_heartbeat: Option<Heartbeat>,
+    /// 待并入的失败记录帧（一测试失败一帧，按序消费，无偏移态，天然无双计）。
+    pending_fails: Vec<(usize, String, String, String)>,
+    /// 最近一次现场帧（last-pc 现场行加可选日志尾）。
+    last_scene: Option<(String, Option<String>)>,
+    /// 全量路径文本（assign 时切窗口片随帧携带，worker 读不到监督者本机文件）。
+    paths: Vec<String>,
+    /// 单测试超时，随 assign 帧携带供 worker 侧看门狗使用。
+    timeout_secs: u64,
+}
+
+impl TcpWindowTransport {
+    /// 构造传输：`listener` 是监督者绑定端点后的监听句柄（跨重启 `accept` 用），
+    /// `paths` 是全量路径文本，`timeout_secs` 随 assign 帧携带。
+    pub(crate) fn new(listener: Arc<Listener>, paths: Vec<String>, timeout_secs: u64) -> Self {
+        Self {
+            listener,
+            writer: None,
+            reader: None,
+            last_heartbeat: None,
+            pending_fails: Vec::new(),
+            last_scene: None,
+            paths,
+            timeout_secs,
+        }
+    }
+
+    /// 写读两侧句柄都 drop，真正断连（防连接半开）。
+    fn drop_conn(&mut self) {
+        self.writer = None;
+        self.reader = None;
+    }
+
+    /// Done 帧处理：把 `last_heartbeat` 的 phase 改 DONE 并保留计数（每窗口只
+    /// 合并一次的计数契约），随后断连判死亡。
+    fn mark_done(&mut self) {
+        match self.last_heartbeat.as_mut() {
+            Some(hb) => hb.phase = "DONE".to_string(),
+            None => {
+                // worker 发 Done 前无 Heartbeat：合成零计数 DONE。
+                self.last_heartbeat = Some(Heartbeat {
+                    phase: "DONE".to_string(),
+                    index: 0,
+                    pass: 0,
+                    fail: 0,
+                    skip: 0,
+                    categories: HashMap::new(),
+                    hb_write_errors: 0,
+                });
+            }
+        }
+        self.drop_conn();
+    }
+}
+
+impl WindowTransport for TcpWindowTransport {
+    fn assign(&mut self, wstart: usize, wend: usize) -> Result<(), SpawnError> {
+        // 无连接时阻塞 `accept` 取一条 worker 连接（io 错误包成 SpawnError）。
+        if self.writer.is_none() {
+            let base = self.listener.accept().map_err(SpawnError)?;
+            // 读侧置非阻塞：std 拒绝 0 时长读超时，改用 `set_nonblocking` 使无数据
+            // 读返回 `WouldBlock`（判「无数据、仍存活」，不误杀正常窗口）。
+            match &base {
+                Conn::Tcp(s) => s.set_nonblocking(true).map_err(SpawnError)?,
+                Conn::Unix(s) => s.set_nonblocking(true).map_err(SpawnError)?,
+            }
+            let writer = base.try_clone().map_err(SpawnError)?;
+            let reader = FrameReader::new(BufReader::new(base));
+            self.writer = Some(writer);
+            self.reader = Some(reader);
+        }
+
+        // 发 assign 帧：窗口路径片随帧携带（worker 读不到监督者本机临时文件）。
+        let window_paths = self.paths[wstart..wend].to_vec();
+        let cmd = ShardCommand::Assign {
+            wstart,
+            wend,
+            paths: window_paths,
+            timeout_secs: self.timeout_secs,
+        };
+        let frame = encode_command(&cmd);
+        let writer = self.writer.as_mut().expect("assign 刚建立写侧句柄");
+        writer.write_all(frame.as_bytes()).map_err(SpawnError)?;
+        Ok(())
+    }
+
+    fn heartbeat(&self) -> Option<Heartbeat> {
+        // `Heartbeat` 未派生 `Clone`（心跳模块零改动），逐字段手工克隆。
+        self.last_heartbeat.as_ref().map(|hb| Heartbeat {
+            phase: hb.phase.clone(),
+            index: hb.index,
+            pass: hb.pass,
+            fail: hb.fail,
+            skip: hb.skip,
+            categories: hb.categories.clone(),
+            hb_write_errors: hb.hb_write_errors,
+        })
+    }
+
+    fn merge_fail(&mut self, stats: &mut RunStats) {
+        // 排空待并入失败帧逐条重建 FailRecord（帧按序消费，无偏移态，无双计）。
+        for (index, category, subkey, message) in self.pending_fails.drain(..) {
+            stats.push_fail_record(index, category, subkey, message);
+        }
+    }
+
+    fn scene(&self) -> Option<String> {
+        self.last_scene.as_ref().map(|(line, _)| line.clone())
+    }
+
+    fn kill(&mut self) {
+        // 发 kill 帧后 drop 写读两侧句柄真正断连（否则连接半开）。
+        if let Some(writer) = self.writer.as_mut() {
+            let frame = encode_command(&ShardCommand::Kill);
+            let _ = writer.write_all(frame.as_bytes());
+        }
+        self.drop_conn();
+    }
+
+    fn poll(&mut self) -> PollResult {
+        // 先取帧（读侧借用在此结束，帧是拥有值），再改状态，避免借用冲突。
+        let frame_result = match self.reader.as_mut() {
+            Some(reader) => reader.read_frame(),
+            None => return PollResult::Alive,
+        };
+        match frame_result {
+            Ok(Some(frame)) => match parse_event(&frame) {
+                Ok(ShardEvent::Heartbeat {
+                    phase,
+                    index,
+                    pass,
+                    fail,
+                    skip,
+                    categories,
+                    hb_write_errors,
+                }) => {
+                    // 中间心跳只跟踪不合并计数（合并契约：每窗口只合并一次 DONE）。
+                    self.last_heartbeat = Some(Heartbeat {
+                        phase,
+                        index,
+                        pass,
+                        fail,
+                        skip,
+                        categories,
+                        hb_write_errors,
+                    });
+                }
+                Ok(ShardEvent::Fail {
+                    index,
+                    category,
+                    subkey,
+                    message,
+                }) => {
+                    self.pending_fails.push((index, category, subkey, message));
+                }
+                Ok(ShardEvent::Scene { line, log_tail }) => {
+                    self.last_scene = Some((line, log_tail));
+                }
+                Ok(ShardEvent::Done) => {
+                    // 窗口结束：改末次心跳 phase 为 DONE 保留计数，断连判死亡
+                    // （与本地子进程退出同构）。
+                    self.mark_done();
+                    return PollResult::Exited(1);
+                }
+                Err(_) => {
+                    // 畸形帧：对端状态不定，判断连（同 EOF 同构恢复）。
+                    self.drop_conn();
+                    return PollResult::Exited(1);
+                }
+            },
+            Ok(None) => {
+                // EOF：对端关闭，判死亡。
+                self.drop_conn();
+                return PollResult::Exited(1);
+            }
+            Err(e) if e.kind() == ErrorKind::WouldBlock => {
+                // 非阻塞读无数据：仍存活，下一拍重试（不得误判断连）。
+                return PollResult::Alive;
+            }
+            Err(_) => {
+                // 真实 io 错误：判断连。
+                self.drop_conn();
+                return PollResult::Exited(1);
+            }
+        }
+        PollResult::Alive
+    }
+
+    fn log_tail(&self) -> Option<String> {
+        self.last_scene.as_ref().and_then(|(_, log_tail)| log_tail.clone())
+    }
+
+    fn reset_round(&mut self) {
+        // 清事件缓冲（心跳 / 待并入失败帧 / 现场归零），跨重启不保留。
+        self.last_heartbeat = None;
+        self.pending_fails.clear();
+        self.last_scene = None;
+    }
+
+    fn archive_window(&mut self) {
+        // 远程无文件归档：失败记录已并入 stats，无数据丢失；drop 连接即收尾。
+        self.drop_conn();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Write;
+    use std::net::TcpStream;
+    use std::time::{Duration, Instant};
+
+    use crate::shard_protocol::encode_event;
+
+    /// TcpWindowTransport 回环：bind 动态端口，客户端连接后发 Heartbeat / Fail /
+    /// Scene 帧，断言 heartbeat / scene / log_tail / merge_fail 读回正确；
+    /// 客户端断连后 poll 返回 Exited（合成码）。
+    #[test]
+    fn tcp_transport_loopback() {
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind 失败");
+        let addr = listener.local_addr().expect("local_addr 失败");
+        let listener = Arc::new(Listener::Tcp(listener));
+
+        let mut transport =
+            TcpWindowTransport::new(Arc::clone(&listener), vec!["a/b.js".to_string(), "c/d.js".to_string()], 10);
+
+        // assign：accept 客户端连接并发 assign 帧。
+        let mut client = TcpStream::connect(addr).expect("客户端连接失败");
+        transport.assign(0, 2).expect("assign 应成功");
+
+        // 发 Heartbeat / Fail / Scene 帧（按序，一测试失败一帧）。
+        client
+            .write_all(
+                encode_event(&ShardEvent::Heartbeat {
+                    phase: "COMPLETED".into(),
+                    index: 0,
+                    pass: 1,
+                    fail: 0,
+                    skip: 0,
+                    categories: HashMap::new(),
+                    hb_write_errors: 0,
+                })
+                .as_bytes(),
+            )
+            .expect("写心跳帧失败");
+        client
+            .write_all(
+                encode_event(&ShardEvent::Fail {
+                    index: 0,
+                    category: "vm: not callable".into(),
+                    subkey: "x".into(),
+                    message: "x is not callable".into(),
+                })
+                .as_bytes(),
+            )
+            .expect("写失败帧失败");
+        client
+            .write_all(
+                encode_event(&ShardEvent::Scene {
+                    line: "pc=1 op=ADD frames=1".into(),
+                    log_tail: Some("log line".into()),
+                })
+                .as_bytes(),
+            )
+            .expect("写现场帧失败");
+
+        // 读帧：poll 逐帧消费，直到 Scene（末帧）读回。
+        let deadline = Instant::now() + Duration::from_secs(5);
+        while transport.scene().is_none() && Instant::now() < deadline {
+            match transport.poll() {
+                PollResult::Exited(_) | PollResult::WaitError(_) => break,
+                PollResult::Alive => {}
+            }
+            std::thread::sleep(Duration::from_millis(10));
+        }
+        assert!(transport.scene().is_some(), "现场帧应读回");
+        let hb = transport.heartbeat().expect("心跳帧应读回");
+        assert_eq!(hb.phase, "COMPLETED");
+        assert_eq!(hb.index, 0);
+        assert_eq!(hb.pass, 1);
+        assert_eq!(transport.scene(), Some("pc=1 op=ADD frames=1".to_string()));
+        assert_eq!(transport.log_tail(), Some("log line".to_string()));
+
+        // merge_fail 排空待并入失败帧；二次合并不重复（无偏移态，排空即空）。
+        let mut stats = RunStats::default();
+        transport.merge_fail(&mut stats);
+        assert_eq!(stats.fail_records.len(), 1, "应并入一条失败记录");
+        assert_eq!(stats.fail_records[0].index, 0);
+        transport.merge_fail(&mut stats);
+        assert_eq!(stats.fail_records.len(), 1, "二次合并不应重复");
+
+        // 客户端断连：poll 返回 Exited（合成码）。
+        drop(client);
+        let mut exited = false;
+        let deadline = Instant::now() + Duration::from_secs(5);
+        while !exited && Instant::now() < deadline {
+            if matches!(transport.poll(), PollResult::Exited(_)) {
+                exited = true;
+            }
+            std::thread::sleep(Duration::from_millis(10));
+        }
+        assert!(exited, "断连后 poll 应返回 Exited");
+    }
+
+    /// Done 帧：把末次心跳 phase 改 DONE 保留计数，poll 返回 Exited；worker 发
+    /// Done 前无 Heartbeat 时合成零计数 DONE。
+    #[test]
+    fn tcp_transport_done_frame_keeps_counts() {
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind 失败");
+        let addr = listener.local_addr().expect("local_addr 失败");
+        let listener = Arc::new(Listener::Tcp(listener));
+
+        let mut transport = TcpWindowTransport::new(Arc::clone(&listener), vec!["a/b.js".to_string()], 10);
+        let mut client = TcpStream::connect(addr).expect("客户端连接失败");
+        transport.assign(0, 1).expect("assign 应成功");
+
+        // 先发一条带计数的心跳，再发 Done：Done 应把 phase 改 DONE 并保留计数。
+        client
+            .write_all(
+                encode_event(&ShardEvent::Heartbeat {
+                    phase: "COMPLETED".into(),
+                    index: 0,
+                    pass: 5,
+                    fail: 2,
+                    skip: 1,
+                    categories: HashMap::new(),
+                    hb_write_errors: 0,
+                })
+                .as_bytes(),
+            )
+            .expect("写心跳帧失败");
+        let deadline = Instant::now() + Duration::from_secs(5);
+        while transport.heartbeat().is_none() && Instant::now() < deadline {
+            let _ = transport.poll();
+            std::thread::sleep(Duration::from_millis(10));
+        }
+        client
+            .write_all(encode_event(&ShardEvent::Done).as_bytes())
+            .expect("写 Done 帧失败");
+
+        // poll 读到 Done 帧应返回 Exited，且心跳 phase 改 DONE、计数保留。
+        let mut exited = false;
+        let deadline = Instant::now() + Duration::from_secs(5);
+        while !exited && Instant::now() < deadline {
+            if matches!(transport.poll(), PollResult::Exited(_)) {
+                exited = true;
+            }
+            std::thread::sleep(Duration::from_millis(10));
+        }
+        assert!(exited, "Done 帧后 poll 应返回 Exited");
+        let hb = transport.heartbeat().expect("Done 后心跳应可读回");
+        assert_eq!(hb.phase, "DONE", "Done 帧应把 phase 改 DONE");
+        assert_eq!(hb.pass, 5, "Done 帧应保留末次心跳计数");
+        assert_eq!(hb.fail, 2);
+        assert_eq!(hb.skip, 1);
+    }
+
+    /// Done 帧前无 Heartbeat：合成零计数 DONE（不 panic、计数全零）。
+    #[test]
+    fn tcp_transport_done_without_heartbeat_synthesizes_zero() {
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind 失败");
+        let addr = listener.local_addr().expect("local_addr 失败");
+        let listener = Arc::new(Listener::Tcp(listener));
+
+        let mut transport = TcpWindowTransport::new(Arc::clone(&listener), vec!["a/b.js".to_string()], 10);
+        let mut client = TcpStream::connect(addr).expect("客户端连接失败");
+        transport.assign(0, 1).expect("assign 应成功");
+
+        client
+            .write_all(encode_event(&ShardEvent::Done).as_bytes())
+            .expect("写 Done 帧失败");
+        let mut exited = false;
+        let deadline = Instant::now() + Duration::from_secs(5);
+        while !exited && Instant::now() < deadline {
+            if matches!(transport.poll(), PollResult::Exited(_)) {
+                exited = true;
+            }
+            std::thread::sleep(Duration::from_millis(10));
+        }
+        assert!(exited, "Done 帧后 poll 应返回 Exited");
+        let hb = transport.heartbeat().expect("无心跳 Done 应合成零计数 DONE");
+        assert_eq!(hb.phase, "DONE");
+        assert_eq!(hb.pass, 0);
+        assert_eq!(hb.fail, 0);
+        assert_eq!(hb.skip, 0);
+        assert!(hb.categories.is_empty());
+    }
+}

crates/oxide_test262/src/window_transport.rs

@@ -10,10 +10,11 @@ use crate::heartbeat::{read_heartbeat, Heartbeat};
 use crate::report::parse_fail_log;
 use crate::stats::RunStats;
 use std::path::{Path, PathBuf};
-use std::process::{Child, Command, ExitStatus};
+use std::process::{Child, Command};
 
 /// 派生窗口子进程失败：spawn 的 io 错误。
-pub(crate) struct SpawnError(std::io::Error);
+#[derive(Debug)]
+pub(crate) struct SpawnError(pub(crate) std::io::Error);
 
 impl std::fmt::Display for SpawnError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@@ -21,11 +22,12 @@ impl std::fmt::Display for SpawnError {
     }
 }
 
-/// 子进程等待三态：死亡（带退出状态）/ 存活 / 等待错误。
+/// 子进程等待三态：死亡（带退出码）/ 存活 / 等待错误。
 /// 三臂在监督策略里处理不同（死亡走心跳裁决、存活继续轮询、错误走 kill 加
-/// 记账），不得压扁。
+/// 记账），不得压扁。退出码仅用于打印（`ExitStatus` 无公开构造器，远程断连
+/// 合成码 1 走同一打印路径，不参与逻辑分支）。
 pub(crate) enum PollResult {
-    Exited(ExitStatus),
+    Exited(usize),
     Alive,
     WaitError(std::io::Error),
 }
@@ -150,7 +152,8 @@ impl WindowTransport for LocalFileWindowTransport {
             return PollResult::Alive;
         };
         match child.try_wait() {
-            Ok(Some(status)) => PollResult::Exited(status),
+            // 退出码仅用于打印：正常退出取 code，信号杀死取缺省 1。
+            Ok(Some(status)) => PollResult::Exited(status.code().unwrap_or(1) as usize),
             Ok(None) => PollResult::Alive,
             Err(err) => PollResult::WaitError(err),
         }