Skip to main content

running_process_platform_internal/
lib.rs

1//! Blessed asynchronous process operations.
2//!
3//! This crate is intentionally published as an implementation detail. It is
4//! the only production owner of the Tokio process primitives used by the
5//! async process API. Higher layers receive typed operations and never name
6//! `tokio::process::Command` directly.
7
8use std::cfg_select;
9#[cfg(feature = "async-process")]
10use std::ffi::{OsStr, OsString};
11#[cfg(feature = "async-process")]
12use std::io;
13#[cfg(feature = "async-process")]
14use std::path::PathBuf;
15#[cfg(feature = "async-process")]
16use std::process::{ExitStatus, Output, Stdio};
17
18#[cfg(feature = "async-process")]
19use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
20#[cfg(feature = "async-process")]
21use tokio::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command};
22
23/// Neutral capability indexes for the eventual workspace-wide host boundary.
24///
25/// The indexes intentionally expose no operations yet: phase 2 establishes
26/// ownership names before later phases move a capability behind them.
27pub mod platform;
28
29#[cfg(all(feature = "independent-spawn", test, not(windows)))]
30#[path = "platform_win/scheduler_error.rs"]
31mod scheduler_error;
32#[cfg(feature = "independent-spawn")]
33pub(crate) use platform_imp::spawn_sync_owned_daemon;
34#[cfg(feature = "independent-spawn")]
35pub use platform_imp::{
36    independent_broker_run, independent_broker_spawn, independent_spawn, IndependentChild,
37};
38#[cfg(feature = "independent-spawn")]
39pub(crate) use platform_imp::{independent_open_regular, INDEPENDENT_ZERO_WRITE_PENDING};
40
41/// Temporary source-compatibility re-export for the pre-boundary PTY API.
42///
43/// New code must use [`platform::terminal`] facade-owned types. This root-only
44/// alias deliberately stays outside the neutral facade and can be removed in
45/// the next major release after downstream users have migrated.
46#[cfg(feature = "pty")]
47#[doc(hidden)]
48pub use portable_pty as portable_pty_compat;
49
50// This is deliberately the crate's only host selector.  Facade modules are
51// neutral; native details live behind the selected private root.
52cfg_select! {
53    target_os = "windows" => {
54        mod platform_win;
55        pub(crate) use platform_win as platform_imp;
56    }
57    target_os = "linux" => {
58        mod platform_linux;
59        pub(crate) use platform_linux as platform_imp;
60    }
61    target_os = "macos" => {
62        mod platform_macos;
63        pub(crate) use platform_macos as platform_imp;
64    }
65}
66
67// Re-export the selected implementation once from this allowed host-selector
68// root. Neutral capability facades re-export only crate-root names and never
69// name the private `platform_imp` alias themselves.
70pub use platform_imp::{
71    assign_child_to_windows_job, cancel_capture_reader, canonical_environment_pairs,
72    capture_reader_done, compat_shell_command, configure_exact_trace, configure_process_command,
73    configure_process_command_for_bounded_owner_death, configure_sync_contained_command,
74    configure_sync_daemon_command, configure_sync_daemon_command_with_inheritance,
75    configure_trampoline_command, current_executable_build_id, exact_trace_capability, exit_code,
76    monitor_console_windows, parent_has_console, prepare_capture_reader, set_process_name,
77    shell_command, soft_terminate_process_group, spawn_sync, spawn_sync_daemon,
78    spawn_sync_daemon_with_inheritance, start_descendant_monitor, start_exact_trace,
79    sync_child_native_handle, trampoline_exit_code, unix_mark_extra_fds_close_on_exec,
80    unix_set_priority, unix_signal_process, unix_signal_process_group, unix_signal_raw,
81    CaptureCancellation, TracedChild, WindowsJobHandle,
82};
83
84#[cfg(feature = "terminal-graphics")]
85pub use platform_imp::active_graphics_probe;
86
87#[cfg(feature = "window-icon")]
88pub use platform_imp::{set_window_icon_impl, window_icon_support_impl};
89
90#[cfg(feature = "async-process")]
91pub(crate) use platform_imp::{
92    async_child_cpu_time, async_child_identity, signal_async_child, signal_async_child_group,
93    AsyncChildIdentity,
94};
95
96#[cfg(feature = "process-inspection")]
97pub use platform_imp::{kill_tree, process_snapshot, process_snapshot_for_pid};
98
99pub use platform_imp::{autostart_register, autostart_render_registration, autostart_unregister};
100
101pub use platform_imp::{process_install_owner_death_cleanup, process_owner_death_cleanup_target};
102
103pub use platform_imp::process_install_shutdown_request_handler;
104
105pub use platform_imp::fs_write_all_to_descriptor;
106
107pub use platform_imp::{process_can_replace_current_image, process_replace_current_image};
108
109pub use platform_imp::{
110    process_executable_path, process_force_kill, process_same_executable_path,
111    process_signal_terminate, ProcessLiveness,
112};
113
114pub use platform_imp::{
115    resources_fd_exhaustion_error, resources_inode_capacity, resources_signals_fd_exhaustion,
116    resources_signals_storage_exhaustion, resources_storage_exhaustion_error,
117};
118
119pub use platform_imp::{
120    executable_file_name, executable_sibling_of_current_image, EXECUTABLE_EXTENSION,
121};
122
123#[cfg(feature = "fs")]
124pub use platform_imp::{
125    fs_create_private_file, fs_decode_path_bytes, fs_encode_path_bytes, fs_file_identity,
126    fs_is_lock_conflict, fs_open_lock_file, fs_path_identity, fs_replace_file, fs_sync_directory,
127    fs_try_lock_exclusive, fs_unlock, fs_user_config_dir, fs_user_data_dir, fs_user_run_data_root,
128    fs_user_runtime_dir, fs_user_state_dir, FsFileIdentity,
129};
130
131pub use platform_imp::{
132    host_boot_id, host_current_process_privilege, host_environment_keys_are_case_insensitive,
133    host_filesystem_device_id, host_hostname, host_login_environment, host_machine_id,
134    host_namespace_id, host_user_machine_identity, HostPrivilegedIdentity,
135};
136
137pub use platform_imp::host_login_environment_block;
138
139pub use platform_imp::terminal_input;
140
141#[cfg(feature = "ipc")]
142pub use platform_imp::{
143    ipc_broker_endpoint_name as IpcBrokerEndpointName, ipc_broker_v1_endpoint_path,
144    ipc_broker_v2_runtime_dir, ipc_current_user_id, ipc_endpoint_is_filesystem_backed,
145    ipc_endpoint_name_limit, ipc_endpoint_scope_bytes, ipc_nonblocking_zero_read_is_pending,
146    ipc_select_endpoint_address, IpcEndpoint, IpcInheritedListener, IpcListener,
147    IpcListenerNonblockingMode, IpcPeerIdentity, IpcPeerIdentitySource, IpcStream,
148};
149
150#[cfg(feature = "private-dir")]
151pub use platform_imp::{
152    private_dir_ensure_owner_private_directory, private_dir_owner_private_directory,
153};
154
155// Retain the implementation-detail root aliases selected by the historical
156// `ipc` capability.  New callers use `platform::private_dir` instead.
157#[cfg(feature = "ipc")]
158pub use platform_imp::{
159    private_dir_ensure_owner_private_directory as ipc_ensure_owner_private_directory,
160    private_dir_owner_private_directory as ipc_owner_private_directory,
161};
162
163/// Failure details for the deprecated 4.x raw descriptor/handle handoff API.
164///
165/// This type exists only at the crate-root compatibility boundary. New product
166/// mechanics use opaque [`platform::ipc::Stream`] operations instead.
167#[cfg(feature = "ipc")]
168#[doc(hidden)]
169#[derive(Clone, Debug, PartialEq, Eq)]
170pub struct LegacyHandoffError {
171    kind: platform::ipc::HandoffTransferErrorKind,
172    raw_os_error: Option<i32>,
173    transferred_bytes: Option<usize>,
174    expected_bytes: Option<usize>,
175    detail: Option<String>,
176}
177
178#[cfg(feature = "ipc")]
179impl LegacyHandoffError {
180    pub(crate) fn new(
181        kind: platform::ipc::HandoffTransferErrorKind,
182        raw_os_error: Option<i32>,
183    ) -> Self {
184        Self {
185            kind,
186            raw_os_error,
187            transferred_bytes: None,
188            expected_bytes: None,
189            detail: None,
190        }
191    }
192
193    pub(crate) fn with_detail(
194        kind: platform::ipc::HandoffTransferErrorKind,
195        raw_os_error: Option<i32>,
196        detail: impl Into<String>,
197    ) -> Self {
198        Self {
199            kind,
200            raw_os_error,
201            transferred_bytes: None,
202            expected_bytes: None,
203            detail: Some(detail.into()),
204        }
205    }
206
207    #[doc(hidden)]
208    pub fn partial(transferred_bytes: usize, expected_bytes: usize) -> Self {
209        Self {
210            kind: platform::ipc::HandoffTransferErrorKind::Failed,
211            raw_os_error: None,
212            transferred_bytes: Some(transferred_bytes),
213            expected_bytes: Some(expected_bytes),
214            detail: Some(format!(
215                "SCM_RIGHTS connection transfer was partial ({transferred_bytes}/{expected_bytes} bytes)"
216            )),
217        }
218    }
219
220    /// Return the policy-neutral failure category.
221    pub fn kind(&self) -> platform::ipc::HandoffTransferErrorKind {
222        self.kind
223    }
224
225    /// Return the native error code retained for legacy public diagnostics.
226    pub fn raw_os_error(&self) -> Option<i32> {
227        self.raw_os_error
228    }
229
230    /// Return a partial payload count when the descriptor may have transferred.
231    pub fn partial_counts(&self) -> Option<(usize, usize)> {
232        self.transferred_bytes.zip(self.expected_bytes)
233    }
234
235    pub(crate) fn detail(&self) -> Option<&str> {
236        self.detail.as_deref()
237    }
238}
239
240/// Whether the deprecated 4.x SCM_RIGHTS compatibility transport is available.
241#[cfg(feature = "ipc")]
242#[doc(hidden)]
243pub const LEGACY_SCM_RIGHTS_TRANSPORT_SUPPORTED: bool =
244    platform_imp::LEGACY_SCM_RIGHTS_TRANSPORT_SUPPORTED;
245
246/// Whether the deprecated 4.x DuplicateHandle compatibility transport is available.
247#[cfg(feature = "ipc")]
248#[doc(hidden)]
249pub const LEGACY_DUPLICATE_HANDLE_TRANSPORT_SUPPORTED: bool =
250    platform_imp::LEGACY_DUPLICATE_HANDLE_TRANSPORT_SUPPORTED;
251
252/// Root-only adapter for the deprecated raw-descriptor handoff API.
253#[cfg(feature = "ipc")]
254#[doc(hidden)]
255pub fn legacy_send_fd_to(
256    socket: &std::path::Path,
257    sent_fd: i32,
258    payload: &[u8],
259) -> Result<(), LegacyHandoffError> {
260    platform_imp::legacy_send_fd_to(socket, sent_fd, payload)
261}
262
263/// Root-only adapter for the deprecated connected raw-descriptor handoff API.
264#[cfg(feature = "ipc")]
265#[doc(hidden)]
266pub fn legacy_send_fd_over(
267    socket_fd: i32,
268    sent_fd: i32,
269    payload: &[u8],
270) -> Result<(), LegacyHandoffError> {
271    platform_imp::legacy_send_fd_over(socket_fd, sent_fd, payload)
272}
273
274/// Root-only adapter for the deprecated raw-handle duplication API.
275#[cfg(feature = "ipc")]
276#[doc(hidden)]
277pub fn legacy_duplicate_handle(
278    source_handle: usize,
279    backend_pid: u32,
280) -> Result<usize, LegacyHandoffError> {
281    platform_imp::legacy_duplicate_handle(source_handle, backend_pid)
282}
283
284/// Temporary source-compatibility conversion for public APIs that predate the
285/// opaque IPC facade.
286///
287/// New code must keep [`IpcStream`] opaque. This root-only adapter exists so
288/// `running-process` can preserve its established raw-stream callback contract
289/// until the next major release without exposing the transport through
290/// [`platform::ipc`].
291#[cfg(feature = "ipc")]
292#[doc(hidden)]
293pub fn into_legacy_ipc_stream(stream: IpcStream) -> interprocess::local_socket::Stream {
294    platform_imp::into_legacy_ipc_stream(stream)
295}
296
297/// Temporary source-compatibility conversion for legacy 4.x callback inputs.
298#[cfg(feature = "ipc")]
299#[doc(hidden)]
300pub fn from_legacy_ipc_stream(stream: interprocess::local_socket::Stream) -> IpcStream {
301    platform_imp::from_legacy_ipc_stream(stream)
302}
303
304/// Temporary source-compatibility conversion for public APIs that return an
305/// `interprocess` endpoint name.
306#[cfg(feature = "ipc")]
307#[doc(hidden)]
308pub fn legacy_ipc_name(path: &str) -> Result<interprocess::local_socket::Name<'_>, String> {
309    platform_imp::legacy_ipc_name(path)
310}
311
312#[cfg(feature = "ipc-async")]
313pub use platform_imp::{
314    IpcAsyncListener, IpcAsyncStream, IpcIntoAsyncListener, IpcIntoAsyncStream,
315};
316
317#[cfg(feature = "pty")]
318pub use platform_imp::terminal::{
319    before_pty_spawn, current_backend_kind, find_child_processes, find_orphan_conhosts,
320    input_payload, is_ignorable_process_control_error, prepare_unmanaged_pty_child,
321    query_responses, resize_pty, shell_argv, signal_pty_tree, terminate_pty_child,
322    wait_before_pty_close_supported, Backend, ChildProcessInfo, ConPtyBackendKind,
323    OrphanConhostInfo, PtyProcessGuard, PtySpawnContext, TerminalInputSession,
324};
325
326#[cfg(feature = "session-relay")]
327pub use platform_imp::relay_local_socket_session;
328
329/// Apply host-owned setup for the legacy Tokio-command compatibility surface.
330///
331/// The public wrapper retains its policy type, while console suppression and
332/// owner-death primitives stay inside the selected platform root.
333#[cfg(feature = "async-process")]
334pub fn configure_compat_tokio_command(
335    command: &mut Command,
336    show_console: bool,
337    kill_when_owner_dies: bool,
338) -> io::Result<()> {
339    platform_imp::configure_compat_tokio_command(command, show_console, kill_when_owner_dies)
340}
341
342/// Complete host-owned setup after a legacy Tokio child has been spawned.
343#[cfg(feature = "async-process")]
344pub fn after_compat_tokio_spawn(child: &Child, kill_when_owner_dies: bool) -> io::Result<()> {
345    platform_imp::after_compat_tokio_spawn(child, kill_when_owner_dies)
346}
347
348/// Stdio policy for one child stream.
349#[cfg(feature = "async-process")]
350#[derive(Debug, Clone, Copy, PartialEq, Eq)]
351pub enum StreamMode {
352    /// Leave the stream connected to the parent process.
353    Inherit,
354    /// Create an asynchronous pipe owned by the child handle.
355    Piped,
356    /// Connect the stream to the platform null device.
357    Null,
358}
359
360#[cfg(feature = "async-process")]
361impl StreamMode {
362    fn apply(self) -> Stdio {
363        match self {
364            Self::Inherit => Stdio::inherit(),
365            Self::Piped => Stdio::piped(),
366            Self::Null => Stdio::null(),
367        }
368    }
369}
370
371/// Typed spawn description accepted by the blessed process boundary.
372#[cfg(feature = "async-process")]
373#[derive(Debug, Clone)]
374pub struct SpawnSpec {
375    program: OsString,
376    args: Vec<OsString>,
377    current_dir: Option<PathBuf>,
378    env: Vec<(OsString, OsString)>,
379    clear_env: bool,
380    stdin: StreamMode,
381    stdout: StreamMode,
382    stderr: StreamMode,
383    create_process_group: bool,
384    kill_when_owner_dies: bool,
385    nice: Option<i32>,
386}
387
388#[cfg(feature = "async-process")]
389impl SpawnSpec {
390    /// Create a direct (non-shell) command description.
391    pub fn new(program: impl Into<OsString>) -> Self {
392        Self {
393            program: program.into(),
394            args: Vec::new(),
395            current_dir: None,
396            env: Vec::new(),
397            clear_env: false,
398            stdin: StreamMode::Inherit,
399            stdout: StreamMode::Inherit,
400            stderr: StreamMode::Inherit,
401            create_process_group: false,
402            kill_when_owner_dies: false,
403            nice: None,
404        }
405    }
406
407    /// Append one argument without requiring UTF-8.
408    pub fn arg(mut self, arg: impl Into<OsString>) -> Self {
409        self.args.push(arg.into());
410        self
411    }
412
413    /// Set the child working directory.
414    pub fn current_dir(mut self, path: impl Into<PathBuf>) -> Self {
415        self.current_dir = Some(path.into());
416        self
417    }
418
419    /// Add an environment override.
420    pub fn env(mut self, key: impl Into<OsString>, value: impl Into<OsString>) -> Self {
421        self.env.push((key.into(), value.into()));
422        self
423    }
424
425    /// Start with an empty inherited environment before applying overrides.
426    pub fn clear_env(mut self, clear: bool) -> Self {
427        self.clear_env = clear;
428        self
429    }
430
431    /// Configure child stdin.
432    pub fn stdin(mut self, mode: StreamMode) -> Self {
433        self.stdin = mode;
434        self
435    }
436
437    /// Configure child stdout.
438    pub fn stdout(mut self, mode: StreamMode) -> Self {
439        self.stdout = mode;
440        self
441    }
442
443    /// Configure child stderr.
444    pub fn stderr(mut self, mode: StreamMode) -> Self {
445        self.stderr = mode;
446        self
447    }
448
449    /// Put the child in its own process group.
450    ///
451    /// This is what makes a group-wide soft signal addressable at all:
452    /// [`PlatformEmergencySignal::terminate_group_soft`] is a no-op without
453    /// it, because on POSIX the negative-PID signal would otherwise reach the
454    /// caller's own group, and on Windows `GenerateConsoleCtrlEvent` only
455    /// routes to children spawned with `CREATE_NEW_PROCESS_GROUP`. It also
456    /// detaches the child from the parent's console Ctrl+C, so it is opt-in.
457    pub fn create_process_group(mut self, create: bool) -> Self {
458        self.create_process_group = create;
459        self
460    }
461
462    /// Kill this child when the spawning process exits unexpectedly.
463    ///
464    /// Linux uses `PR_SET_PDEATHSIG(SIGTERM)` plus a pre-exec hard-exit race
465    /// guard when the parent changed before that signal could be armed.
466    /// Windows assigns the child to a
467    /// process-wide kill-on-close Job Object. macOS forks a kqueue supervisor
468    /// before exec and reports spawn success only after its owner and child
469    /// watches are registered.
470    pub fn kill_when_owner_dies(mut self, kill: bool) -> Self {
471        self.kill_when_owner_dies = kill;
472        self
473    }
474
475    /// Apply the host's existing niceness policy at child creation.
476    ///
477    /// On Unix this is the requested `setpriority(PRIO_PROCESS)` niceness.
478    /// Windows maps the established niceness bands to process creation
479    /// priority classes; it is deliberately a coarse host mapping rather
480    /// than a claim that numeric nice values are portable.
481    pub fn nice(mut self, nice: Option<i32>) -> Self {
482        self.nice = nice;
483        self
484    }
485
486    /// Spawn using the canonical asynchronous platform operation.
487    pub async fn spawn(self) -> io::Result<PlatformChild> {
488        let mut command = Command::new(&self.program);
489        command.args(&self.args);
490        if let Some(current_dir) = self.current_dir.as_deref() {
491            command.current_dir(current_dir);
492        }
493        if self.clear_env {
494            command.env_clear();
495        }
496        for (key, value) in &self.env {
497            command.env(key, value);
498        }
499        command
500            .stdin(self.stdin.apply())
501            .stdout(self.stdout.apply())
502            .stderr(self.stderr.apply());
503        platform_imp::configure_command(
504            &mut command,
505            self.create_process_group,
506            self.kill_when_owner_dies,
507            self.nice,
508        )?;
509
510        let child = command.spawn()?;
511        platform_imp::after_spawn(&child, self.kill_when_owner_dies)?;
512        Ok(PlatformChild::new(child, self.create_process_group))
513    }
514}
515
516/// Owned child handle returned by [`SpawnSpec::spawn`].
517#[cfg(feature = "async-process")]
518pub struct PlatformChild {
519    child: Child,
520    stdin: Option<ChildStdin>,
521    stdout: Option<ChildStdout>,
522    stderr: Option<ChildStderr>,
523    signal: PlatformEmergencySignal,
524}
525
526#[cfg(feature = "async-process")]
527impl PlatformChild {
528    fn new(mut child: Child, own_process_group: bool) -> Self {
529        let signal = PlatformEmergencySignal {
530            identity: async_child_identity(&child),
531            own_process_group,
532            // The legacy AsyncProcess actor historically retained only this
533            // numeric child-group leader on macOS. Keep it launch-bound for
534            // that API's compatibility path; sessions deliberately never use
535            // it because their control capability promises identity safety.
536            legacy_group_pid: child.id(),
537        };
538        Self {
539            stdin: child.stdin.take(),
540            stdout: child.stdout.take(),
541            stderr: child.stderr.take(),
542            child,
543            signal,
544        }
545    }
546
547    /// Return the operating-system process identifier, if available.
548    pub fn id(&self) -> Option<u32> {
549        self.child.id()
550    }
551
552    /// Wait for completion without capturing output.
553    pub async fn wait(&mut self) -> io::Result<ExitStatus> {
554        self.child.wait().await
555    }
556
557    /// Terminate the child and wait for its exit.
558    pub async fn kill(&mut self) -> io::Result<()> {
559        self.child.kill().await
560    }
561
562    /// Capture piped stdout and stderr while waiting for the child.
563    pub async fn wait_with_output(self) -> io::Result<Output> {
564        let Self {
565            mut child,
566            stdin,
567            stdout,
568            stderr,
569            ..
570        } = self;
571        // Match Tokio's `Child::wait_with_output` contract: one-shot output
572        // closes an owned stdin pipe so a child waiting for EOF can finish.
573        drop(stdin);
574        let (status, stdout, stderr) = tokio::try_join!(
575            child.wait(),
576            read_owned_to_end(stdout),
577            read_owned_to_end(stderr),
578        )?;
579        Ok(Output {
580            status,
581            stdout,
582            stderr,
583        })
584    }
585
586    /// Write bytes to piped stdin and flush them.
587    pub async fn write_stdin(&mut self, bytes: &[u8]) -> io::Result<()> {
588        let stdin = self.stdin.as_mut().ok_or_else(stdin_not_piped)?;
589        stdin.write_all(bytes).await?;
590        stdin.flush().await
591    }
592
593    /// Close the piped stdin handle, delivering EOF to the child.
594    ///
595    /// This operation is idempotent. Closing an inherited or null stdin is
596    /// also a no-op because there is no owned pipe to close.
597    pub fn close_stdin(&mut self) {
598        drop(self.stdin.take());
599    }
600
601    /// Read all bytes from piped stdout without waiting for process exit.
602    pub async fn read_stdout_to_end(&mut self) -> io::Result<Vec<u8>> {
603        let stdout = self.stdout.as_mut().ok_or_else(stdout_not_piped)?;
604        let mut bytes = Vec::new();
605        stdout.read_to_end(&mut bytes).await?;
606        Ok(bytes)
607    }
608
609    /// Read all bytes from piped stderr without waiting for process exit.
610    pub async fn read_stderr_to_end(&mut self) -> io::Result<Vec<u8>> {
611        let stderr = self.stderr.as_mut().ok_or_else(stderr_not_piped)?;
612        let mut bytes = Vec::new();
613        stderr.read_to_end(&mut bytes).await?;
614        Ok(bytes)
615    }
616
617    /// Split this child into sealed actor capabilities.
618    ///
619    /// The lifecycle wait handle, emergency termination handle, input pipe,
620    /// and output readers are deliberately separate so the actor can keep
621    /// accepting control commands while an asynchronous exit wait is pending.
622    pub fn into_actor_parts(
623        self,
624    ) -> (
625        PlatformLifecycle,
626        PlatformEmergencySignal,
627        Option<PlatformStdin>,
628        Option<PlatformOutput>,
629        Option<PlatformOutput>,
630    ) {
631        (
632            PlatformLifecycle { child: self.child },
633            self.signal,
634            self.stdin.map(|stdin| PlatformStdin { stdin }),
635            self.stdout.map(PlatformOutput::stdout),
636            self.stderr.map(PlatformOutput::stderr),
637        )
638    }
639}
640
641/// Opaque exit-wait capability owned by a process actor.
642#[cfg(feature = "async-process")]
643pub struct PlatformLifecycle {
644    child: Child,
645}
646
647#[cfg(feature = "async-process")]
648impl PlatformLifecycle {
649    /// Wait asynchronously for the child to exit.
650    pub async fn wait(&mut self) -> io::Result<ExitStatus> {
651        self.child.wait().await
652    }
653
654    /// Request direct-child termination through the still-owned child handle
655    /// without waiting for its reaping result.
656    ///
657    /// This is the identity-safe fallback when a host cannot provide a
658    /// separately usable launch-bound signal capability (for example a Linux
659    /// kernel without pidfds). The actor continues to own this lifecycle
660    /// handle and performs the eventual reap itself.
661    pub fn start_kill(&mut self) -> io::Result<()> {
662        self.child.start_kill()
663    }
664}
665
666/// Opaque, non-reap-capable emergency termination capability.
667///
668/// It can be used while the actor has a pending wait on
669/// [`PlatformLifecycle`], but it cannot observe or consume the exit result.
670#[cfg(feature = "async-process")]
671pub struct PlatformEmergencySignal {
672    identity: Option<AsyncChildIdentity>,
673    own_process_group: bool,
674    legacy_group_pid: Option<u32>,
675}
676
677#[cfg(feature = "async-process")]
678impl PlatformEmergencySignal {
679    /// Request immediate termination without waiting for process reaping.
680    pub fn kill(&self) -> io::Result<()> {
681        let Some(identity) = self.identity.as_ref() else {
682            return Err(signal_target_unavailable());
683        };
684        signal_async_child(identity)
685    }
686
687    /// Ask the child's whole process group to shut down gracefully.
688    ///
689    /// Returns `Ok(false)` when the child was not spawned with
690    /// [`SpawnSpec::create_process_group`]: there is no group to address, and
691    /// signalling anyway would hit the caller's own group on POSIX or the
692    /// caller's console on Windows. A missing or mismatched launch identity
693    /// instead reports an unavailable target; it never falls back to a
694    /// numeric group identifier that might have been reused.
695    pub fn terminate_group_soft(&self) -> io::Result<bool> {
696        if !self.own_process_group {
697            return Ok(false);
698        }
699        let Some(identity) = self.identity.as_ref() else {
700            return Err(signal_target_unavailable());
701        };
702        signal_async_child_group(identity).map(|()| true)
703    }
704
705    /// Legacy AsyncProcess-only graceful group termination.
706    ///
707    /// Most hosts retain an identity-safe asynchronous signal capability. On
708    /// macOS there is no pidfd-equivalent for the Tokio child path, while the
709    /// pre-session AsyncProcess contract historically sent SIGTERM to the
710    /// launch child's numeric process group. Preserve that established
711    /// best-effort behavior only for the legacy actor; sessions keep using
712    /// [`Self::terminate_group_soft`] and therefore remain identity-safe.
713    pub fn terminate_group_soft_legacy(&self) -> io::Result<bool> {
714        if !self.own_process_group {
715            return Ok(false);
716        }
717        if let Some(identity) = self.identity.as_ref() {
718            return signal_async_child_group(identity).map(|()| true);
719        }
720        let pid = self
721            .legacy_group_pid
722            .ok_or_else(signal_target_unavailable)?;
723        crate::platform::process::soft_terminate_process_group(pid).map(|()| true)
724    }
725
726    /// Return direct-child CPU time when this host can still verify the
727    /// launch identity. Unsupported hosts and already-reused identities are
728    /// reported as `None`, never as a PID-only best effort.
729    pub fn cpu_time(&self) -> io::Result<Option<std::time::Duration>> {
730        self.identity
731            .as_ref()
732            .map_or(Ok(None), async_child_cpu_time)
733    }
734}
735
736#[cfg(feature = "async-process")]
737fn signal_target_unavailable() -> io::Error {
738    io::Error::new(
739        io::ErrorKind::BrokenPipe,
740        "child process launch identity is no longer available",
741    )
742}
743
744/// Opaque piped stdin capability owned by a process actor.
745#[cfg(feature = "async-process")]
746pub struct PlatformStdin {
747    stdin: ChildStdin,
748}
749
750#[cfg(feature = "async-process")]
751impl PlatformStdin {
752    /// Write and flush bytes to the child stdin pipe.
753    pub async fn write(&mut self, bytes: &[u8]) -> io::Result<()> {
754        self.stdin.write_all(bytes).await?;
755        self.stdin.flush().await
756    }
757}
758
759/// Opaque stdout or stderr reader owned by a process actor.
760#[cfg(feature = "async-process")]
761pub struct PlatformOutput {
762    reader: OutputReader,
763    read_pending: bool,
764}
765
766#[cfg(feature = "async-process")]
767enum OutputReader {
768    Stdout(ChildStdout),
769    Stderr(ChildStderr),
770}
771
772#[cfg(feature = "async-process")]
773impl PlatformOutput {
774    fn stdout(stdout: ChildStdout) -> Self {
775        Self {
776            reader: OutputReader::Stdout(stdout),
777            read_pending: false,
778        }
779    }
780
781    fn stderr(stderr: ChildStderr) -> Self {
782        Self {
783            reader: OutputReader::Stderr(stderr),
784            read_pending: false,
785        }
786    }
787
788    /// Drain this output endpoint to EOF without blocking a runtime worker.
789    pub async fn read_to_end(self) -> io::Result<Vec<u8>> {
790        match self.reader {
791            OutputReader::Stdout(stdout) => read_owned_to_end(Some(stdout)).await,
792            OutputReader::Stderr(stderr) => read_owned_to_end(Some(stderr)).await,
793        }
794    }
795
796    /// Read the next asynchronous chunk from this output endpoint.
797    ///
798    /// The caller owns the buffer and therefore controls the amount of data
799    /// retained at each read. EOF is reported as `Ok(0)`.
800    pub async fn read_chunk(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
801        self.read_pending = true;
802        let result = match &mut self.reader {
803            OutputReader::Stdout(stdout) => stdout.read(buffer).await,
804            OutputReader::Stderr(stderr) => stderr.read(buffer).await,
805        };
806        self.read_pending = false;
807        result
808    }
809
810    /// Abandon output and await destruction of this reader's pending I/O.
811    ///
812    /// This is not direct-child reaping or lossless EOF. The owning actor must
813    /// retain this future until completion: dropping it is not an acknowledgement
814    /// that platform I/O storage has been released. Cancellation requests alone
815    /// do not count as completion, and an unresponsive driver may keep it pending.
816    pub async fn shutdown(self) -> io::Result<()> {
817        match self.reader {
818            OutputReader::Stdout(stdout) => {
819                platform_imp::shutdown_output_reader(stdout, self.read_pending).await
820            }
821            OutputReader::Stderr(stderr) => {
822                platform_imp::shutdown_output_reader(stderr, self.read_pending).await
823            }
824        }
825    }
826}
827
828#[cfg(all(test, feature = "async-process"))]
829mod output_shutdown_tests {
830    use super::*;
831    use std::time::Duration;
832
833    #[test]
834    fn silent_output_fixture() {
835        if std::env::var_os("RUNNING_PROCESS_OUTPUT_SHUTDOWN_FIXTURE").is_some() {
836            std::thread::sleep(Duration::from_secs(30));
837        }
838    }
839
840    #[tokio::test]
841    async fn shutdown_finishes_a_cancelled_pending_read_before_child_exit() {
842        let child = SpawnSpec::new(std::env::current_exe().expect("test executable"))
843            .arg("--exact")
844            .arg("output_shutdown_tests::silent_output_fixture")
845            .env("RUNNING_PROCESS_OUTPUT_SHUTDOWN_FIXTURE", "1")
846            .stdin(StreamMode::Null)
847            .stdout(StreamMode::Piped)
848            .stderr(StreamMode::Null)
849            .spawn()
850            .await
851            .expect("spawn silent fixture");
852        let (mut lifecycle, _, _, stdout, _) = child.into_actor_parts();
853        let mut stdout = stdout.expect("stdout pipe");
854        // Drain the libtest header, then establish a genuinely pending read.
855        let mut bytes = [0; 1024];
856        loop {
857            match tokio::time::timeout(Duration::from_millis(20), stdout.read_chunk(&mut bytes))
858                .await
859            {
860                Ok(Ok(0)) => panic!("fixture exited before pending read"),
861                Ok(Ok(_)) => {}
862                Ok(Err(error)) => panic!("fixture read failed: {error}"),
863                Err(_) => break,
864            }
865        }
866        let shutdown = tokio::time::timeout(Duration::from_secs(2), stdout.shutdown()).await;
867        // Reap even if the assertion fails; a passing shutdown cannot depend
868        // on this kill, which occurs only after the acknowledgement deadline.
869        lifecycle.start_kill().expect("kill fixture");
870        lifecycle.wait().await.expect("reap fixture");
871        shutdown
872            .expect("pending read shutdown must finish")
873            .expect("output shutdown");
874    }
875}
876
877#[cfg(feature = "async-process")]
878fn stdin_not_piped() -> io::Error {
879    io::Error::new(io::ErrorKind::BrokenPipe, "child stdin is not piped")
880}
881
882#[cfg(feature = "async-process")]
883fn stdout_not_piped() -> io::Error {
884    io::Error::new(io::ErrorKind::BrokenPipe, "child stdout is not piped")
885}
886
887#[cfg(feature = "async-process")]
888fn stderr_not_piped() -> io::Error {
889    io::Error::new(io::ErrorKind::BrokenPipe, "child stderr is not piped")
890}
891
892#[cfg(feature = "async-process")]
893async fn read_owned_to_end<R>(reader: Option<R>) -> io::Result<Vec<u8>>
894where
895    R: AsyncRead + Unpin,
896{
897    let Some(mut reader) = reader else {
898        return Ok(Vec::new());
899    };
900    let mut bytes = Vec::new();
901    reader.read_to_end(&mut bytes).await?;
902    Ok(bytes)
903}
904
905/// Build a shell command using the host platform's supported shell.
906#[cfg(feature = "async-process")]
907pub fn shell_spec(command: impl AsRef<OsStr>) -> SpawnSpec {
908    platform_imp::shell_spec(command.as_ref())
909}
910
911#[cfg(all(test, feature = "async-process"))]
912mod tests {
913    use super::{shell_spec, SpawnSpec, StreamMode};
914
915    fn fixture_command() -> SpawnSpec {
916        #[cfg(windows)]
917        {
918            shell_spec("echo async-platform-internal")
919        }
920        #[cfg(not(windows))]
921        {
922            shell_spec("printf async-platform-internal")
923        }
924    }
925
926    #[tokio::test]
927    async fn blessed_spawn_captures_output_without_sync_wait() {
928        let output = fixture_command()
929            .stdout(StreamMode::Piped)
930            .stderr(StreamMode::Piped)
931            .spawn()
932            .await
933            .expect("spawn")
934            .wait_with_output()
935            .await
936            .expect("wait with output");
937
938        assert!(output.status.success());
939        let expected = if cfg!(windows) {
940            b"async-platform-internal\r\n".as_slice()
941        } else {
942            b"async-platform-internal".as_slice()
943        };
944        assert_eq!(output.stdout, expected);
945        assert!(output.stderr.is_empty());
946    }
947
948    #[tokio::test]
949    async fn blessed_spawn_reports_missing_program() {
950        let result = SpawnSpec::new("running-process-program-that-does-not-exist")
951            .spawn()
952            .await;
953        assert!(result.is_err());
954    }
955
956    #[tokio::test]
957    async fn one_shot_output_closes_owned_stdin() {
958        #[cfg(windows)]
959        let spec = shell_spec("more > nul & echo done");
960        #[cfg(not(windows))]
961        let spec = shell_spec("cat > /dev/null; printf done");
962
963        let output = tokio::time::timeout(
964            std::time::Duration::from_secs(2),
965            spec.stdin(StreamMode::Piped)
966                .stdout(StreamMode::Piped)
967                .stderr(StreamMode::Piped)
968                .spawn()
969                .await
970                .expect("spawn")
971                .wait_with_output(),
972        )
973        .await
974        .expect("stdin is closed for one-shot output")
975        .expect("output succeeds");
976
977        let expected = if cfg!(windows) {
978            b"done\r\n".as_slice()
979        } else {
980            b"done".as_slice()
981        };
982        assert_eq!(output.stdout, expected);
983    }
984}