diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 0899e4d8..1bae92c2 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -7,6 +7,9 @@ this package adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ## [Unreleased] +- Route result put/page/release and batch plan/run/resume through the resident kernel on Windows, preserving the exact public confinement command and scrubbed runner environment. Full-access calls reuse the existing fixed worker on every platform; restricted POSIX calls retain their explicit platform runner until native range ownership is available. Admission failure never falls back or replays (#872). +- Windows 的 result put/page/release 与 batch plan/run/resume 改由常驻内核执行，保留公开 confinement 的原始命令及清理后的 runner 环境。所有平台的完全访问调用复用既有固定 worker；受限 POSIX 调用在原生进程范围所有权可用前保留明确的平台 runner。准入失败不回退、不重放（#872）。 + - Connect explicit session Bash results to sealed spill recovery, require `shell-spill-v1` before public activation, and bind retained session callbacks to their exact publishing Agent. Fresh and session projections share the existing root-owned kernel and preserve their separate semantics; accepted runtime distribution remains pending (#872). - 将显式会话 Bash 结果接入已封存的 spill 恢复，公开启用前要求 `shell-spill-v1`，并把保留的会话回调绑定到发布它的准确 Agent。Fresh 与会话投影共用既有根作用域内核，保留各自语义；合格运行时分发仍待完成（#872）。 diff --git a/docs/resident-confined-workers.md b/docs/resident-confined-workers.md new file mode 100644 index 00000000..e3551bfc --- /dev/null +++ b/docs/resident-confined-workers.md @@ -0,0 +1,52 @@ +# Resident result and batch execution + +[中文](常驻结果与批处理.md) + +Native and PTC result put/page/release share the same bridge. PTC batch plan/run/resume +uses that bridge too. On Windows, all of these calls now use the owning resident kernel; +the public subprocess service starts the kernel once, not another OPS worker per call. +The original reference/audit implementations, IDs, result budgets, write-ahead records, +root/session binding and no-follow checks remain unchanged. + +The permission boundary is explicit: + +| Current policy/platform | Execution path | +| --- | --- | +| Full access, every supported platform | Existing trusted fixed worker leaf inside the resident kernel. | +| Read-only or workspace-write, Windows | Exact public `sandbox.confine()` command, launched inside the existing atomic Windows Job range. | +| Read-only or workspace-write, POSIX | Existing public confined runner. Native arbitrary-runner ownership is not established on this platform; this remaining route is not a kernel-error fallback. | + +Batch operations are data, but their existing root-relative path checks are not treated as +proof that OS confinement can be removed. The restricted Windows path preserves the +original argv, working directory, public scrubbed parent environment and Electron Node-runner +adaptation. It reports the actual confinement enforcement and preserves runner-failure and +denial-signature classification. No shell configuration environment is added to a file worker. +Current authority is checked again after asynchronous preparation and before dispatch. + +The existing `execute` control frame may carry trusted launch facts for batch/result only. +An older kernel without `confined-worker-v1` refuses the request before dispatch. The envelope +remains limited to 2 MiB, launch metadata to 64 KiB, and the shared request window to 32 calls. +There is no second scheduler. Cancellation and completion retain the same request ID. + +The owned range is established before sending request bytes. A started receipt precedes +bounded stdin delivery and EOF; stdin and both output readers join before terminal settlement. +The native envelope deadline covers launch, the physical started acknowledgement, stdin delivery +and a stalled runner before worker decoding. The existing bounded reply queue preserves +started-before-completed ordering when cancelled. A joined input-delivery error does not erase +a known nonzero runner exit or captured stderr; timeout and cancellation remain interrupted. +Cancellation terminates the whole owned range, including the public confinement wrapper. +Unconfirmed cleanup poisons the connection instead of reporting successful cleanup. A lost +receipt never authorizes replay, and interrupted batch effects retain their existing unknown +state and recoverable audit identity when available. + +Cache purge continues to use the dedicated resident teardown path. The remaining one-shot +transport is used only by restricted POSIX batch/result calls and the explicit WSL read path. +This slice does not claim POSIX confined-runner migration or completion of issue #872. + +Windows verification used worker SHA-256 +`024d5e80b9befe77dce3b3ad4bf6c2d4d40e236047054acc79dd9f7196192b2c`. +The eight real routing checks and thirteen Native result checks passed, including actual +public confinement, root-link replacement, cancellation, stalled input, early runner refusal, +and zero public one-shot OPS launches through disposal. The original route failed all three +initial routing checks; the intermediate kernels separately failed the stdin deadline and both +early-refusal checks. These are correctness and ownership observations, not performance results. diff --git a/docs/resident-kernel.md b/docs/resident-kernel.md index 1838ca14..b13272c6 100644 --- a/docs/resident-kernel.md +++ b/docs/resident-kernel.md @@ -46,9 +46,10 @@ worker. Output drains retain bounded tails with explicit loss information. EOF a cancel and join admitted executions. No worker is detached or restarted. This removes repeated **host runner** startup for local file dispatch; it still pays native -child startup per file request. WSL reads and confined PTC batch/result operations -retain their existing runners because their enforced execution contexts cannot be replaced by the -trusted local-file path. The final A/B report must measure all of those costs. This transport is +child startup per file request. Windows batch/result calls now retain exact public confinement +inside an owned native range; full-access calls reuse the fixed leaf. Restricted POSIX batch/result +calls and explicit WSL reads retain their platform runners; see [result and batch routing](resident-confined-workers.md). +The final A/B report must measure all of those costs. This transport is not evidence that session shells, DSH jobs, complete multi-process cleanup or the performance targets have been delivered. @@ -116,7 +117,7 @@ row remains a delivery requirement even if a smaller slice does not exercise it | Cancel first interrupts, then may reset only that shell tree | No inner process cancellation boundary exists. | Hard cancellation reports shell-reset/state loss; other sessions continue. | | Explicit background tasks use DSH job_list/job_output/job_kill | Rust `Job` is a file-batch audit record, not the DSH process-job service. | Immediate jobId, independent output, atomic startup/publication handoff, job-tree kill, session/host cleanup. | | Ordinary command descendants do not become implicit services | Pending shell/task ownership implementation. | Unregistered descendants cannot silently cross calls; published jobs are not killed merely because their startup request completed. | -| Migration leaves one execution architecture | Local file calls use the shared client with no one-shot fallback. WSL and confined batch/result/cleanup paths remain explicit exceptions. | Complete shell/job ownership and eligible remaining migrations without weakening their execution constraints. | +| Migration leaves one execution architecture | Local files, result cleanup and Windows batch/result calls use the shared client without one-shot fallback. Restricted POSIX batch/result and explicit WSL reads retain their platform runners. | Complete remaining platform migrations without weakening their execution constraints. | | End-to-end performance and resource evidence | Windows baseline: 824 calls, 0 failures, including 200 hot samples for each of four items. No resident benefit measured. | Resident A/B, Native/PTC, cold/warm, file/command/mixed/concurrent/long workloads; p50/p95/p99, CPU/RSS/handles and failures. | ## Rust reuse boundary diff --git "a/docs/\345\270\270\351\251\273\345\206\205\346\240\270.md" "b/docs/\345\270\270\351\251\273\345\206\205\346\240\270.md" index 3b4736e3..569298e7 100644 --- "a/docs/\345\270\270\351\251\273\345\206\205\346\240\270.md" +++ "b/docs/\345\270\270\351\251\273\345\206\205\346\240\270.md" @@ -36,8 +36,9 @@ dispatch，并拥有独立终止句柄。Windows 子进程使用嵌套 Job；POS EOF 与关闭会取消并等待已准入执行结束，不 detach、不重启 worker。 本地文件 dispatch 已消除重复的**宿主 runner** 启动；每次文件请求仍有原生子进程启动成本。 -WSL 读取、受约束的 PTC batch/result 操作保留原有 runner，不能用可信本地文件路径 -替代它们的强制执行上下文。最终 A/B 报告必须实测这些成本。这条 transport 不代表会话 shell、DSH jobs、完整多进程回收 +Windows batch/result 调用在所属原生进程范围内保留原始公开 confinement，完全访问调用复用固定 +leaf。受限 POSIX batch/result 与明确启用的 WSL 读取保留平台 runner；参见[结果与批处理路由](常驻结果与批处理.md)。 +最终 A/B 报告必须实测这些成本。这条 transport 不代表会话 shell、DSH jobs、完整多进程回收 或性能目标已经交付。 所属结果缓存清理复用同一常驻连接及既有类型化 `result/purge` 操作，不再启动公开的一次性 worker。 @@ -91,7 +92,7 @@ hook，主入口检查即失败：两个缓存均未完成清理；独立集成 | 先中断，必要时仅重置该 shell 树 | 尚无内部进程取消边界。 | 硬取消明确报告 shell-reset/状态丢失，其他会话继续。 | | 显式后台任务使用 DSH job_list/job_output/job_kill | Rust `Job` 是文件批处理审计记录，不是 DSH 进程任务服务。 | 立即返回 jobId、独立输出、明确启动/发布交接、任务树 kill、会话/宿主回收。 | | 普通命令后代不能自动成为服务 | shell/task 所有权待实现。 | 未登记后代不能静默跨调用存活；已发布任务不因启动请求完成而结束。 | -| 迁移后只有一套执行架构 | 本地文件调用使用共享客户端，无一次性 fallback；WSL 与受约束 batch/result/清理路径仍是明确例外。 | 完成 shell/job 所有权与其余适合迁移的路径，不削弱原执行约束。 | +| 迁移后只有一套执行架构 | 本地文件、结果清理及 Windows batch/result 调用使用共享客户端，无一次性 fallback；受限 POSIX batch/result 与明确启用的 WSL 读取保留平台 runner。 | 完成剩余平台迁移，不削弱原执行约束。 | | 完整调用的性能与资源证据 | Windows 基线共 824 次调用、0 失败，四项各含 200 次热态采样；尚未测得常驻收益。 | 常驻 A/B、Native/PTC、冷/热、文件/命令/混合/并发/长任务；p50/p95/p99、CPU/RSS/句柄与失败样本。 | ## Rust 可复用边界 diff --git "a/docs/\345\270\270\351\251\273\347\273\223\346\236\234\344\270\216\346\211\271\345\244\204\347\220\206.md" "b/docs/\345\270\270\351\251\273\347\273\223\346\236\234\344\270\216\346\211\271\345\244\204\347\220\206.md" new file mode 100644 index 00000000..2d58b659 --- /dev/null +++ "b/docs/\345\270\270\351\251\273\347\273\223\346\236\234\344\270\216\346\211\271\345\244\204\347\220\206.md" @@ -0,0 +1,43 @@ +# 常驻结果与批处理执行 + +[English](resident-confined-workers.md) + +Native 与 PTC 的 result put/page/release 共用同一桥接层，PTC 的 batch plan/run/resume +也走该桥接层。在 Windows 上，这些调用都使用所属常驻内核；公开 subprocess 服务只启动一次 +内核，不再每次调用启动 OPS worker。原有引用与审计实现、ID、结果预算、写前记录、根目录和 +会话绑定、no-follow 检查均保持不变。 + +权限边界明确区分： + +| 当前策略与平台 | 执行路径 | +| --- | --- | +| 所有支持平台的完全访问 | 常驻内核内既有的可信固定 worker leaf。 | +| Windows 的只读或工作区写入 | 原始公开 `sandbox.confine()` 命令，在既有原子 Windows Job 范围内启动。 | +| POSIX 的只读或工作区写入 | 既有公开受限 runner；该平台尚未建立任意 runner 的原生所有权。这条剩余路径不是内核错误后的回退。 | + +批处理只接受数据，但现有根目录相对路径检查并不被当作可移除 OS confinement 的证明。 +受限 Windows 路径保留原始 argv、工作目录、公开清理后的父环境及 Electron Node-runner 适配。 +结果报告真实的 confinement enforcement，保留 runner 失败及拒绝签名分类；不会向文件 worker +加入 shell 配置环境。异步准备结束后、派发前再次核对当前权限。 + +既有 `execute` 控制帧可携带仅供 batch/result 使用的可信启动信息。旧内核若缺少 +`confined-worker-v1`，会在派发前拒绝请求。Envelope 仍限 2 MiB，启动信息限 64 KiB，共享 +请求窗口仍为 32 个调用；不增加第二个调度器。取消与完成保持同一个 request ID。 + +请求字节发送前已建立所属进程范围。Started 回执先于有界 stdin 传输及 EOF，stdin 与两个输出 +reader 均完成回收后才报告终态。原生 envelope deadline 覆盖启动、物理 started 确认、stdin 传输， +以及 worker 解码前停滞的 runner。既有有界回复队列在取消时仍保持 started 先于 completed。 +已回收的输入传输错误不会抹去已知非零 runner 退出码或已捕获 stderr；超时与取消仍报告 interrupted。 +取消会终止整个所属范围，包括公开 confinement wrapper。 +无法确认回收时将连接置为不可用，而不报告清理成功。回执丢失不授权重放，中断的批处理继续保留 +既有 unknown 效果状态，以及可用时的恢复审计标识。 + +缓存 purge 继续使用独立的常驻销毁路径。剩余一次性 transport 只服务于受限 POSIX batch/result +和明确启用的 WSL 读取。本切片不代表 POSIX 受限 runner 已迁移，也不代表议题 #872 已完成。 + +Windows 验证使用 worker SHA-256 +`024d5e80b9befe77dce3b3ad4bf6c2d4d40e236047054acc79dd9f7196192b2c`。 +8 项真实路由检查与 13 项 Native result 检查通过，覆盖公开 confinement、根链接替换、取消、 +stdin 停滞、runner 提前拒绝，以及直到销毁结束均无公开一次性 OPS 启动。原有路径在最初 3 项 +路由检查中全部失败；中间版本内核分别在 stdin deadline 和两个提前拒绝检查中失败。这些是 +正确性与所有权观察，不是性能结果。 diff --git a/native/crates/ops-process/src/shell/mod.rs b/native/crates/ops-process/src/shell/mod.rs index 5941700b..e186fe17 100644 --- a/native/crates/ops-process/src/shell/mod.rs +++ b/native/crates/ops-process/src/shell/mod.rs @@ -3,6 +3,7 @@ use std::{ffi::OsString, io, path::PathBuf}; mod capture; pub use capture::{Capture, CaptureRead, CaptureSnapshot, CaptureView, Retention, SpillSnapshot}; +pub const MAX_INPUT_BYTES: usize = 2 * 1024 * 1024; pub struct Launch { pub argv: Vec, @@ -26,8 +27,8 @@ pub(crate) fn budget(cap: usize) -> io::Result<()> { mod windows; #[cfg(windows)] pub use windows::{ - Baseline, Channel, ChannelSnapshot, ChannelView, MAX_SPILL_BYTES, Range, SpillStore, - signal_console_and_exit, + Baseline, Channel, ChannelSnapshot, ChannelView, MAX_SPILL_BYTES, Range, RangeCancellation, + SpillStore, signal_console_and_exit, }; /// POSIX group creation would escape the public host's PGID fallback owner. diff --git a/native/crates/ops-process/src/shell/windows/launch.rs b/native/crates/ops-process/src/shell/windows/launch.rs index e280bc61..d4341062 100644 --- a/native/crates/ops-process/src/shell/windows/launch.rs +++ b/native/crates/ops-process/src/shell/windows/launch.rs @@ -1,5 +1,5 @@ use super::{argv, channel::Channel, check, job::Job, owned}; -use super::{output::Output, spill_file::SpillStore, spill_io::Cancellation}; +use super::{output::Output, spill_file::SpillStore, spill_io::Cancellation, stdin::Input}; use crate::shell::{Capture, CaptureView, Launch, Retention, budget}; use std::{ fs::File, @@ -19,11 +19,34 @@ use windows_sys::Win32::{ }; type Reader = JoinHandle>; +#[cfg(test)] +#[path = "stdin_tests.rs"] +mod stdin_tests; +/// Requests termination; the owning Range must still observe emptiness and join all I/O. +#[derive(Clone)] +pub struct RangeCancellation { + job: Arc, + input: Option, + output: [Cancellation; 2], +} +impl RangeCancellation { + pub fn cancel(&self) -> io::Result<()> { + if let Some(input) = &self.input { + input.cancel(); + } + for output in &self.output { + output.cancel(); + } + self.job.terminate() + } +} pub struct Range { pub(super) job: Arc, process: Option, pid: Option, readers: Vec, + input: Option, + input_error: Option, observed: [Arc; 2], outputs: Option<[CaptureView; 2]>, spill: Option, @@ -38,6 +61,8 @@ impl Range { process: None, pid: None, readers: Vec::new(), + input: None, + input_error: None, observed: std::array::from_fn(|_| Arc::new(AtomicU64::new(0))), outputs: None, spill: None, @@ -77,6 +102,10 @@ impl Range { pub fn pid(&self) -> Option { self.pid } + /// Delivery failure is distinct from joined captures and confirmed process exit. + pub fn input_error(&self) -> Option<&io::Error> { + self.input_error.as_ref() + } pub fn output_views(&self) -> Option<[CaptureView; 2]> { self.outputs.clone() } @@ -94,6 +123,20 @@ impl Range { .as_ref() .is_some_and(|views| views.iter().any(|view| view.snapshot().cleanup_unconfirmed)) } + pub fn cancellation(&self) -> RangeCancellation { + RangeCancellation { + job: self.job.clone(), + input: self.input.as_ref().map(|input| input.control.clone()), + output: self.spill_cancel.clone(), + } + } + /// Deliver one record only after the caller's start receipt has been sent. + pub fn write_input(&mut self, bytes: Vec) -> io::Result<()> { + self.input + .as_mut() + .ok_or_else(|| io::Error::other("stdin was not requested"))? + .start(bytes) + } fn poison_outputs(&mut self) { self.poisoned = true; if let Some(views) = &self.outputs { @@ -140,10 +183,25 @@ impl Range { pub fn spawn(&mut self, spec: Launch) -> io::Result { self.spawn_with_retention(spec, Retention::Prefix) } - pub fn spawn_with_retention( + pub fn spawn_with_retention(&mut self, spec: Launch, retention: Retention) -> io::Result { + let stderr_cap = spec.output_cap; + self.spawn_io(spec, retention, stderr_cap, false) + } + /// Start with empty owned stdin; bytes are withheld until write_input(). + pub fn spawn_with_stdin( + &mut self, + spec: Launch, + retention: Retention, + stderr_cap: usize, + ) -> io::Result { + self.spawn_io(spec, retention, stderr_cap, true) + } + fn spawn_io( &mut self, mut spec: Launch, retention: Retention, + stderr_cap: usize, + piped: bool, ) -> io::Result { if self.process.is_some() || self.settled { return Err(io::Error::new( @@ -152,9 +210,10 @@ impl Range { )); } budget(spec.output_cap)?; + budget(stderr_cap)?; let outputs = [ CaptureView::new(spec.output_cap, retention)?, - CaptureView::new(spec.output_cap, retention)?, + CaptureView::new(stderr_cap, retention)?, ]; let mut command = argv::command_line(&spec.argv)?; let executable = argv::wide(&spec.argv[0])?; @@ -176,7 +235,20 @@ impl Range { bInheritHandle: 1, }; // Match fresh host execution: EOF from a closed pipe, not a NUL device. - let input = eof_pipe(&sa)?; + let (input, writer) = if piped { + let (input, writer) = pipe(&sa)?; + check(unsafe { + SetHandleInformation( + input.as_raw_handle(), + HANDLE_FLAG_INHERIT, + HANDLE_FLAG_INHERIT, + ) + })?; + check(unsafe { SetHandleInformation(writer.as_raw_handle(), HANDLE_FLAG_INHERIT, 0) })?; + (input, Some(Input::new(writer))) + } else { + (eof_pipe(&sa)?, None) + }; let (out_read, out_write) = pipe(&sa)?; let (err_read, err_write) = pipe(&sa)?; let inherited = [ @@ -230,6 +302,7 @@ impl Range { self.pid = Some(info.dwProcessId); self.process = Some(process); self.outputs = Some(outputs.clone()); + self.input = writer; drop(input); drop(out_write); drop(err_write); @@ -252,7 +325,7 @@ impl Range { err_read, Output::new( outputs[1].clone(), - spec.output_cap, + stderr_cap, self.spill.clone(), "stderr", self.spill_cancel[1].clone(), @@ -270,6 +343,9 @@ impl Range { result } pub fn terminate(&mut self, deadline: Instant) -> io::Result<()> { + if let Some(input) = &mut self.input { + input.cancel(); + } for control in &self.spill_cancel { control.cancel(); } @@ -277,7 +353,7 @@ impl Range { .job .terminate() .and_then(|()| self.wait_empty(deadline)); - if result.is_err() && self.spill.is_some() { + if result.is_err() && (self.spill.is_some() || self.input.is_some()) { self.poison_outputs(); } result @@ -288,11 +364,14 @@ impl Range { self.wait_root_exit(deadline)?; } self.settled = true; + if let Some(input) = &mut self.input { + input.cancel(); + } Ok(()) } /// Read only after confirmed Job emptiness; descendants cannot retain writers then. pub fn captures(&mut self) -> io::Result> { - if self.spill.is_some() { + if self.spill.is_some() || self.input.is_some() { return self.captures_until(Instant::now() + Duration::from_secs(5)); } if !self.settled { @@ -306,7 +385,17 @@ impl Range { if !self.settled { return Err(io::Error::other("shell range is not confirmed empty")); } - while self.readers.iter().any(|reader| !reader.is_finished()) { + while self.readers.iter().any(|reader| !reader.is_finished()) + || self + .input + .as_ref() + .is_some_and(|input| !input.is_finished()) + { + if let Some(input) = &mut self.input + && input.control.cancelled() + { + input.cancel(); + } for control in &self.spill_cancel { if control.cancelled() { control.cancel(); @@ -316,6 +405,9 @@ impl Range { for control in &self.spill_cancel { control.cancel(); } + if let Some(input) = &mut self.input { + input.cancel(); + } self.poison_outputs(); return Err(io::Error::new( io::ErrorKind::TimedOut, @@ -324,6 +416,10 @@ impl Range { } thread::sleep(Duration::from_millis(2)); } + // Attempt both joins even when input delivery or one reader failed. + if let Err(error) = self.input.as_mut().map_or(Ok(()), Input::join) { + self.input_error.get_or_insert(error); + } let result = join_readers(&mut self.readers); if self.cleanup_unconfirmed() { Err(io::Error::other( @@ -348,6 +444,9 @@ impl Drop for Range { // No detach on an unconfirmed I/O cancellation. Fatal transport teardown // must terminate this managed kernel if the operation cannot acknowledge. let _ = join_readers(&mut self.readers); + if let Some(input) = &mut self.input { + let _ = input.join(); + } } } } diff --git a/native/crates/ops-process/src/shell/windows/mod.rs b/native/crates/ops-process/src/shell/windows/mod.rs index 0dcec9d5..8722f379 100644 --- a/native/crates/ops-process/src/shell/windows/mod.rs +++ b/native/crates/ops-process/src/shell/windows/mod.rs @@ -9,9 +9,10 @@ mod security; mod signal; mod spill_file; mod spill_io; +mod stdin; pub use channel::{Channel, ChannelSnapshot, ChannelView}; pub use identity::Baseline; -pub use launch::Range; +pub use launch::{Range, RangeCancellation}; pub use signal::signal_console_and_exit; pub use spill_file::{MAX_SPILL_BYTES, SpillStore}; diff --git a/native/crates/ops-process/src/shell/windows/stdin.rs b/native/crates/ops-process/src/shell/windows/stdin.rs new file mode 100644 index 00000000..06f6da32 --- /dev/null +++ b/native/crates/ops-process/src/shell/windows/stdin.rs @@ -0,0 +1,87 @@ +//! One owned, bounded input record. The writer closes EOF only after its bytes settle. +use super::spill_io::Cancellation; +use crate::shell::MAX_INPUT_BYTES; +use std::{ + fs::File, + io::{self, Write}, + os::windows::io::{AsRawHandle, OwnedHandle}, + thread::{self, JoinHandle}, +}; + +pub(super) struct Input { + writer: Option, + thread: Option>>, + pub control: Cancellation, + #[cfg(test)] + pub written: std::sync::Arc, +} +impl Input { + pub fn new(writer: OwnedHandle) -> Self { + Self { + writer: Some(writer), + thread: None, + control: Cancellation::default(), + #[cfg(test)] + written: Default::default(), + } + } + pub fn start(&mut self, bytes: Vec) -> io::Result<()> { + if bytes.len() > MAX_INPUT_BYTES { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "stdin exceeds 2 MiB", + )); + } + let writer = self + .writer + .take() + .ok_or_else(|| io::Error::other("stdin already closed or started"))?; + let control = self.control.clone(); + #[cfg(test)] + let written = self.written.clone(); + self.thread = Some( + thread::Builder::new() + .name("ops-owned-stdin".into()) + .spawn(move || { + control.register()?; + let mut file = File::from(writer); + let mut remaining = bytes.as_slice(); + while !remaining.is_empty() { + let count = control.operation(Some(file.as_raw_handle()), false, || { + file.write(&remaining[..remaining.len().min(8192)]) + })?; + if count == 0 { + return Err(io::Error::new( + io::ErrorKind::WriteZero, + "stdin wrote no bytes", + )); + } + remaining = &remaining[count..]; + #[cfg(test)] + written.fetch_add(count, std::sync::atomic::Ordering::Release); + } + // File drop closes the sole parent writer before the thread reports completion. + drop(file); + #[cfg(test)] + control.wait_close_gate(); + Ok(()) + })?, + ); + Ok(()) + } + pub fn cancel(&mut self) { + self.writer.take(); + self.control.cancel(); + } + pub fn is_finished(&self) -> bool { + self.writer.is_none() && self.thread.as_ref().is_none_or(JoinHandle::is_finished) + } + pub fn join(&mut self) -> io::Result<()> { + self.writer.take(); + self.thread.take().map_or(Ok(()), |thread| { + thread + .join() + .map_err(|_| io::Error::other("stdin writer panicked"))? + }) + } +} diff --git a/native/crates/ops-process/src/shell/windows/stdin_tests.rs b/native/crates/ops-process/src/shell/windows/stdin_tests.rs new file mode 100644 index 00000000..daa6dee7 --- /dev/null +++ b/native/crates/ops-process/src/shell/windows/stdin_tests.rs @@ -0,0 +1,286 @@ +use super::*; +use crate::shell::MAX_INPUT_BYTES; +use std::{ + fs, + io::Write, + process::{Command, Stdio}, + sync::atomic::AtomicUsize, +}; + +const CHILD: &str = "shell::windows::launch::stdin_tests::owned_input_child"; +struct Release(Arc); +impl Drop for Release { + fn drop(&mut self) { + self.0.release(); + } +} +struct Fixture { + range: Range, + directory: std::path::PathBuf, +} +impl Fixture { + fn new(mode: &str) -> Self { + static NEXT: AtomicUsize = AtomicUsize::new(0); + let directory = std::env::temp_dir().join(format!( + "ops-stdin-{}-{}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + fs::create_dir(&directory).unwrap(); + let mut range = Range::new().unwrap(); + range + .spawn_with_stdin( + Launch { + argv: vec![ + std::env::current_exe().unwrap().into(), + "--ignored".into(), + "--exact".into(), + CHILD.into(), + "--nocapture".into(), + ], + cwd: directory.clone(), + env: vec![ + ("OPS_INPUT_MODE".into(), mode.into()), + ("OPS_INPUT_DIR".into(), directory.clone().into()), + ], + output_cap: 128 * 1024, + }, + Retention::Tail, + 16 * 1024, + ) + .unwrap(); + Self { range, directory } + } + fn gate(&self, file: &str) { + until(|| self.directory.join(file).is_file()); + } +} +impl Drop for Fixture { + fn drop(&mut self) { + self.range + .terminate(Instant::now() + Duration::from_secs(5)) + .unwrap(); + let _ = self + .range + .captures_until(Instant::now() + Duration::from_secs(5)); + fs::remove_dir_all(&self.directory).unwrap(); + } +} +fn until(check: impl Fn() -> bool) { + let deadline = Instant::now() + Duration::from_secs(5); + while !check() && Instant::now() < deadline { + thread::sleep(Duration::from_millis(2)); + } + assert!( + check(), + "fixture condition was not observed before deadline" + ); +} +#[test] +#[ignore = "owned subprocess fixture, invoked explicitly by stdin lifecycle tests"] +fn owned_input_child() { + let directory = std::path::PathBuf::from(std::env::var_os("OPS_INPUT_DIR").unwrap()); + match std::env::var("OPS_INPUT_MODE").unwrap().as_str() { + "echo" => { + let mut bytes = Vec::new(); + std::io::stdin().read_to_end(&mut bytes).unwrap(); + fs::write( + directory.join("received"), + format!( + "{}:{}", + bytes.len(), + bytes.iter().map(|value| u64::from(*value)).sum::() + ), + ) + .unwrap(); + std::io::stdout().write_all(&bytes).unwrap(); + std::io::stdout().flush().unwrap(); + std::io::stderr().write_all(&vec![b'e'; 20000]).unwrap(); + std::process::exit(0); + } + "descendant" => { + let mut child = Command::new(std::env::current_exe().unwrap()) + .args(["--ignored", "--exact", CHILD, "--nocapture"]) + .env_clear() + .env("OPS_INPUT_DIR", &directory) + .env("OPS_INPUT_MODE", "leaf") + .stdin(Stdio::inherit()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + fs::write(directory.join("child-pid"), child.id().to_string()).unwrap(); + fs::write(directory.join("ready"), []).unwrap(); + let _ = child.wait(); + } + "leaf" => { + fs::write(directory.join("child-ready"), []).unwrap(); + } + mode => panic!("unexpected fixture mode {mode}"), + } + fs::write(directory.join("ready"), []).unwrap(); + loop { + thread::sleep(Duration::from_secs(60)); + } +} + +#[test] +fn input_is_exact_and_eof_precedes_exit_with_distinct_bounded_output_tails() { + let mut fixture = Fixture::new("echo"); + let bytes: Vec = (0..MAX_INPUT_BYTES).map(|index| index as u8).collect(); + fixture.range.write_input(bytes.clone()).unwrap(); + assert_eq!( + fixture + .range + .wait_root_exit(Instant::now() + Duration::from_secs(5)) + .unwrap(), + 0 + ); + fixture + .range + .wait_empty(Instant::now() + Duration::from_secs(5)) + .unwrap(); + let captures = fixture + .range + .captures_until(Instant::now() + Duration::from_secs(5)) + .unwrap(); + assert_eq!( + fs::read_to_string(fixture.directory.join("received")).unwrap(), + format!( + "{}:{}", + bytes.len(), + bytes.iter().map(|value| u64::from(*value)).sum::() + ) + ); + assert_eq!(captures[0].bytes, bytes[bytes.len() - 128 * 1024..]); + assert_eq!(captures[1].bytes, vec![b'e'; 16 * 1024]); + assert!(captures.iter().all(|capture| capture.truncated)); + assert!(!fixture.range.cleanup_unconfirmed()); +} + +#[test] +fn cancellation_joins_blocked_stdin_and_children_created_before_input() { + let mut fixture = Fixture::new("descendant"); + fixture.gate("child-ready"); + let child: u32 = fs::read_to_string(fixture.directory.join("child-pid")) + .unwrap() + .parse() + .unwrap(); + assert!( + fixture.range.job.contains_pid(child).unwrap(), + "the early descendant belongs to the Job before any input" + ); + assert!(fixture.range.active_processes().unwrap() >= 2); + fixture + .range + .write_input(vec![b'x'; MAX_INPUT_BYTES]) + .unwrap(); + until(|| { + fixture + .range + .input + .as_ref() + .unwrap() + .written + .load(Ordering::Acquire) + >= 65536 + }); + assert!( + !fixture.range.input.as_ref().unwrap().is_finished(), + "the owned child never reads stdin" + ); + fixture.range.cancellation().cancel().unwrap(); + fixture + .range + .terminate(Instant::now() + Duration::from_secs(5)) + .unwrap(); + assert!( + fixture + .range + .captures_until(Instant::now() + Duration::from_secs(5)) + .is_ok(), + "joined stdout/stderr are independent of partial input delivery" + ); + assert!( + fixture.range.input_error().is_some(), + "partial cancelled input is not successful delivery" + ); + assert_eq!(fixture.range.active_processes().unwrap(), 0); + assert!(fixture.range.input.as_ref().unwrap().is_finished()); + assert!(fixture.range.readers.is_empty()); + assert!(!fixture.range.cleanup_unconfirmed()); +} + +#[test] +fn an_unjoined_input_owner_poison_cannot_be_cleared_by_late_finish() { + let mut fixture = Fixture::new("echo"); + let (gate, entered) = fixture + .range + .input + .as_ref() + .unwrap() + .control + .gate_next_close(); + let _release = Release(gate.clone()); + fixture.range.write_input(b"bounded".to_vec()).unwrap(); + entered.recv_timeout(Duration::from_secs(5)).unwrap(); + fixture + .range + .wait_root_exit(Instant::now() + Duration::from_secs(5)) + .unwrap(); + fixture + .range + .wait_empty(Instant::now() + Duration::from_secs(5)) + .unwrap(); + let result = fixture.range.captures_until(Instant::now()); + gate.release(); + assert!(result.is_err()); + assert!(fixture.range.cleanup_unconfirmed()); + assert!( + fixture + .range + .captures_until(Instant::now() + Duration::from_secs(5)) + .is_err() + ); + assert!( + fixture + .range + .output_views() + .unwrap() + .iter() + .all(|view| !view.snapshot().eof) + ); +} + +#[test] +fn input_record_cap_rejects_without_sending_bytes_or_losing_ownership() { + let mut fixture = Fixture::new("echo"); + assert!( + fixture + .range + .write_input(vec![0; MAX_INPUT_BYTES + 1]) + .is_err() + ); + assert_eq!( + fixture + .range + .input + .as_ref() + .unwrap() + .written + .load(Ordering::Acquire), + 0 + ); + fixture.range.write_input(Vec::new()).unwrap(); + assert_eq!( + fixture + .range + .wait_root_exit(Instant::now() + Duration::from_secs(5)) + .unwrap(), + 0 + ); + assert_eq!( + fs::read_to_string(fixture.directory.join("received")).unwrap(), + "0:0" + ); +} diff --git a/native/examples/shell_started_blocked.rs b/native/examples/shell_started_blocked.rs index ade33831..3e6ce779 100644 --- a/native/examples/shell_started_blocked.rs +++ b/native/examples/shell_started_blocked.rs @@ -66,9 +66,11 @@ mod probe { &std::env::temp_dir(), ops_process::shell::MAX_SPILL_BYTES, )?)?; - (manager.output())( + (manager.started_output())( json!({"type":"ready","pid":std::process::id(),"capabilities":shell::capabilities()}), - )?; + )? + .recv() + .map_err(io::Error::other)??; let result = (|| { while let Some(bytes) = read_frame(&mut reader)? { let value: Value = serde_json::from_slice(&bytes)?; diff --git a/native/src/resident.rs b/native/src/resident.rs index 43dbcf7f..c0d47719 100644 --- a/native/src/resident.rs +++ b/native/src/resident.rs @@ -15,13 +15,33 @@ use std::{ }; const WINDOW: usize = 32; -const FRAME_BYTES: usize = envelope::MAX_REQUEST_BYTES + 256; +#[path = "resident/confined.rs"] +mod confined; +const FRAME_BYTES: usize = envelope::MAX_REQUEST_BYTES + confined::LAUNCH_BYTES + 256; type Output = Arc>; type Active = Arc>>>>; #[derive(Default)] struct RequestState { cancelled: bool, - leaf: Option, + leaf: Option, +} +enum Running { + Leaf(Cancel), + #[cfg(windows)] + Confined(ops_process::shell::RangeCancellation), +} +impl Running { + fn cancel(&self) { + match self { + Self::Leaf(leaf) => { + let _ = leaf.cancel(); + } + #[cfg(windows)] + Self::Confined(range) => { + let _ = range.cancel(); + } + } + } } fn send(output: &Output, value: Value) -> io::Result<()> { @@ -43,13 +63,35 @@ fn cancel(state: &Arc>) { let mut state = state.lock().unwrap(); state.cancelled = true; if let Some(leaf) = &state.leaf { - let _ = leaf.cancel(); + leaf.cancel(); } } -fn execute(input: Vec, id: &str, state: &Arc>, output: &Output) -> Value { +fn execute( + input: Vec, + launch: Option, + id: &str, + state: &Arc>, + output: &Output, + shutdown: &ControlPipe, + started: &crate::shell::Started, +) -> Value { if state.lock().unwrap().cancelled { return interrupted(); } + if let Some(launch) = launch { + #[cfg(windows)] + return confined::execute(input, launch, id, state, started, &|| { + let _ = shutdown.shutdown(); + }); + #[cfg(not(windows))] + { + let _ = (launch, shutdown, started); + return refused( + "EXECUTOR_UNAVAILABLE", + "confined worker ownership requires Windows Jobs", + ); + } + } let execution = (|| { Execution::spawn( &std::env::current_exe()?, @@ -69,9 +111,9 @@ fn execute(input: Vec, id: &str, state: &Arc>, output: & }; { let mut current = state.lock().unwrap(); - current.leaf = Some(execution.cancellation()); + current.leaf = Some(Running::Leaf(execution.cancellation())); if current.cancelled { - let _ = current.leaf.as_ref().unwrap().cancel(); + current.leaf.as_ref().unwrap().cancel(); } } // From here the caller must conservatively treat an unconfirmed effect as @@ -149,7 +191,7 @@ pub fn serve() -> io::Result<()> { capabilities .as_array_mut() .unwrap() - .push(json!("shell-spill-v1")); + .extend([json!("shell-spill-v1"), json!("confined-worker-v1")]); capabilities .as_array_mut() .unwrap() @@ -183,6 +225,11 @@ pub fn serve() -> io::Result<()> { if input.len() > envelope::MAX_REQUEST_BYTES { return Err(io::Error::other("request exceeds 2 MiB")); } + let launch = message + .get("launch") + .map(|launch| confined::parse(launch, envelope)) + .transpose()?; + let shutdown = reader.get_ref().try_clone()?; let state = Arc::new(Mutex::new(RequestState::default())); { let mut requests = active.lock().unwrap(); @@ -196,8 +243,21 @@ pub fn serve() -> io::Result<()> { requests.insert(id.clone(), state.clone()); } let (output, active) = (output.clone(), active.clone()); + let started = shells.started_output(); + let completed = shells.output(); tasks.push(thread::spawn(move || { - let exchange = execute(input, &id, &state, &output); + let confined = launch.is_some(); + let exchange = + execute(input, launch, &id, &state, &output, &shutdown, &started); + if confined { + // Share the bounded transport queue with started, including + // a start still pending when cancellation/deadline won. + active.lock().unwrap().remove(&id); + let _ = completed( + json!({"type":"completed","requestId":id,"exchange":exchange}), + ); + return; + } // Release the admission slot BEFORE publishing completion. // Holding the writer limits removed-but-blocked replies to one; // other slow replies retain their slots and apply backpressure. diff --git a/native/src/resident/confined.rs b/native/src/resident/confined.rs new file mode 100644 index 00000000..d05e4d60 --- /dev/null +++ b/native/src/resident/confined.rs @@ -0,0 +1,263 @@ +//! Exact host-confined worker launches share the existing resident request lifetime. +use serde_json::Value; +use std::{collections::BTreeMap, io}; +#[cfg(all(windows, test))] +#[path = "confined_tests.rs"] +mod lifecycle_tests; + +pub(super) const LAUNCH_BYTES: usize = 64 * 1024; +pub(super) struct Launch { + argv: Vec, + cwd: String, + env: BTreeMap, + #[cfg(windows)] + deadline: std::time::Instant, +} +pub(super) fn parse(value: &Value, envelope: &Value) -> io::Result { + let timeout = envelope["budget"]["timeoutMs"] + .as_u64() + .filter(|value| (1..=2_147_483_647).contains(value)) + .ok_or_else(|| io::Error::other("confined launch requires timeoutMs 1..2147483647"))?; + #[cfg(not(windows))] + let _ = timeout; + if envelope["component"] != "ptc" + || !matches!( + envelope["request"]["tool"].as_str(), + Some("batch" | "result") + ) + { + return Err(io::Error::other( + "confined launch requires a trusted batch/result envelope", + )); + } + if serde_json::to_vec(value)?.len() > LAUNCH_BYTES { + return Err(io::Error::other("confined launch exceeds 64 KiB")); + } + let object = value + .as_object() + .ok_or_else(|| io::Error::other("launch must be an object"))?; + if object.len() != 3 + || object + .keys() + .any(|key| !["argv", "cwd", "env"].contains(&key.as_str())) + { + return Err(io::Error::other("launch requires only argv, cwd and env")); + } + let text = |value: &Value| { + value + .as_str() + .map(str::to_owned) + .ok_or_else(|| io::Error::other("launch values must be strings")) + }; + let launch = Launch { + #[cfg(windows)] + deadline: std::time::Instant::now() + std::time::Duration::from_millis(timeout), + argv: value["argv"] + .as_array() + .ok_or_else(|| io::Error::other("argv must be an array"))? + .iter() + .map(text) + .collect::>()?, + cwd: text(&value["cwd"])?, + env: value["env"] + .as_object() + .ok_or_else(|| io::Error::other("env must be an object"))? + .iter() + .map(|(key, value)| Ok((key.clone(), text(value)?))) + .collect::>()?, + }; + if launch.argv.is_empty() + || !std::path::Path::new(&launch.argv[0]).is_absolute() + || !std::path::Path::new(&launch.cwd).is_absolute() + || launch.argv.iter().any(|value| value.contains('\0')) + || launch.cwd.contains('\0') + || launch + .env + .iter() + .any(|(key, value)| key.is_empty() || key.contains(['\0', '=']) || value.contains('\0')) + { + return Err(io::Error::other("invalid exact confined launch")); + } + #[cfg(not(windows))] + return Err(io::Error::new( + io::ErrorKind::Unsupported, + "confined worker ownership requires Windows Jobs", + )); + #[cfg(windows)] + Ok(launch) +} + +#[cfg(windows)] +pub(super) fn execute( + input: Vec, + launch: Launch, + id: &str, + state: &std::sync::Arc>, + started: &crate::shell::Started, + fatal: &dyn Fn(), +) -> Value { + use ops_process::shell::{Range, Retention}; + use serde_json::json; + use std::{ + thread, + time::{Duration, Instant}, + }; + let mut range = match Range::new() { + Ok(range) => range, + Err(error) => return super::refused("EXECUTOR_UNAVAILABLE", &error.to_string()), + }; + let result = (|| -> io::Result> { + check(state, launch.deadline)?; + range.spawn_with_stdin( + ops_process::shell::Launch { + argv: launch.argv.into_iter().map(Into::into).collect(), + cwd: launch.cwd.into(), + env: launch + .env + .into_iter() + .map(|(key, value)| (key.into(), value.into())) + .collect(), + output_cap: dsh_ops_core::contract::envelope::MAX_OUTPUT_BYTES, + }, + Retention::Tail, + 16 * 1024, + )?; + { + let mut current = state.lock().unwrap(); + current.leaf = Some(super::Running::Confined(range.cancellation())); + if current.cancelled { + current.leaf.as_ref().unwrap().cancel(); + } + } + // Ownership precedes the physical start receipt, which precedes every stdin byte. + let receipt = started(json!({"type":"started","requestId":id}))?; + loop { + check(state, launch.deadline)?; + match receipt.recv_timeout(Duration::from_millis(2)) { + Ok(result) => { + result?; + break; + } + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {} + Err(error) => return Err(io::Error::other(error)), + } + } + check(state, launch.deadline)?; + range.write_input(input)?; + loop { + check(state, launch.deadline)?; + if let Some(code) = range.root_exit()? { + return Ok(Some(code)); + } + thread::sleep(Duration::from_millis(2)); + } + })(); + // Both cleanup phases run even after a spawn/input/capture error. A poisoned owner + // is retained by Drop; close independent control first so the host kills the kernel. + let stop = range.terminate(Instant::now() + Duration::from_secs(5)); + let captures = range.captures_until(Instant::now() + Duration::from_secs(5)); + if stop.is_err() || range.cleanup_unconfirmed() { + fatal(); + } + let mut current = state.lock().unwrap(); + current.leaf = None; + match result.and_then(|exit| { + stop?; + Ok((exit, captures?)) + }) { + Ok((exit, captures)) => { + let stopped = current.cancelled || (exit == Some(0) && range.input_error().is_some()); + let mut exchange = json!({"exitCode":exit,"interrupted":stopped,"readers":true, + "lossy":captures[0].truncated,"stderr":String::from_utf8_lossy(&captures[1].bytes)}); + if exit == Some(0) && !stopped && !captures[0].truncated { + match serde_json::from_slice::(&captures[0].bytes) { + Ok(value) if value.is_object() => exchange["value"] = value, + Ok(_) => exchange["invalid"] = json!("shape"), + Err(_) => exchange["invalid"] = json!("json"), + } + } + exchange + } + Err(error) if range.pid().is_none() => { + super::refused("EXECUTOR_UNAVAILABLE", &error.to_string()) + } + Err(error) => { + json!({"exitCode":null,"interrupted":true,"readers":true,"lossy":false,"stderr":error.to_string()}) + } + } +} + +#[cfg(windows)] +fn check( + state: &std::sync::Mutex, + deadline: std::time::Instant, +) -> io::Result<()> { + if std::time::Instant::now() >= deadline { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "confined worker deadline expired", + )); + } + if state.lock().unwrap().cancelled { + return Err(io::Error::new( + io::ErrorKind::Interrupted, + "confined worker cancelled", + )); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + #[test] + fn launch_is_bounded_and_cannot_change_the_protocol_shape() { + let exe = std::env::current_exe() + .unwrap() + .to_string_lossy() + .into_owned(); + let cwd = std::env::current_dir() + .unwrap() + .to_string_lossy() + .into_owned(); + let valid = json!({"argv":[exe],"cwd":cwd,"env":{}}); + let envelope = + json!({"component":"ptc","request":{"tool":"batch"},"budget":{"timeoutMs":1000}}); + if cfg!(windows) { + assert!(parse(&valid, &envelope).is_ok()); + } else { + assert_eq!( + parse(&valid, &envelope).err().unwrap().kind(), + io::ErrorKind::Unsupported + ); + } + let mut extra = valid.clone(); + extra["command"] = json!("untrusted"); + assert!(parse(&extra, &envelope).is_err()); + let mut large = valid.clone(); + large["env"] = json!({"x":"a".repeat(LAUNCH_BYTES)}); + assert!(parse(&large, &envelope).is_err()); + let mut relative = valid.clone(); + relative["argv"] = json!(["worker"]); + assert!(parse(&relative, &envelope).is_err()); + assert!( + parse( + &valid, + &json!({"component":"file","request":{"tool":"read"},"budget":{"timeoutMs":1000}}) + ) + .is_err() + ); + for timeout in [ + json!(0), + json!(-1), + json!(2_147_483_648u64), + json!(1.5), + json!(null), + ] { + let mut invalid = envelope.clone(); + invalid["budget"]["timeoutMs"] = timeout; + assert!(parse(&valid, &invalid).is_err()); + } + } +} diff --git a/native/src/resident/confined_tests.rs b/native/src/resident/confined_tests.rs new file mode 100644 index 00000000..88f35ef8 --- /dev/null +++ b/native/src/resident/confined_tests.rs @@ -0,0 +1,200 @@ +//! Real owned process plus the production bounded reply writer, with deterministic backpressure. +use super::*; +use serde_json::json; +use std::{ + fs::{self, OpenOptions}, + io::Read, + os::windows::fs::OpenOptionsExt, + sync::{ + Arc, Condvar, Mutex, + atomic::{AtomicUsize, Ordering}, + mpsc, + }, + thread, + time::{Duration, Instant}, +}; +const CHILD: &str = "resident::confined::lifecycle_tests::blocked_child"; +#[test] +#[ignore = "owned subprocess fixture, invoked explicitly by confined lifecycle test"] +fn blocked_child() { + if std::env::var_os("OPS_EARLY_EXIT").is_some() { + eprintln!("denied by confined fixture"); + std::process::exit(23); + } + let root = std::env::current_dir().unwrap(); + let _held = OpenOptions::new() + .create_new(true) + .write(true) + .share_mode(0) + .open(root.join("held")) + .unwrap(); + fs::write(root.join("ready"), "ready").unwrap(); + let mut byte = [0]; + if std::io::stdin().read(&mut byte).unwrap_or(0) != 0 { + fs::write(root.join("received"), byte).unwrap(); + } + loop { + thread::park(); + } +} +#[test] +fn early_runner_refusal_preserves_exit_and_stderr_after_partial_input() { + let launch = Launch { + argv: vec![ + std::env::current_exe() + .unwrap() + .to_string_lossy() + .into_owned(), + "--ignored".into(), + "--exact".into(), + CHILD.into(), + "--nocapture".into(), + ], + cwd: std::env::current_dir() + .unwrap() + .to_string_lossy() + .into_owned(), + env: BTreeMap::from([("OPS_EARLY_EXIT".into(), "1".into())]), + deadline: Instant::now() + Duration::from_secs(5), + }; + let started: crate::shell::Started = Arc::new(|_| { + let (sender, receiver) = mpsc::sync_channel(1); + sender.send(Ok(())).unwrap(); + Ok(receiver) + }); + let state = Arc::new(Mutex::new(super::super::RequestState::default())); + let exchange = execute( + vec![b'x'; 2 * 1024 * 1024], + launch, + "refusal", + &state, + &started, + &|| panic!("early refusal must join"), + ); + assert_eq!(exchange["exitCode"], 23, "{exchange}"); + assert_eq!(exchange["interrupted"], false, "{exchange}"); + assert!( + exchange["stderr"] + .as_str() + .unwrap() + .contains("denied by confined fixture") + ); + assert!(state.lock().unwrap().leaf.is_none()); +} +struct Gate(Arc<(Mutex, Condvar)>); +impl Gate { + fn release(&self) { + *self.0.0.lock().unwrap() = true; + self.0.1.notify_all(); + } +} +impl Drop for Gate { + fn drop(&mut self) { + self.release(); + } +} +#[test] +fn deadline_joins_owned_runner_while_start_receipt_is_blocked_and_preserves_order() { + static NEXT: AtomicUsize = AtomicUsize::new(0); + let directory = std::env::temp_dir().join(format!( + "ops-confined-{}-{}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + fs::create_dir(&directory).unwrap(); + let gate = Gate(Arc::new((Mutex::new(false), Condvar::new()))); + let blocked = gate.0.clone(); + let (entered, entering) = mpsc::sync_channel(1); + let (sent, frames) = mpsc::channel(); + let mut manager = crate::shell::Manager::new( + move |value| { + if value["type"] == "started" { + entered.send(()).unwrap(); + let mut released = blocked.0.lock().unwrap(); + while !*released { + released = blocked.1.wait(released).unwrap(); + } + } + sent.send(value).unwrap(); + Ok(()) + }, + || panic!("cleanup must be confirmed"), + ) + .unwrap(); + let launch = Launch { + argv: vec![ + std::env::current_exe() + .unwrap() + .to_string_lossy() + .into_owned(), + "--ignored".into(), + "--exact".into(), + CHILD.into(), + "--nocapture".into(), + ], + cwd: directory.to_string_lossy().into_owned(), + env: BTreeMap::new(), + deadline: Instant::now() + Duration::from_millis(1500), + }; + let state = Arc::new(Mutex::new(super::super::RequestState::default())); + let active = state.clone(); + let started = manager.started_output(); + let completed = manager.output(); + let (settled, terminal) = mpsc::sync_channel(1); + let runner = thread::spawn(move || { + let exchange = execute( + vec![b'x'; 2 * 1024 * 1024], + launch, + "blocked", + &active, + &started, + &|| panic!("owned input cleanup must settle"), + ); + completed(json!({"type":"completed","requestId":"blocked","exchange":exchange})).unwrap(); + settled.send(exchange).unwrap(); + }); + entering.recv_timeout(Duration::from_secs(5)).unwrap(); + let ready_deadline = Instant::now() + Duration::from_secs(1); + while !directory.join("ready").exists() && Instant::now() < ready_deadline { + thread::sleep(Duration::from_millis(2)); + } + let child_was_ready = directory.join("ready").exists(); + let result = terminal.recv_timeout(Duration::from_secs(4)); + let child_released = result.is_ok() + && OpenOptions::new() + .write(true) + .share_mode(0) + .open(directory.join("held")) + .is_ok(); + // Even a failing regression releases the writer and cancels the owned child before joining. + let completed_while_blocked = result.is_ok(); + if !completed_while_blocked { + super::super::cancel(&state); + } + gate.release(); + runner.join().unwrap(); + let exchange = result.unwrap_or_else(|_| terminal.recv().unwrap()); + let first = frames.recv_timeout(Duration::from_secs(2)).unwrap(); + let second = frames.recv_timeout(Duration::from_secs(2)).unwrap(); + manager.shutdown().unwrap(); + let stdin_untouched = !directory.join("received").exists(); + fs::remove_dir_all(directory).unwrap(); + assert!(child_was_ready); + assert!( + completed_while_blocked, + "deadline cannot wait for the physical start writer" + ); + assert!( + child_released, + "owned child must have released its exclusive handle before completion" + ); + assert!( + stdin_untouched, + "no request bytes may precede the physical start receipt" + ); + assert_eq!(first["type"], "started"); + assert_eq!(second["type"], "completed"); + assert_eq!(exchange["interrupted"], true); + assert!(exchange.get("value").is_none()); + assert!(state.lock().unwrap().leaf.is_none()); +} diff --git a/native/src/shell/manager.rs b/native/src/shell/manager.rs index a9435e87..b8064fc8 100644 --- a/native/src/shell/manager.rs +++ b/native/src/shell/manager.rs @@ -1,4 +1,4 @@ -use super::{protocol, session}; +use super::{Started, protocol, session}; use ops_process::shell::{ChannelView, SpillStore}; use serde_json::{Value, json}; use std::{ @@ -15,7 +15,6 @@ use std::{ const RECORDS: usize = 32; pub(super) const REPLY_WINDOW: usize = 512; type Output = Arc io::Result<()> + Send + Sync>; -type Started = Arc io::Result>> + Send + Sync>; type Response = (Value, Option>>); pub(super) struct State { pub spill: OnceLock, @@ -165,6 +164,9 @@ impl Manager { pub fn output(&self) -> Arc io::Result<()> + Send + Sync> { self.state.output.clone() } + pub fn started_output(&self) -> Started { + self.state.started.clone() + } pub fn reap(&mut self) -> io::Result<()> { let mut index = 0; while index < self.tasks.len() { @@ -327,7 +329,7 @@ impl Manager { let close = value["type"] == "shell.close"; let session = self.state.sessions.lock().unwrap().get(&key).cloned(); let Some(session) = session else { - return (self.state.output)( + return (self.output())( json!({"type":"shell.closed","requestId":id,"cleanupConfirmed":true}), ); }; diff --git a/native/src/shell/mod.rs b/native/src/shell/mod.rs index a11316af..bc3f8481 100644 --- a/native/src/shell/mod.rs +++ b/native/src/shell/mod.rs @@ -1,4 +1,9 @@ //! Kernel-owned shell queues. The host supplies current authority at every head. +pub type Started = std::sync::Arc< + dyn Fn(serde_json::Value) -> std::io::Result>> + + Send + + Sync, +>; #[cfg(windows)] mod driver; #[cfg(windows)] @@ -38,6 +43,9 @@ impl Manager { ) -> std::sync::Arc std::io::Result<()> + Send + Sync> { std::sync::Arc::new(|_| Err(ops_process::shell::unsupported())) } + pub fn started_output(&self) -> Started { + std::sync::Arc::new(|_| Err(ops_process::shell::unsupported())) + } pub fn reap(&mut self) -> std::io::Result<()> { Ok(()) } diff --git a/src/execution/jobs.ts b/src/execution/jobs.ts index 661da1f5..3b8896a8 100644 --- a/src/execution/jobs.ts +++ b/src/execution/jobs.ts @@ -21,6 +21,8 @@ import { sessionTempDir } from '../session-temp.js' import { denied, authorityChanged, executionRestriction, normalizeNativeDenial, credentialDenied } from '../permissions.js' import { inside, canonicalAnchor, sameAuthority } from '../authority-paths.js' import { prepareWorkspace } from '../workspace-preparation.js' +import { requireResident } from './resident/provider.js' +import { confinedWorkerLaunch } from './resident/worker-launch.js' export { resolveWorker } from './runtime.js' export function failure(code: string, message: string, effectState: EffectState = 'none') { return { status: 'error', error: { code, phase: effectState === 'none' ? 'preflight' : 'execute', retryable: false, effectState, @@ -100,6 +102,9 @@ async function invokeTrusted(ctx: Context, file: string, tool: string, args: unk } const denyRoots = (config?.denyCredentialReads ?? true) !== false ? credentialRoots() : [] if (Buffer.byteLength(JSON.stringify(workerArgs)) > 2 * 1024 * 1024) return failure('OUTPUT_LIMIT', 'Request exceeds the 2 MiB byte budget.') + // Restricted POSIX runners retain their explicit platform route until native + // range ownership is established there. This is never a resident-error fallback. + const resident = process.platform === 'win32' || policy.mode === 'danger-full-access' ? requireResident(ctx, { binaryPath: file }) : undefined let argv = [file] if (policy.mode !== 'danger-full-access') { const sandbox = ctx.get('sandbox') @@ -121,6 +126,7 @@ async function invokeTrusted(ctx: Context, file: string, tool: string, args: unk return failure('EXECUTOR_UNAVAILABLE', 'Session result storage is unavailable under current host authority.') } if (confined) argv = exposePosixCache(argv, storage) + const launch = resident && confined ? await confinedWorkerLaunch(argv, policy.workspaceRoot) : undefined const envelope = workerEnvelope('ptc', { root: operationRoot, scope: String(session.id), tool, args: workerArgs, read_only: readOnly, deny_roots: denyRoots, deny_credential_names: (config?.denyCredentialReads ?? true) !== false, ...(jobId ? { job_id: jobId } : {}), ...indexCacheRequest({ enabled: config?.indexCache === true }) }, undefined, storage) const input = JSON.stringify(envelope) if (Buffer.byteLength(input) > 2 * 1024 * 1024) return failure('OUTPUT_LIMIT', 'Request exceeds the 2 MiB byte budget.') @@ -135,9 +141,12 @@ async function invokeTrusted(ctx: Context, file: string, tool: string, args: unk return authorityChanged(current, operation) } if (admission && !admission()) return publicationDenied() - const result = await exchange(subprocess, { argv, cwd: policy.workspaceRoot, - stdio: { stdin: { data: input }, stdout: { maxBytes: envelope.budget.maxBytes }, stderr: { maxBytes: 16 * 1024 } }, - graceMs: 1000, signal }, () => { spawned = true }) + const started = () => { spawned = true } + const result = resident + ? launch ? await resident.invokeConfined(envelope, launch, signal, started) : await resident.invoke(envelope, signal, started) + : await exchange(subprocess, { argv, cwd: policy.workspaceRoot, + stdio: { stdin: { data: input }, stdout: { maxBytes: envelope.budget.maxBytes }, stderr: { maxBytes: 16 * 1024 } }, + graceMs: 1000, signal }, started) if (!result.readers) return errorResult('RESULT_UNKNOWN', mutates ? 'Worker output readers are unavailable; inspect audit records.' : 'Read-only worker output readers are unavailable; no data was changed.', mutates ? 'unknown' : 'none') if (result.interrupted) { const code = mutates ? 'RESULT_UNKNOWN' : signal.reason instanceof InvocationDeadlineError ? 'TIMEOUT' : signal.aborted ? 'ABORTED' : 'RESULT_UNKNOWN' diff --git a/src/execution/resident/client.ts b/src/execution/resident/client.ts index 1fa32bf2..a8a6ece0 100644 --- a/src/execution/resident/client.ts +++ b/src/execution/resident/client.ts @@ -10,6 +10,7 @@ import { ResidentShellChannel } from './shell-channel.js' import type { ShellTransport } from './shell-protocol.js' import { ResidentTaskChannel } from './task-channel.js' import type { TaskTransport } from './task-protocol.js' +import { validateWorkerLaunch, type ConfinedWorkerLaunch } from './worker-launch.js' type Envelope = ReturnType interface Pending { @@ -29,6 +30,7 @@ export interface ResidentClient { readonly shell: ShellTransport readonly task: TaskTransport invoke(envelope: Envelope, signal?: AbortSignal, started?: () => void): Promise + invokeConfined(envelope: Envelope, launch: ConfinedWorkerLaunch, signal?: AbortSignal, started?: () => void): Promise purgeResults(root: string, scope: string, storage: WorkerStorage, signal: AbortSignal): Promise stop(): Promise } @@ -144,13 +146,21 @@ class Client implements ResidentClient { invoke(envelope: Envelope, signal?: AbortSignal, started?: () => void): Promise { return this.dispatch(envelope, signal, started) } + invokeConfined(envelope: Envelope, launch: ConfinedWorkerLaunch, signal?: AbortSignal, started?: () => void): Promise { + try { + if (!this.task.capabilities.includes('confined-worker-v1')) throw new ResidentNotStarted('Resident kernel does not support confined-worker-v1; update its runtime before executing restricted batch/result calls') + if (envelope.component !== 'ptc' || !['batch', 'result'].includes(String(envelope.request.tool))) throw new ResidentNotStarted('Confined execution requires a trusted batch/result envelope') + validateWorkerLaunch(launch) + return this.dispatch(envelope, signal, started, false, launch) + } catch (error) { return Promise.reject(error) } + } /** Only already-owned cache teardown may use the live connection after ordinary admission closes. */ purgeResults(root: string, scope: string, storage: WorkerStorage, signal: AbortSignal): Promise { const envelope = workerEnvelope('ptc', { root, scope, tool: 'result', args: { action: 'purge' }, read_only: false }, 5000, storage) envelope.budget.maxBytes = 4096 return this.dispatch(envelope, signal, undefined, true) } - private dispatch(envelope: Envelope, signal?: AbortSignal, started?: () => void, cleanup = false): Promise { + private dispatch(envelope: Envelope, signal?: AbortSignal, started?: () => void, cleanup = false, launch?: ConfinedWorkerLaunch): Promise { if (signal?.aborted) return Promise.reject(new InvocationNotStarted()) if (!(cleanup ? this.pid && !this.closing && !this.terminal : this.available)) return Promise.reject(new ResidentNotStarted('Resident kernel is unavailable; reload the plugin to start a new instance')) if (this.pending.size >= 32) return Promise.reject(new ResidentNotStarted('Resident kernel has 32 in-flight requests; admission is full')) @@ -159,7 +169,7 @@ class Client implements ResidentClient { if (envelope.version !== 1 || typeof envelope.requestId !== 'string' || !/^[A-Za-z0-9_-]{1,128}$/.test(envelope.requestId) || this.pending.has(envelope.requestId)) throw new ResidentNotStarted('Invalid or duplicate resident request identity') if (Buffer.byteLength(JSON.stringify(envelope)) > ENVELOPE_BYTES) throw new ResidentNotStarted('Resident request exceeds the 2 MiB byte budget') - bytes = encodeFrame({ type: 'execute', envelope }) + bytes = encodeFrame({ type: 'execute', envelope, ...(launch ? { launch } : {}) }) } catch (error) { return Promise.reject(error) } return new Promise((resolve, reject) => { const abort = () => { diff --git a/src/execution/resident/frames.ts b/src/execution/resident/frames.ts index f37f794e..8af2e954 100644 --- a/src/execution/resident/frames.ts +++ b/src/execution/resident/frames.ts @@ -2,7 +2,8 @@ import type { Duplex } from 'node:stream' export const ENVELOPE_BYTES = 2 * 1024 * 1024 -export const FRAME_BYTES = ENVELOPE_BYTES + 256 +export const LAUNCH_BYTES = 64 * 1024 +export const FRAME_BYTES = ENVELOPE_BYTES + LAUNCH_BYTES + 256 const QUEUED_FRAMES = 65 // File-only: 32 executions, one cancellation each, one shutdown. interface Queued { bytes: Buffer; sending(): void } diff --git a/src/execution/resident/worker-launch.ts b/src/execution/resident/worker-launch.ts new file mode 100644 index 00000000..f47a6c23 --- /dev/null +++ b/src/execution/resident/worker-launch.ts @@ -0,0 +1,25 @@ +/** Exact public confinement launch facts; never exposed as tool arguments. */ +import path from 'node:path' +import { LAUNCH_BYTES } from './frames.js' +import { loadScrubbedParentEnv } from '../../adapter/dsh/host.js' +import { adaptElectronRunner } from '../../shells/environment.js' + +export interface ConfinedWorkerLaunch { + argv: string[] + cwd: string + env: Record +} +export async function confinedWorkerLaunch(argv: string[], cwd: string): Promise { + const launch = { argv: [...argv], cwd, + env: adaptElectronRunner(argv, await loadScrubbedParentEnv()) as Record } + validateWorkerLaunch(launch) + return launch +} +export function validateWorkerLaunch(launch: ConfinedWorkerLaunch): void { + const text = (value: unknown): value is string => typeof value === 'string' && !value.includes('\0') + if (!launch || !Array.isArray(launch.argv) || !launch.argv.length || !launch.argv.every(text) + || !path.isAbsolute(launch.argv[0]!) || !text(launch.cwd) || !path.isAbsolute(launch.cwd) + || !launch.env || typeof launch.env !== 'object' || Array.isArray(launch.env) + || Object.entries(launch.env).some(([key, value]) => !key || !text(key) || key.includes('=') || !text(value)) + || Buffer.byteLength(JSON.stringify(launch)) > LAUNCH_BYTES) throw new Error('Invalid or oversized confined worker launch') +} diff --git a/test/authority.test.mjs b/test/authority.test.mjs index 3434b9d3..db755a84 100644 --- a/test/authority.test.mjs +++ b/test/authority.test.mjs @@ -6,7 +6,7 @@ import Subprocess from '@deepseek-ai/dsh-subprocess-local' import { createScope } from '@deepseek-ai/dsh-scope' import { assert, test, report, withTempDir, skipSuite, waitFor } from './lib/harness.mjs' import { bootHost, shutdown } from './lib/host.mjs' -import { mountResident, mountFakeResident } from './lib/resident.mjs' +import { mountResident, mountFakeResident, mountExchangeResident } from './lib/resident.mjs' import { invoke as fileCall } from '../lib/file/worker.js' import { invoke as ptcCall } from '../lib/ptc/worker.js' import { tryResolveWorker } from '../lib/ptc/runtime.js' @@ -124,7 +124,8 @@ await test('PTC stderr restrictions do not prove selector denial; runner failure await ctx.plugin(inner => inner.provide('subprocess', { spawn: () => ({ done: Promise.resolve({ exitCode }), collected: { stdout: { readFrom: () => ({ text: '', lossy: false }) }, stderr: { readFrom: () => ({ text: stderr, lossy: false }) }, } }) })) - const call = () => ptcCall(ctx, 'unused', 'result', { action: 'page' }, execution(root), signal()) + await mountExchangeResident(ctx, { file: process.execPath, cwd: root }) + const call = () => ptcCall(ctx, process.execPath, 'result', { action: 'page' }, execution(root), signal()) let value = await call() assert.equal(value.error.code, 'EXECUTION_RESTRICTED') assert.equal(value.error.effectState, 'unknown'); assert.equal(value.error.permission, undefined) @@ -152,12 +153,13 @@ await test('PTC canonical containment rejects an external linked root without re stdout: { readFrom: () => ({ text: '{"status":"completed"}', lossy: false }) }, stderr: { readFrom: () => ({ text: '', lossy: false }) }, } } } })) - let value = await ptcCall(ctx, 'unused', 'result', { root: link, action: 'page' }, execution(root), signal()) + await mountExchangeResident(ctx, { file: process.execPath, cwd: root }) + let value = await ptcCall(ctx, process.execPath, 'result', { root: link, action: 'page' }, execution(root), signal()) assert.equal(value.error.code, 'POLICY_DENIED'); assert.equal(value.error.permission.required, 'danger-full-access') assert.equal(spawned, 0) for (const current of ['read-only', 'danger-full-access']) { mode = current - value = await ptcCall(ctx, 'unused', 'result', { root: link, action: 'page' }, execution(root), signal()) + value = await ptcCall(ctx, process.execPath, 'result', { root: link, action: 'page' }, execution(root), signal()) assert.equal(value.status, 'completed', JSON.stringify(value)); assert.equal(received.request.root, link) } } finally { await shutdown(ctx) } @@ -172,7 +174,8 @@ await test('PTC preparation snapshot changes stop before spawn without inventing await ctx.plugin(inner => inner.provide('sandboxPolicy', { resolve: () => ({ mode, workspaceRoot: root }) })) await ctx.plugin(inner => inner.provide('sandbox', { confine: async argv => { mode = 'danger-full-access'; return { argv, enforcement: 'full' } } })) await ctx.plugin(inner => inner.provide('subprocess', { spawn() { spawned++; throw new Error('must not spawn') } })) - const value = await ptcCall(ctx, 'unused', 'result', { action: 'page' }, execution(root), signal()) + await mountFakeResident(ctx, { file: process.execPath }) + const value = await ptcCall(ctx, process.execPath, 'result', { action: 'page' }, execution(root), signal()) assert.equal(value.error.code, 'AUTHORITY_CHANGED'); assert.equal(value.error.permission, undefined) assert.equal(spawned, 0) } finally { await shutdown(ctx) } diff --git a/test/lib/resident.mjs b/test/lib/resident.mjs index 96369500..257e578f 100644 --- a/test/lib/resident.mjs +++ b/test/lib/resident.mjs @@ -49,6 +49,11 @@ export function mountFakeResident(ctx, options = {}) { if (!options.invoke) throw new Error('Publication-only resident fixture must not execute') return options.invoke(envelope, signal, onStarted) }, + async invokeConfined(envelope, launch, signal, onStarted) { + if (!client.available) throw new ResidentNotStarted('Resident fixture is unavailable') + if (!options.invoke) throw new Error('Publication-only resident fixture must not execute') + return options.invoke(envelope, signal, onStarted, launch) + }, async stop() { client.available = false }, } inner.effect(() => () => client.stop(), 'fixture: resident kernel') @@ -65,8 +70,8 @@ export function providePublicationAuthorities(ctx, agents) { /** Convert existing raw-output subprocess mocks into explicit kernel receipts. */ export function mountExchangeResident(ctx, { file = 'unused', cwd = process.cwd() } = {}) { - return mountFakeResident(ctx, { file, invoke: (envelope, signal, started) => exchange(ctx.get('subprocess'), { - argv: [file], cwd, signal, graceMs: 1000, + return mountFakeResident(ctx, { file, invoke: (envelope, signal, started, launch) => exchange(ctx.get('subprocess'), { + argv: launch?.argv ?? [file], cwd: launch?.cwd ?? cwd, ...(launch ? { env: launch.env } : {}), signal, graceMs: 1000, stdio: { stdin: { data: JSON.stringify(envelope) }, stdout: { maxBytes: envelope.budget.maxBytes }, stderr: { maxBytes: 16384 } }, }, started) }) } diff --git a/test/native-result.test.mjs b/test/native-result.test.mjs index 734d337f..a363c1f3 100644 --- a/test/native-result.test.mjs +++ b/test/native-result.test.mjs @@ -48,8 +48,19 @@ requireWorker('native-result') let serial = 0 async function setup(root) { const { ctx } = await bootHost() - await ctx.plugin(Subprocess); await ctx.plugin(Sandbox) + let workerStarts = 0 + class Observed extends Subprocess { + spawn(spec) { + if (process.platform === 'win32' && spec.argv.includes(WORKER)) { + assert.deepEqual(spec.argv, [WORKER, '--serve'], 'Native/PTC result calls must not launch a public one-shot worker') + workerStarts++ + } + return super.spawn(spec) + } + } + await ctx.plugin(Observed); await ctx.plugin(Sandbox) await mountResident(ctx, { file: WORKER }) + ctx.effect(() => () => { if (process.platform === 'win32') assert.equal(workerStarts, 1) }) await ctx.plugin(inner => inner.provide('sandboxPolicy', { resolve: ({ session } = {}) => ({ mode: session?.mode ?? 'workspace-write', workspaceRoot: root, sessionId: session?.id }) })) inventory(ctx) const file = { fiber: await mountNativeCatalogue(ctx, { binaryPath: WORKER, required: true }) } diff --git a/test/ptc-audit-refs.test.mjs b/test/ptc-audit-refs.test.mjs index ec47ade3..e5f32d58 100644 --- a/test/ptc-audit-refs.test.mjs +++ b/test/ptc-audit-refs.test.mjs @@ -9,6 +9,7 @@ import { definition } from '../lib/ptc/contract.js' import { expireResultSession } from '../lib/kernel/result-lifecycle.js' import { assert, test, report, withTempDir } from './lib/harness.mjs' import { WORKER, requireWorker, callWorker, envelopeOf, request } from './lib/boundary.mjs' +import { mountResident } from './lib/resident.mjs' requireWorker('ptc-audit-refs') let serial = 0 @@ -17,29 +18,23 @@ async function withBridge(body) { fs.writeFileSync(path.join(root, 'source.txt'), 'original source\n') const context = new Context(); context.logger.level = 0 await context.plugin(Subprocess) - const real = context.get('subprocess') + const kernel = await mountResident(context, { file: WORKER, cwd: root }) + const original = kernel.invoke.bind(kernel) const session = { id: 'audit-refs-' + ++serial, header: { cwd: root } } const calls = [], cacheRoots = new Set() let nextFault + kernel.invoke = async (envelope, signal, started) => { + const fault = nextFault; nextFault = undefined + const call = { request: envelope.request, started: false }; calls.push(call) + if (envelope.storage?.root) cacheRoots.add(envelope.storage.root) + if (fault === 'before-spawn') throw new Error('fixture: admission refused before any worker was created') + const result = await original(envelope, signal, () => { call.started = true; started?.() }) + call.native = result.value + // The native leaf really persisted its plan/run. Substitute only its received exchange. + return !fault ? result : fault === 'lost-reply' ? { ...result, value: undefined, invalid: 'json' } : { ...result, value: fault } + } const ctx = { root: context.root, get(name) { if (name === 'sandboxPolicy') return { resolve: () => ({ mode: 'danger-full-access', workspaceRoot: root }) } - if (name === 'subprocess') return { spawn(spec) { - const envelope = JSON.parse(spec.stdio.stdin.data), fault = nextFault - nextFault = undefined - const call = { request: envelope.request, started: false }; calls.push(call) - if (envelope.storage?.root) cacheRoots.add(envelope.storage.root) - if (fault === 'before-spawn') throw new Error('fixture: spawn refused before any process was created') - const handle = real.spawn(spec); call.started = true - const done = handle.done.then(outcome => { - call.native = JSON.parse(handle.collected.stdout.readFrom(0).text) - return outcome - }) - const stdout = !fault ? handle.collected.stdout : { readFrom() { - // Native has really persisted its plan/run. Substitute only the received reply. - return { text: fault === 'lost-reply' ? 'incomplete JSON' : JSON.stringify(fault), lossy: false } - } } - return { done, collected: { stdout, stderr: handle.collected.stderr } } - } } return context.get(name) } } const api = { @@ -51,7 +46,7 @@ async function withBridge(body) { const entry = definition(tool, execute) const value = await (projected ? entry.execute(args, {}) : execute()) if (projected) assert.deepEqual(validateJsonSchemaValue(entry.output.schema, value), []) - assert.equal(calls.length, before + 1, 'one requested operation starts at most one worker; no audit probe') + assert.equal(calls.length, before + 1, 'one requested operation submits at most one resident request; no audit probe') return { value, call: calls.at(-1) } }, async page(id) { return (await this.call('result', { action: 'page', id })).value }, diff --git a/test/ptc.test.mjs b/test/ptc.test.mjs index 790d4e51..d73d4601 100644 --- a/test/ptc.test.mjs +++ b/test/ptc.test.mjs @@ -4,7 +4,7 @@ import { selectExecutor, executeCall, prepareStep } from './lib/executor.mjs' */ import { ownWindowsFixture, skipSuite, waitFor } from './lib/harness.mjs' import { workingDirectoryOver } from './lib/host.mjs' -import { mountResident, mountFakeResident } from './lib/resident.mjs' +import { mountResident, mountFakeResident, mountExchangeResident } from './lib/resident.mjs' import assert from 'node:assert/strict' import fs from 'node:fs' import os from 'node:os' @@ -37,8 +37,10 @@ const workerTriples = { 'darwin-arm64': 'aarch64-apple-darwin', 'darwin-x64': 'x 'linux-arm64': 'aarch64-unknown-linux-musl', 'linux-x64': 'x86_64-unknown-linux-musl', 'win32-arm64': 'aarch64-pc-windows-msvc', 'win32-x64': 'x86_64-pc-windows-msvc' } const workerName = process.platform === 'win32' ? 'dsh-ops-worker.exe' : 'dsh-ops-worker' +const stagedWorker = tryResolveWorker() const workerCandidates = [ ...(process.env.DSH_OPS_WORKER_BIN !== undefined ? [process.env.DSH_OPS_WORKER_BIN] : [ + ...(stagedWorker.ok ? [stagedWorker.file] : []), path.resolve('native/target/release', workerName), ...(workerTriples[`${process.platform}-${process.arch}`] ? [path.resolve('native/target', workerTriples[`${process.platform}-${process.arch}`], 'release', workerName)] : []), ]), @@ -236,7 +238,7 @@ try { } }) }) const live = []; ctx.provide('agents', { list: () => live }) - await mountFakeResident(ctx, { file }) + await mountExchangeResident(ctx, { file, cwd: dir }) const component = await ctx.plugin(ptc, { binaryPath: file, required: true, alwaysLoaded: ['batch', 'result'] }) const makeAgent = async (id, presentation, parent) => { const agent = { id, session: { id, mode: 'workspace-write', header: { cwd: dir } } } @@ -303,6 +305,13 @@ try { assert.equal((await direct(pure, 'ops_batch', {action:'run',operations:[{op:'remove',path:'host-dest'}]})).error.code,'EXECUTOR_UNAVAILABLE') assert.equal(calls.length,before); assert.equal(hostJobs.size,0) const fakeOutcome = (outcome, stderr = '', lossy = false) => ({ get(name) { + if (name === 'opsKernel') { + const invoke = async (_envelope, _signal, started) => { + started?.() + return { exitCode: outcome.exitCode, interrupted: Boolean(outcome.signal), readers: true, lossy, stderr, value: {} } + } + return { file, available: true, invoke, invokeConfined: (envelope, _launch, signal, started) => invoke(envelope, signal, started) } + } if (name === 'subprocess') return { spawn() { return { done: Promise.resolve(outcome), collected: { stdout: { readFrom: () => ({ text: '{}', lossy }) }, stderr: { readFrom: () => ({ text: stderr }) } } } } } if (name === 'sandbox') return { confine: async argv => ({ argv, enforcement: 'full', denialSignatures: ['fixture access denied'], @@ -355,6 +364,11 @@ try { const workerStarted = new Promise(resolve => { enteredWorker = resolve }) const interruptedCtx = { get(name) { if (name === 'sandboxPolicy') return { resolve: () => ({mode:'danger-full-access',workspaceRoot:dir}) } + if (name === 'opsKernel') return { file, available: true, invoke(_envelope, _signal, started) { + started?.(); enteredWorker() + return new Promise(resolve => abortDuringWork.signal.addEventListener('abort', () => resolve({ exitCode: null, + interrupted: true, readers: true, lossy: false, stderr: '' }), { once: true })) + } } if (name === 'subprocess') return { spawn() { enteredWorker() return { collected: { stdout: {readFrom:()=>({text:'',lossy:false})},stderr:{readFrom:()=>({text:'',lossy:false})} }, diff --git a/test/resident-kernel.test.mjs b/test/resident-kernel.test.mjs index 0b783768..f158e4bb 100644 --- a/test/resident-kernel.test.mjs +++ b/test/resident-kernel.test.mjs @@ -23,7 +23,7 @@ async function within(work, ms = 10_000, label = 'resident check') { })]) } finally { clearTimeout(timer) } } -function fixture({ hold = false, ready = true, holdExit = false } = {}) { +function fixture({ hold = false, ready = true, holdExit = false, capabilities = [] } = {}) { const input = new PassThrough(), frames = [], writes = [], effects = [] let done, exit, terminated = 0, spawns = 0, active = true const finished = new Promise(resolve => { done = resolve }), empty = new Promise(resolve => { exit = resolve }) @@ -56,13 +56,53 @@ function fixture({ hold = false, ready = true, holdExit = false } = {}) { } const starting = startResident(ctx, '/worker', '/workspace') const send = value => input.write(`${JSON.stringify(value)}\n`) - if (ready) send({ type: 'ready', version: 1, pid: 123 }) + if (ready) send({ type: 'ready', version: 1, pid: 123, capabilities }) return { ctx, starting, send, input, frames, writes, handle, finish, releaseExit: () => exit(true), get spawns() { return spawns }, get terminated() { return terminated }, get spec() { return spec }, async dispose() { active = false; await Promise.all(effects.map(effect => effect())) }, } } +await test('confined worker capability and launch bounds refuse before admission without fallback', async () => { + const host = fixture(), client = await host.starting + const request = { ...envelope('confined-old'), component: 'ptc', request: { tool: 'batch', args: { action: 'plan' } } } + const launch = { argv: ['/trusted-runner', '--', '/worker'], cwd: '/workspace', env: { PATH: '/bin' } } + try { + await assert.rejects(client.invokeConfined(request, launch), /confined-worker-v1/) + assert.equal(host.frames.length, 0); assert.equal(host.spawns, 1) + } finally { await host.dispose() } + const current = fixture({ capabilities: ['confined-worker-v1'] }), ready = await current.starting + try { + for (const value of [{ ...launch, argv: [] }, { ...launch, argv: ['relative'] }, + { ...launch, env: { SECRET: 'x'.repeat(65536) } }, { ...launch, env: { 'bad=key': 'x' } }, { ...launch, cwd: 'relative' }]) { + await assert.rejects(ready.invokeConfined(request, value), /Invalid or oversized/) + } + await assert.rejects(ready.invokeConfined(envelope('wrong-component'), launch), /trusted batch\/result/) + assert.equal(current.frames.length, 0); assert.equal(current.spawns, 1) + } finally { await current.dispose() } +}) + +await test('confined calls share existing admission, start receipts and targeted cancellation', async () => { + const host = fixture({ capabilities: ['confined-worker-v1'] }), client = await host.starting + const controller = new AbortController(), launch = { argv: ['/trusted-runner', '--', '/worker'], cwd: '/workspace', env: { PATH: '/bin' } } + const request = { ...envelope('confined'), component: 'ptc', request: { tool: 'result', args: { action: 'page' } } } + let started = 0 + try { + const confined = client.invokeConfined(request, launch, controller.signal, () => ++started) + assert.deepEqual(host.frames[0], { type: 'execute', envelope: request, launch }) + const siblings = Array.from({ length: 31 }, (_, index) => client.invoke(envelope(`sibling-${index}`))) + await assert.rejects(client.invoke(envelope('overflow')), ResidentNotStarted) + host.send({ type: 'started', requestId: 'confined' }); await flush(); assert.equal(started, 1) + controller.abort(); await flush() + assert.deepEqual(host.frames.filter(frame => frame.type === 'cancel'), [{ type: 'cancel', requestId: 'confined' }]) + assert.equal(host.terminated, 0) + host.send({ type: 'completed', requestId: 'confined', exchange: { ...receipt, interrupted: true } }) + assert.equal((await confined).interrupted, true) + for (let index = 0; index < 31; index++) host.send({ type: 'completed', requestId: `sibling-${index}`, exchange: receipt }) + await Promise.all(siblings); assert.equal(client.available, true); assert.equal(host.spawns, 1) + } finally { await host.dispose() } +}) + await test('forty simultaneous session expiry notifications drain through the bounded client without lost purges', async () => { const host = fixture(), client = await host.starting let completed = 0, cursor = 0, settled = false diff --git a/test/resident-ptc-routing.test.mjs b/test/resident-ptc-routing.test.mjs new file mode 100644 index 00000000..54245df7 --- /dev/null +++ b/test/resident-ptc-routing.test.mjs @@ -0,0 +1,197 @@ +/** Actual public policy/confinement and resident transport; no shell or model API. */ +import fs from 'node:fs' +import path from 'node:path' +import Subprocess from '@deepseek-ai/dsh-subprocess-local' +import Sandbox from '@deepseek-ai/dsh-sandbox-local' +import Policy, { setSandboxMode } from '@deepseek-ai/dsh-sandbox-policy' +import { residentKernel } from '../lib/integration/index.js' +import { invoke } from '../lib/execution/jobs.js' +import { NativeFileTools } from '../lib/projections/native/runtime.js' +import { workerEnvelope } from '../lib/execution/protocol.js' +import { confinedWorkerLaunch } from '../lib/execution/resident/worker-launch.js' +import { createResidentTaskJobs } from './lib/resident-task-jobs.mjs' +import { assert, test, report, withTempDir, waitFor } from './lib/harness.mjs' +import { WORKER, requireWorker } from './lib/boundary.mjs' + +requireWorker('resident-ptc-routing') +async function fixture(root, mode = 'workspace-write', sandboxConfig = {}) { + const host = await createResidentTaskJobs({ cwd: root }), starts = [] + class Observed extends Subprocess { + spawn(spec) { const handle = super.spawn(spec); starts.push({ argv: [...spec.argv], handle }); return handle } + } + await host.ctx.plugin(Policy, { mode, workspaceRoot: root }) + await host.ctx.plugin(Observed); await host.ctx.plugin(Sandbox, sandboxConfig) + const provider = await host.ctx.plugin(residentKernel({ binaryPath: WORKER })) + const kernel = host.ctx.get('opsKernel'), native = new NativeFileTools({ ctx: host.ctx, config: { workerPath: WORKER, required: true } }) + const confined = [], launches = [], sandbox = host.ctx.get('sandbox'), wrap = sandbox.confine.bind(sandbox) + sandbox.confine = async (...args) => { + const value = await wrap(...args); confined.push({ argv: [...value.argv], enforcement: value.enforcement }); return value + } + if (typeof kernel.invokeConfined === 'function') { + const invokeConfined = kernel.invokeConfined.bind(kernel) + kernel.invokeConfined = (envelope, launch, signal, started) => { + launches.push(structuredClone(launch)); return invokeConfined(envelope, launch, signal, started) + } + } + await native.start([]) + const call = (tool, args, signal = new AbortController().signal, agent = host.agent) => + invoke(host.ctx, WORKER, tool, args, { agent }, signal, { denyCredentialReads: true }) + return { ...host, starts, kernel, provider, native, call, confined, launches, + async close() { await native.stop(); await host.dispose() }, + assertResidentOnly() { + assert.deepEqual(starts.map(value => value.argv), [[WORKER, '--serve']], 'calls and disposal must not start a public one-shot worker') + } } +} + +await test('full-access batch plan/run/resume and audit paging reuse the kernel', () => withTempDir('resident-ptc-full', async root => { + fs.writeFileSync(path.join(root, 'one.txt'), 'one'); fs.writeFileSync(path.join(root, 'two.txt'), 'two') + const host = await fixture(root, 'danger-full-access') + try { + const planned = await host.call('batch', { action: 'plan', operations: [{ op: 'copy', path: 'one.txt', to: 'copy.txt' }] }) + assert.equal(planned.status, 'planned', JSON.stringify(planned)); assert.equal(fs.existsSync(path.join(root, 'copy.txt')), false) + const resumed = await host.call('batch', { action: 'resume', id: planned.detailsRef }) + assert.equal(resumed.status, 'completed', JSON.stringify(resumed)); assert.equal(fs.readFileSync(path.join(root, 'copy.txt'), 'utf8'), 'one') + const ran = await host.call('batch', { action: 'run', operations: [{ op: 'move', path: 'two.txt', to: 'moved.txt' }] }) + assert.equal(ran.status, 'completed', JSON.stringify(ran)); assert.equal(fs.existsSync(path.join(root, 'two.txt')), false) + const audit = await host.call('result', { action: 'page', id: ran.detailsRef }) + assert.equal(audit.items[0].state, 'completed', JSON.stringify(audit)) + host.assertResidentOnly() + await host.provider.dispose() + const unavailable = await host.call('result', { action: 'page', id: ran.detailsRef }) + assert.equal(unavailable.error.code, 'EXECUTOR_UNAVAILABLE'); assert.equal(unavailable.error.effectState, 'none') + host.assertResidentOnly() + } finally { await host.close() } + host.assertResidentOnly(); assert.equal(await host.starts[0].handle.waitForExit(), true) +})) + +await test('restricted results and batches retain public confinement, owner checks and read-only paging', () => withTempDir('resident-ptc-restricted', async root => { + fs.writeFileSync(path.join(root, 'source.txt'), 'source') + const host = await fixture(root) + try { + const stored = await host.call('result', { action: 'put', items: [{ n: 1, text: 'x'.repeat(4000) }, { n: 2 }] }) + assert.match(stored.id, /^ref:\/\//, JSON.stringify(stored)) + const page = await host.native.call('result', { action: 'page', id: stored.id, fields: ['n'] }, { agent: host.agent, signal: new AbortController().signal }) + assert.deepEqual(page.items, [{ n: 1 }, { n: 2 }]) + const sibling = await host.createAgent() + const foreign = await host.call('result', { action: 'page', id: stored.id }, undefined, sibling.agent) + assert.ok(foreign.error, 'a reference never transfers session authority') + const planned = await host.call('batch', { action: 'plan', operations: [{ op: 'copy', path: 'source.txt', to: 'copy.txt' }] }) + assert.equal(planned.status, 'planned', JSON.stringify(planned)) + const resumed = await host.call('batch', { action: 'resume', id: planned.detailsRef }) + assert.equal(resumed.status, 'completed', JSON.stringify(resumed)) + const ran = await host.call('batch', { action: 'run', operations: [{ op: 'remove', path: 'copy.txt' }] }) + assert.equal(ran.status, 'completed', JSON.stringify(ran)) + setSandboxMode(host.agent.session, 'read-only') + const readonly = await host.call('result', { action: 'page', id: planned.detailsRef }) + assert.equal(readonly.items[0].state, 'completed', JSON.stringify(readonly)) + assert.ok((await host.call('result', { action: 'page', id: stored.id })).error, 'cache authority includes the permission mode') + for (const [tool, args] of [['result', { action: 'put', items: [] }], ['result', { action: 'release', id: stored.id }], + ['batch', { action: 'run', operations: [{ op: 'remove', path: 'source.txt' }] }]]) { + assert.equal((await host.call(tool, args)).error.code, 'POLICY_DENIED') + } + setSandboxMode(host.agent.session, 'workspace-write') + assert.equal((await host.call('result', { action: 'release', id: stored.id })).status, 'released') + assert.ok((await host.call('result', { action: 'page', id: stored.id })).error) + assert.equal(fs.readFileSync(path.join(root, 'source.txt'), 'utf8'), 'source') + if (process.platform === 'win32') { + host.assertResidentOnly(); assert.ok(host.launches.length >= 8) + for (const launch of host.launches) { + assert.ok(host.confined.some(value => JSON.stringify(value.argv) === JSON.stringify(launch.argv)), 'preserve exact public confinement argv') + assert.equal(launch.cwd, root); assert.equal(Object.keys(launch.env).some(key => key.toUpperCase() === 'DSH_SUBPROCESS_CONTROL'), false) + } + assert.equal(stored.sandbox.enforcement, host.confined[0].enforcement) + } + } finally { await host.close() } + if (process.platform === 'win32') host.assertResidentOnly() +})) + +for (const mode of ['danger-full-access', ...(process.platform === 'win32' ? ['workspace-write'] : [])]) +await test(`${mode} cancellation after resident start keeps unknown batch effects without public-worker replay`, () => withTempDir('resident-ptc-cancel', async root => { + fs.writeFileSync(path.join(root, 'source.txt'), 'safe') + const host = await fixture(root, mode), controller = new AbortController() + const original = host.kernel.invoke.bind(host.kernel), confined = host.kernel.invokeConfined?.bind(host.kernel) + const start = (envelope, started) => () => { started?.(); if (envelope.request.tool === 'batch') controller.abort(new Error('cancel admitted batch')) } + host.kernel.invoke = (envelope, signal, started) => original(envelope, signal, start(envelope, started)) + if (confined) host.kernel.invokeConfined = (envelope, launch, signal, started) => confined(envelope, launch, signal, start(envelope, started)) + try { + const value = await host.call('batch', { action: 'run', operations: [{ op: 'copy', path: 'source.txt', to: 'copy.txt' }] }, controller.signal) + assert.equal(value.error?.code, 'RESULT_UNKNOWN', JSON.stringify(value)); assert.equal(value.error.effectState, 'unknown') + assert.match(value.detailsRef, /^j-/) + const fresh = await host.call('batch', { action: 'plan', operations: [{ op: 'copy', path: 'source.txt', to: 'later.txt' }] }) + assert.equal(fresh.status, 'planned', JSON.stringify(fresh)); host.assertResidentOnly() + } finally { await host.close() } +})) + +if (process.platform === 'win32') await test('exact OS confinement refuses a root junction swapped after host preparation', () => withTempDir('resident-ptc-root-race', async base => { + const root = path.join(base, 'workspace'), inside = path.join(root, 'inside'), outside = path.join(base, 'outside') + fs.mkdirSync(inside, { recursive: true }); fs.mkdirSync(outside) + fs.writeFileSync(path.join(inside, 'source.txt'), 'inside'); fs.writeFileSync(path.join(outside, 'source.txt'), 'outside') + const alias = path.join(root, 'selected'); fs.symlinkSync(inside, alias, 'junction') + const host = await fixture(root), original = host.kernel.invokeConfined.bind(host.kernel) + let swapped = false + host.kernel.invokeConfined = (envelope, launch, signal, started) => { + if (envelope.request.tool === 'batch' && envelope.request.root === alias) { + fs.unlinkSync(alias); fs.symlinkSync(outside, alias, 'junction'); swapped = true + } + return original(envelope, launch, signal, started) + } + try { + const value = await host.call('batch', { root: alias, action: 'run', operations: [{ op: 'copy', path: 'source.txt', to: 'escaped.txt' }] }) + assert.equal(swapped, true, 'swap occurs only after the host admitted the original inside root') + assert.notEqual(value.status, 'completed', JSON.stringify(value)) + assert.equal(fs.existsSync(path.join(outside, 'escaped.txt')), false) + assert.equal(fs.existsSync(path.join(outside, '.dsh-ops')), false) + assert.equal(fs.readFileSync(path.join(outside, 'source.txt'), 'utf8'), 'outside') + assert.equal(host.kernel.available, true); host.assertResidentOnly() + } finally { await host.close() } +})) + +if (process.platform === 'win32') await test('confined stdin starvation expires the native budget and joins descendants without relying on caller timeout', () => withTempDir('resident-ptc-input-budget', async root => { + const host = await fixture(root), marker = path.join(root, 'owned-pids.json'), controller = new AbortController() + const script = `const fs=require('node:fs'),{spawn}=require('node:child_process'); const child=spawn(process.execPath,['-e','setInterval(()=>{},1000)'],{stdio:'ignore',windowsHide:true}); fs.writeFileSync(${JSON.stringify(marker)},JSON.stringify([process.pid,child.pid])); setInterval(()=>{},1000)` + const policy = host.ctx.get('sandboxPolicy').resolve({ session: host.agent.session }) + const confined = await host.ctx.get('sandbox').confine([process.execPath, '-e', script], policy) + const launch = await confinedWorkerLaunch(confined.argv, root) + const envelope = workerEnvelope('ptc', { root, scope: host.agent.session.id, tool: 'result', + args: { action: 'put', items: [{ value: 'x'.repeat(1800000) }] }, read_only: false }, 1500) + let watchdog = false, pending, timer + const alive = pid => { try { process.kill(pid, 0); return true } catch { return false } } + try { + timer = setTimeout(() => { watchdog = true; controller.abort(new Error('external cleanup watchdog')) }, 6000) + pending = host.kernel.invokeConfined(envelope, launch, controller.signal) + await waitFor(() => fs.existsSync(marker), { timeoutMs: 5000, what: 'owned stdin-stalled process and descendant' }) + const pids = JSON.parse(fs.readFileSync(marker, 'utf8')) + assert.equal(pids.length, 2); assert.ok(pids.every(alive)) + const value = await pending + clearTimeout(timer) + await waitFor(() => pids.every(pid => !alive(pid)), { timeoutMs: 2000, what: 'stdin-stalled owned descendants to exit' }) + assert.equal(watchdog, false, 'native envelope deadline must settle before the external cleanup watchdog') + assert.equal(value.interrupted, true); assert.equal(value.value, undefined) + fs.writeFileSync(path.join(root, 'source.txt'), 'safe') + assert.equal((await host.call('batch', { action: 'plan', operations: [{ op: 'copy', path: 'source.txt', to: 'after.txt' }] })).status, 'planned') + host.assertResidentOnly() + } finally { clearTimeout(timer); controller.abort(); try { await pending } finally { await host.close() } } +})) +if (process.platform === 'win32') for (const [signature, code, effect] of [ + ['fixture runner failed', 'EXECUTOR_UNAVAILABLE', 'none'], + ['permission denied', 'EXECUTION_RESTRICTED', 'unknown'], +]) await test(`early runner refusal preserves ${code} despite undelivered large stdin`, () => withTempDir('resident-ptc-refusal', async root => { + // A real operator-configured public runner exits before reading any request bytes. + // The fixture produces diagnostics only; it never invokes the worker or applies effects. + const runner = path.join(root, 'refusing-runner.cjs') + fs.writeFileSync(runner, `require('node:fs').writeSync(2,${JSON.stringify(signature + '\n')}); process.exit(23)`) + const host = await fixture(root, 'workspace-write', { + runnerCommand: [process.execPath, runner], runnerFailureSignatures: ['fixture runner failed'], + }) + const original = host.kernel.invokeConfined.bind(host.kernel), receipts = [] + host.kernel.invokeConfined = async (...args) => { const value = await original(...args); receipts.push(value); return value } + try { + const value = await host.call('batch', { action: 'plan', operations: [{ op: 'remove', path: 'x'.repeat(1800000) }] }) + assert.equal(value.error?.code, code, JSON.stringify(value)); assert.equal(value.error.effectState, effect) + assert.equal(receipts.length, 1); assert.equal(receipts[0].exitCode, 23) + assert.equal(receipts[0].interrupted, false); assert.equal(receipts[0].stderr.trim(), signature) + assert.equal(fs.existsSync(path.join(root, '.dsh-ops')), false) + assert.equal(host.kernel.available, true); host.assertResidentOnly() + } finally { await host.close() } +})) +report('resident-ptc-routing') diff --git a/test/resident-worker-launch.test.mjs b/test/resident-worker-launch.test.mjs new file mode 100644 index 00000000..2989fa8b --- /dev/null +++ b/test/resident-worker-launch.test.mjs @@ -0,0 +1,42 @@ +/** Match the public subprocess base without forwarding ambient host secrets. */ +import { scrubbedParentEnv } from '@deepseek-ai/dsh-subprocess' +import { confinedWorkerLaunch, validateWorkerLaunch } from '../lib/execution/resident/worker-launch.js' +import { assert, test, report } from './lib/harness.mjs' + +await test('confined worker launch preserves exact argv/cwd and the current public scrubbed environment', async () => { + const names = ['AWS_SECRET_ACCESS_KEY', 'DSH_SUBPROCESS_CONTROL'], before = names.map(name => process.env[name]) + try { + for (const name of names) process.env[name] = 'fixture-not-forwarded' + const argv = [process.execPath, 'trusted-runner', '--', '/fixed-worker'], cwd = process.cwd() + const expected = scrubbedParentEnv(), launch = await confinedWorkerLaunch(argv, cwd) + assert.deepEqual(launch.argv, argv); assert.notEqual(launch.argv, argv); assert.equal(launch.cwd, cwd) + assert.deepEqual(launch.env, expected) + for (const name of names) assert.equal(launch.env[name], undefined) + } finally { + names.forEach((name, index) => { if (before[index] === undefined) delete process.env[name]; else process.env[name] = before[index] }) + } +}) + +await test('only the actual Electron executable receives its required Node-runner environment', async () => { + const descriptor = Object.getOwnPropertyDescriptor(process.versions, 'electron') + Object.defineProperty(process.versions, 'electron', { value: '44.0.0', configurable: true }) + try { + const launch = await confinedWorkerLaunch([process.execPath, 'trusted-runner', '--', '/fixed-worker'], process.cwd()) + const expected = Object.fromEntries(Object.entries(scrubbedParentEnv()).filter(([key]) => + (process.platform === 'win32' ? key.toUpperCase() : key) !== 'ELECTRON_RUN_AS_NODE')) + assert.deepEqual(launch.env, { ...expected, ELECTRON_RUN_AS_NODE: '1' }) + const external = await confinedWorkerLaunch(['/other-executable'], process.cwd()) + assert.deepEqual(external.env, scrubbedParentEnv()) + } finally { + if (descriptor) Object.defineProperty(process.versions, 'electron', descriptor) + else delete process.versions.electron + } +}) + +await test('launch validation refuses oversized or malformed facts before a transport can admit them', async () => { + await assert.rejects(confinedWorkerLaunch([process.execPath, 'x'.repeat(65536)], process.cwd()), /oversized/) + const launch = { argv: [process.execPath], cwd: process.cwd(), env: {} } + for (const value of [{ ...launch, env: { PATH: '\0' } }, { ...launch, env: { 'bad=key': 'value' } }, + { ...launch, argv: ['relative'] }, { ...launch, cwd: 'relative' }]) assert.throws(() => validateWorkerLaunch(value), /Invalid/) +}) +report('resident-worker-launch')