winx_code_agent/runtime/
selection.rs

1use std::sync::Arc;
2
3use std::path::{Path, PathBuf};
4use std::process::{Command, Stdio};
5use std::time::{Duration, Instant};
6
7use crate::daemon::{
8    default_socket_path, DaemonClient, DaemonProcessRole, DaemonShellRuntime, HelloResult,
9    BUILD_IDENTITY_CAPABILITY, TYPED_ACTION_RESULT_CAPABILITY,
10};
11use crate::errors::{Result, WinxError};
12
13use super::{EmbeddedShellRuntime, ShellRuntime};
14
15const GUARDIAN_LIFECYCLE_CAPABILITY: &str = "guardian_activity_clock";
16const PLANNED_CONTROL_RESTART_CAPABILITY: &str = "planned_control_restart";
17const SESSION_NEGOTIATE_CAPABILITY: &str = "session.negotiate";
18const DAEMON_TRANSITION_TIMEOUT: Duration = Duration::from_secs(3);
19
20#[derive(Clone, Copy, Debug, Eq, PartialEq)]
21pub enum RuntimeMode {
22    Daemon,
23    Embedded,
24}
25
26fn enabled(value: Option<&str>) -> bool {
27    value.and_then(crate::config::parse_bool).unwrap_or(false)
28}
29
30/// Pure runtime-mode selection, kept separate from process spawning so its
31/// default and kill switches are deterministic under test.
32pub fn select_runtime_mode(
33    embedded: Option<&str>,
34    runtime: Option<&str>,
35    sandbox: Option<&str>,
36) -> Result<RuntimeMode> {
37    select_runtime_mode_for_platform(cfg!(any(unix, windows)), embedded, runtime, sandbox)
38}
39
40fn select_runtime_mode_for_platform(
41    daemon_supported: bool,
42    embedded: Option<&str>,
43    runtime: Option<&str>,
44    sandbox: Option<&str>,
45) -> Result<RuntimeMode> {
46    if enabled(sandbox) || enabled(embedded) {
47        return Ok(RuntimeMode::Embedded);
48    }
49    match runtime {
50        None | Some("" | "daemon") if daemon_supported => Ok(RuntimeMode::Daemon),
51        None | Some("" | "embedded") => Ok(RuntimeMode::Embedded),
52        Some("daemon") => Err(WinxError::ConfigurationError(
53            "WINX_RUNTIME=\"daemon\" requires Unix or Windows; use `embedded` on this OS"
54                .to_string(),
55        )),
56        Some(other) => Err(WinxError::ConfigurationError(format!(
57            "invalid WINX_RUNTIME={other:?}; expected `daemon` or `embedded`"
58        ))),
59    }
60}
61
62pub fn configured_runtime_mode() -> Result<RuntimeMode> {
63    let embedded = std::env::var("WINX_EMBEDDED").ok();
64    let runtime = std::env::var("WINX_RUNTIME").ok();
65    let sandbox = std::env::var("WINX_SANDBOX").ok();
66    select_runtime_mode(embedded.as_deref(), runtime.as_deref(), sandbox.as_deref())
67}
68
69pub async fn configured_shell_runtime() -> Result<Arc<dyn ShellRuntime>> {
70    match configured_runtime_mode()? {
71        RuntimeMode::Embedded => Ok(Arc::new(EmbeddedShellRuntime)),
72        RuntimeMode::Daemon => {
73            let socket = default_socket_path();
74            let binary = configured_daemon_binary()?;
75            ensure_control_daemon_at(&socket, &binary).await?;
76            Ok(Arc::new(DaemonShellRuntime::new(socket)))
77        }
78    }
79}
80
81pub fn configured_daemon_binary() -> Result<PathBuf> {
82    if let Some(path) = std::env::var_os("WINXD_BIN") {
83        return Ok(PathBuf::from(path));
84    }
85    let executable = std::env::current_exe().map_err(|error| {
86        WinxError::ConfigurationError(format!("cannot locate current Winx executable: {error}"))
87    })?;
88    let sibling = executable.with_file_name(if cfg!(windows) { "winxd.exe" } else { "winxd" });
89    if sibling.is_file() {
90        Ok(sibling)
91    } else {
92        Err(WinxError::ConfigurationError(format!(
93            "winxd not found beside {} (set WINXD_BIN or WINX_EMBEDDED=1)",
94            executable.display()
95        )))
96    }
97}
98
99fn has_capability(hello: &HelloResult, name: &str) -> bool {
100    hello.capabilities.iter().any(|capability| capability == name)
101}
102
103fn control_is_current(hello: &HelloResult) -> bool {
104    hello.process_role == Some(DaemonProcessRole::Control)
105        && [
106            GUARDIAN_LIFECYCLE_CAPABILITY,
107            TYPED_ACTION_RESULT_CAPABILITY,
108            SESSION_NEGOTIATE_CAPABILITY,
109            BUILD_IDENTITY_CAPABILITY,
110        ]
111        .into_iter()
112        .all(|capability| has_capability(hello, capability))
113        && hello.build_matches(&crate::build_info::BuildIdentity::current())
114}
115
116/// Ensure the current control-plane feature set is reachable. An older `winxd`
117/// that advertises safe planned restarts is replaced in place; per-session
118/// guardians keep owning their PTYs throughout the control-plane transition.
119pub async fn ensure_control_daemon_at(socket: &Path, daemon_binary: &Path) -> Result<()> {
120    ensure_daemon_at(socket, daemon_binary).await?;
121    let client = DaemonClient::new(socket);
122    let hello = client.hello().await?;
123    if control_is_current(&hello) {
124        return Ok(());
125    }
126    if !has_capability(&hello, PLANNED_CONTROL_RESTART_CAPABILITY) {
127        return Err(WinxError::ConfigurationError(format!(
128            "the running winxd (pid {}) lacks the current guardian lifecycle controls and cannot be safely \
129             restarted automatically; stop it manually, then retry",
130            hello.daemon_pid
131        )));
132    }
133
134    tracing::info!(
135        pid = hello.daemon_pid,
136        "restarting older winxd control plane while preserving session guardians"
137    );
138    let restarted = restart_control_daemon_from_hello(socket, daemon_binary, hello).await?;
139    for capability in [
140        GUARDIAN_LIFECYCLE_CAPABILITY,
141        TYPED_ACTION_RESULT_CAPABILITY,
142        SESSION_NEGOTIATE_CAPABILITY,
143        BUILD_IDENTITY_CAPABILITY,
144    ] {
145        if !has_capability(&restarted, capability) {
146            return Err(WinxError::ConfigurationError(format!(
147                "restarted winxd at {} still lacks required capability {capability}",
148                socket.display()
149            )));
150        }
151    }
152    let expected = crate::build_info::BuildIdentity::current();
153    if restarted.process_role != Some(DaemonProcessRole::Control)
154        || !restarted.build_matches(&expected)
155    {
156        return Err(WinxError::ConfigurationError(format!(
157            "restarted daemon at {} is not the expected winxd build {}",
158            socket.display(),
159            expected.display
160        )));
161    }
162    Ok(())
163}
164
165/// Safely restart only `winxd`; guardian processes and their PTYs remain alive.
166pub async fn restart_control_daemon_at(socket: &Path, daemon_binary: &Path) -> Result<HelloResult> {
167    let hello = DaemonClient::new(socket).hello().await?;
168    restart_control_daemon_from_hello(socket, daemon_binary, hello).await
169}
170
171async fn restart_control_daemon_from_hello(
172    socket: &Path,
173    daemon_binary: &Path,
174    previous: HelloResult,
175) -> Result<HelloResult> {
176    if !has_capability(&previous, PLANNED_CONTROL_RESTART_CAPABILITY) {
177        return Err(WinxError::ConfigurationError(format!(
178            "winxd pid {} does not advertise safe planned control restarts",
179            previous.daemon_pid
180        )));
181    }
182    signal_control_shutdown(socket, previous.daemon_pid).await?;
183
184    let client = DaemonClient::new(socket);
185    let deadline = Instant::now() + DAEMON_TRANSITION_TIMEOUT;
186    while Instant::now() < deadline {
187        match client.hello().await {
188            Ok(current) if current.daemon_epoch != previous.daemon_epoch => return Ok(current),
189            Ok(_) => tokio::time::sleep(Duration::from_millis(25)).await,
190            Err(_) => break,
191        }
192    }
193    if let Ok(current) = client.hello().await {
194        if current.daemon_epoch == previous.daemon_epoch {
195            return Err(WinxError::ShellInitializationError(format!(
196                "winxd pid {} did not stop within {} seconds",
197                previous.daemon_pid,
198                DAEMON_TRANSITION_TIMEOUT.as_secs()
199            )));
200        }
201        return Ok(current);
202    }
203
204    ensure_daemon_at(socket, daemon_binary).await?;
205    let restarted = client.hello().await?;
206    if restarted.daemon_epoch == previous.daemon_epoch {
207        return Err(WinxError::ShellInitializationError(
208            "winxd restart returned the previous daemon epoch".to_string(),
209        ));
210    }
211    Ok(restarted)
212}
213
214/// Planned control restart: SIGTERM on Unix, the `winx.shutdown` RPC on
215/// Windows. Either way guardians are separate detached processes and keep
216/// their PTYs.
217#[cfg_attr(unix, allow(clippy::unused_async))] // only the Windows branch awaits
218async fn signal_control_shutdown(socket: &Path, pid: u32) -> Result<()> {
219    #[cfg(unix)]
220    {
221        let _ = socket;
222        crate::os::unix::signal_process(pid, libc::SIGTERM).map_err(Into::into)
223    }
224    #[cfg(not(unix))]
225    {
226        let _ = pid;
227        DaemonClient::new(socket).shutdown().await
228    }
229}
230
231/// Start a daemon process that outlives this one: a new session on Unix, a
232/// detached process outside our console and job on Windows. Windows job
233/// objects may forbid breakaway (MCP clients often wrap servers in one), so a
234/// refused breakaway retries without it and the daemon then lives as long as
235/// that job does.
236fn spawn_detached(daemon_binary: &Path, socket: &Path) -> Result<std::process::Child> {
237    let build = |breakaway: bool| {
238        let mut command = Command::new(daemon_binary);
239        command
240            .arg("--socket")
241            .arg(socket)
242            .stdin(Stdio::null())
243            .stdout(Stdio::null())
244            .stderr(Stdio::null());
245        #[cfg(unix)]
246        {
247            let _ = breakaway;
248            crate::os::unix::configure_detached(&mut command);
249        }
250        #[cfg(windows)]
251        crate::os::windows::configure_detached(&mut command, breakaway);
252        command
253    };
254    let failure = |error: std::io::Error| {
255        WinxError::ShellInitializationError(format!(
256            "failed to auto-start {}: {error}",
257            daemon_binary.display()
258        ))
259    };
260    match build(true).spawn() {
261        Ok(child) => Ok(child),
262        Err(error) if cfg!(windows) => {
263            tracing::debug!(%error, "detached spawn with job breakaway refused; retrying inside the job");
264            build(false).spawn().map_err(failure)
265        }
266        Err(error) => Err(failure(error)),
267    }
268}
269
270/// Ensure a compatible daemon is reachable, starting `daemon_binary` only when
271/// the socket cannot be reached. A reachable incompatible daemon is never
272/// replaced or killed.
273pub async fn ensure_daemon_at(socket: &Path, daemon_binary: &Path) -> Result<()> {
274    let client = DaemonClient::new(socket);
275    match client.hello().await {
276        Ok(_) => return Ok(()),
277        Err(error @ WinxError::ConfigurationError(_)) => return Err(error),
278        Err(_) => {}
279    }
280
281    let mut child = spawn_detached(daemon_binary, socket)?;
282
283    // Several adapters starting at once each spawn a daemon; the kernel lets
284    // exactly one claim the socket and the others exit at once. An early
285    // child exit is therefore not a failure yet: keep asking the socket until
286    // the deadline, because the winner may belong to a sibling adapter.
287    let deadline = Instant::now() + DAEMON_TRANSITION_TIMEOUT;
288    let mut child_exit = None;
289    while Instant::now() < deadline {
290        match client.hello().await {
291            Ok(_) => {
292                if child_exit.is_none() {
293                    std::thread::spawn(move || {
294                        let _ = child.wait();
295                    });
296                }
297                return Ok(());
298            }
299            Err(error @ WinxError::ConfigurationError(_)) => return Err(error),
300            Err(_) => {}
301        }
302        if child_exit.is_none() {
303            child_exit = child.try_wait()?;
304        }
305        tokio::time::sleep(Duration::from_millis(25)).await;
306    }
307    Err(WinxError::ShellInitializationError(match child_exit {
308        Some(status) => format!(
309            "winxd exited ({status}) and nothing answered at {} within {} seconds",
310            socket.display(),
311            DAEMON_TRANSITION_TIMEOUT.as_secs()
312        ),
313        None => format!("timed out waiting for winxd at {}", socket.display()),
314    }))
315}
316
317#[cfg(test)]
318mod tests {
319    use super::{select_runtime_mode_for_platform, RuntimeMode};
320
321    #[test]
322    fn platform_without_unix_sockets_defaults_to_embedded() {
323        assert!(matches!(
324            select_runtime_mode_for_platform(false, None, None, None),
325            Ok(RuntimeMode::Embedded)
326        ));
327        assert!(select_runtime_mode_for_platform(false, None, Some("daemon"), None).is_err());
328    }
329
330    #[test]
331    fn protocol_1_4_control_without_session_negotiate_requires_upgrade() {
332        let hello = crate::daemon::HelloResult {
333            protocol_major: 1,
334            protocol_minor: 4,
335            capabilities: vec![
336                super::GUARDIAN_LIFECYCLE_CAPABILITY.to_string(),
337                super::TYPED_ACTION_RESULT_CAPABILITY.to_string(),
338                super::PLANNED_CONTROL_RESTART_CAPABILITY.to_string(),
339            ],
340            max_frame_bytes: 1024,
341            daemon_epoch: "control-14".to_string(),
342            daemon_pid: 14,
343            process_role: None,
344            build: None,
345        };
346        assert!(!super::control_is_current(&hello));
347        assert!(super::has_capability(&hello, super::PLANNED_CONTROL_RESTART_CAPABILITY));
348    }
349
350    #[test]
351    fn current_control_requires_the_exact_role_and_build() {
352        let capabilities = vec![
353            super::GUARDIAN_LIFECYCLE_CAPABILITY.to_string(),
354            super::TYPED_ACTION_RESULT_CAPABILITY.to_string(),
355            super::SESSION_NEGOTIATE_CAPABILITY.to_string(),
356            super::BUILD_IDENTITY_CAPABILITY.to_string(),
357        ];
358        let hello = crate::daemon::HelloResult::current(
359            crate::daemon::DaemonProcessRole::Control,
360            capabilities,
361            "current",
362        );
363        assert!(super::control_is_current(&hello));
364
365        let mut wrong_role = hello.clone();
366        wrong_role.process_role = Some(crate::daemon::DaemonProcessRole::Guardian);
367        assert!(!super::control_is_current(&wrong_role));
368
369        let mut wrong_build = hello;
370        if let Some(build) = wrong_build.build.as_mut() {
371            build.display.push_str("-different");
372        }
373        assert!(!super::control_is_current(&wrong_build));
374    }
375}
winx_code_agent/runtime/selection.rs

winx_code_agent/runtime/
selection.rs
