winx_code_agent/runtime/
mod.rs

1//! Shell execution boundary shared by embedded and daemon-backed runtimes.
2
3mod selection;
4mod session_store;
5
6use std::future::Future;
7use std::path::PathBuf;
8use std::pin::Pin;
9use std::sync::atomic::{AtomicBool, Ordering};
10use std::sync::Arc;
11use std::time::{Duration, Instant};
12
13use serde::{Deserialize, Serialize};
14use tokio::sync::Mutex;
15
16use crate::errors::{Result, WinxError};
17use crate::state::bash_state::BashState;
18use crate::state::pty::SharedPtyShell;
19use crate::tools::bash_command::BashCommandResult;
20use crate::types::BashCommand;
21
22pub use selection::{
23    configured_daemon_binary, ensure_control_daemon_at, ensure_daemon_at, restart_control_daemon_at,
24};
25pub use selection::{
26    configured_runtime_mode, configured_shell_runtime, select_runtime_mode, RuntimeMode,
27};
28pub(crate) use session_store::lock_session_store;
29pub use session_store::{SessionStore, ShellTarget};
30
31/// Boxed future returned by [`ShellRuntime`] implementations.
32pub type ShellRuntimeFuture<'a> =
33    Pin<Box<dyn Future<Output = Result<BashCommandResult>> + Send + 'a>>;
34pub type ShellRuntimeUnitFuture<'a> = Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>>;
35pub type ShellRuntimeBoolFuture<'a> = Pin<Box<dyn Future<Output = Result<bool>> + Send + 'a>>;
36pub type ShellRuntimeDetailedFuture<'a> =
37    Pin<Box<dyn Future<Output = Result<BashCommandRuntimeResult>> + Send + 'a>>;
38
39/// Internal orchestration details layered around the stable public
40/// `BashCommandResult`. Adding a separate type avoids adding required fields to
41/// the existing public result struct.
42#[derive(Clone, Debug, Eq, PartialEq)]
43pub struct BashCommandRuntimeResult {
44    pub result: BashCommandResult,
45    pub compact_output: Option<String>,
46    pub command_generation: Option<u64>,
47    pub execution_token: Option<ShellExecutionToken>,
48    pub generation_bound_actions: bool,
49    /// Runtime-owned file containing output dropped from the in-memory head.
50    /// The current result retains the tail; consumers should read this file only
51    /// when they need omitted history.
52    pub dropped_output_file: Option<PathBuf>,
53    /// Authoritative runtime truncation state. It is never inferred from
54    /// terminal text, which is controlled by the child process.
55    pub output_truncated: bool,
56}
57
58impl BashCommandRuntimeResult {
59    pub fn legacy(result: BashCommandResult) -> Self {
60        Self {
61            result,
62            compact_output: None,
63            command_generation: None,
64            execution_token: None,
65            generation_bound_actions: false,
66            dropped_output_file: None,
67            output_truncated: false,
68        }
69    }
70
71    pub(crate) fn with_dropped_output_file(mut self, path: Option<PathBuf>) -> Self {
72        self.dropped_output_file = path;
73        self
74    }
75}
76
77#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
78pub struct ShellExecutionToken {
79    pub guardian_epoch: String,
80    pub session_epoch: String,
81    pub generation: u64,
82}
83
84#[derive(Clone, Debug, Default, Serialize, Deserialize)]
85pub struct ShellActionOptions {
86    pub compact_output: bool,
87    pub expected_generation: Option<u64>,
88    /// Full execution identity used by protocol 1.5 peers. The legacy numeric
89    /// generation remains for source and protocol-1.4 compatibility only.
90    #[serde(default, skip_serializing_if = "Option::is_none")]
91    pub expected_execution: Option<ShellExecutionToken>,
92    /// Adapter-to-control precondition. A control process must compare this
93    /// with the effective guardian immediately before relaying an action.
94    #[serde(default, skip_serializing_if = "Option::is_none")]
95    pub expected_guardian_epoch: Option<String>,
96    /// Internal safety contract for an execution that is already represented
97    /// by an MCP Task and therefore cannot fall back to an unbound process.
98    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
99    pub require_generation_binding: bool,
100    /// Stable key for cancelling a launch that is queued inside a remote
101    /// guardian before an execution token exists.
102    #[serde(default, skip_serializing_if = "Option::is_none")]
103    pub cancellation_key: Option<String>,
104    /// In-process launch gate. Skipped on the wire; control/guardian safety is
105    /// represented by the serializable preconditions above.
106    #[serde(skip)]
107    pub launch_cancelled: Option<Arc<AtomicBool>>,
108    /// Deterministic coverage for the automatic PTY-reset path. Never exists
109    /// in production builds or on the daemon wire.
110    #[cfg(test)]
111    #[serde(skip)]
112    pub(crate) force_clear_to_run_failure: bool,
113}
114
115impl ShellActionOptions {
116    pub(crate) fn is_default(&self) -> bool {
117        let is_default = !self.compact_output
118            && self.expected_generation.is_none()
119            && self.expected_execution.is_none()
120            && self.expected_guardian_epoch.is_none()
121            && !self.require_generation_binding
122            && self.cancellation_key.is_none();
123        #[cfg(test)]
124        {
125            is_default && !self.force_clear_to_run_failure
126        }
127        #[cfg(not(test))]
128        {
129            is_default
130        }
131    }
132
133    pub(crate) fn is_launch_cancelled(&self) -> bool {
134        self.launch_cancelled.as_ref().is_some_and(|flag| flag.load(Ordering::SeqCst))
135    }
136}
137
138impl PartialEq for ShellActionOptions {
139    fn eq(&self, other: &Self) -> bool {
140        let equal = self.compact_output == other.compact_output
141            && self.expected_generation == other.expected_generation
142            && self.expected_execution == other.expected_execution
143            && self.expected_guardian_epoch == other.expected_guardian_epoch
144            && self.require_generation_binding == other.require_generation_binding
145            && self.cancellation_key == other.cancellation_key;
146        #[cfg(test)]
147        {
148            equal && self.force_clear_to_run_failure == other.force_clear_to_run_failure
149        }
150        #[cfg(not(test))]
151        {
152            equal
153        }
154    }
155}
156
157impl Eq for ShellActionOptions {}
158
159#[derive(Clone, Debug, Default, Eq, PartialEq)]
160pub struct ShellSessionConfiguration {
161    pub attach_hint: Option<String>,
162    pub attached_existing: bool,
163}
164
165pub type ShellRuntimeConfigureFuture<'a> =
166    Pin<Box<dyn Future<Output = Result<ShellSessionConfiguration>> + Send + 'a>>;
167
168/// Observational snapshot of one shell session, answered by
169/// `Initialize(type = "status")` without touching the PTY.
170#[derive(Clone, Debug, Default, Eq, PartialEq, serde::Serialize)]
171pub struct ShellSessionStatus {
172    /// `daemon` or `embedded`.
173    pub runtime: &'static str,
174    /// Guardian process behind this session (daemon runtime only).
175    #[serde(skip_serializing_if = "Option::is_none")]
176    pub guardian_pid: Option<u32>,
177    #[serde(skip_serializing_if = "Option::is_none")]
178    pub guardian_epoch: Option<String>,
179    #[serde(skip_serializing_if = "Option::is_none")]
180    pub guardian_build: Option<String>,
181    pub command_running: bool,
182    #[serde(skip_serializing_if = "Option::is_none")]
183    pub running_command_id: Option<String>,
184    pub background_command_ids: Vec<String>,
185    #[serde(skip_serializing_if = "Option::is_none")]
186    pub last_activity_unix_ms: Option<u64>,
187}
188
189pub type ShellRuntimeStatusFuture<'a> =
190    Pin<Box<dyn Future<Output = Result<ShellSessionStatus>> + Send + 'a>>;
191
192/// State transition requested by the Initialize tool.
193#[derive(Clone, Copy, Debug, Eq, PartialEq)]
194pub enum ShellSessionTransition {
195    FirstCall,
196    ModeChange,
197    Reset,
198    WorkspaceChange,
199}
200
201/// Narrow process-agnostic boundary for `BashCommand` execution.
202pub trait ShellRuntime: Send + Sync {
203    fn configure_session<'a>(
204        &'a self,
205        bash_state: &'a mut BashState,
206        transition: ShellSessionTransition,
207    ) -> ShellRuntimeConfigureFuture<'a>;
208
209    /// Reattach an adapter to its existing session. External runtimes retain
210    /// their historical mode-refresh behavior; native runtimes override this
211    /// with a non-mutating fast path.
212    fn attach_session<'a>(
213        &'a self,
214        bash_state: &'a mut BashState,
215    ) -> ShellRuntimeConfigureFuture<'a> {
216        self.configure_session(bash_state, ShellSessionTransition::ModeChange)
217    }
218
219    fn run_action<'a>(
220        &'a self,
221        bash_state: &'a Arc<Mutex<Option<BashState>>>,
222        command: BashCommand,
223    ) -> ShellRuntimeFuture<'a>;
224
225    /// Rich adapter path used by MCP orchestration. External runtime
226    /// implementations remain source-compatible through this default.
227    fn run_action_detailed<'a>(
228        &'a self,
229        bash_state: &'a Arc<Mutex<Option<BashState>>>,
230        command: BashCommand,
231        _options: ShellActionOptions,
232    ) -> ShellRuntimeDetailedFuture<'a> {
233        Box::pin(async move {
234            self.run_action(bash_state, command).await.map(BashCommandRuntimeResult::legacy)
235        })
236    }
237
238    fn supports_generation_bound_actions(&self) -> ShellRuntimeBoolFuture<'_> {
239        Box::pin(async { Ok(false) })
240    }
241
242    /// Session-aware capability probe. Daemon-backed runtimes override this so
243    /// the answer comes from the guardian that owns this shell, not merely from
244    /// the stable control process in front of it.
245    fn supports_generation_bound_actions_for<'a>(
246        &'a self,
247        _bash_state: &'a Arc<Mutex<Option<BashState>>>,
248    ) -> ShellRuntimeBoolFuture<'a> {
249        self.supports_generation_bound_actions()
250    }
251
252    /// Describe the session without mutating it. The default inspects the
253    /// in-process PTY; daemon-backed runtimes ask the guardian that owns it.
254    fn describe_session<'a>(
255        &'a self,
256        bash_state: &'a Arc<Mutex<Option<BashState>>>,
257    ) -> ShellRuntimeStatusFuture<'a> {
258        Box::pin(async move {
259            let guard = bash_state.lock().await;
260            let state = guard.as_ref().ok_or(WinxError::BashStateNotInitialized)?;
261            let (command_running, running_command_id) = {
262                let shell = state.pty_shell.lock().await;
263                shell.as_ref().map_or((false, None), |shell| {
264                    (
265                        shell.command_running,
266                        shell.command_running.then(|| shell.last_command.clone()),
267                    )
268                })
269            };
270            let background_command_ids =
271                lock_session_store().background_ids_for(&state.current_thread_id);
272            Ok(ShellSessionStatus {
273                runtime: "embedded",
274                command_running,
275                running_command_id,
276                background_command_ids,
277                ..ShellSessionStatus::default()
278            })
279        })
280    }
281
282    fn interrupt<'a>(
283        &'a self,
284        bash_state: &'a Arc<Mutex<Option<BashState>>>,
285    ) -> ShellRuntimeUnitFuture<'a>;
286
287    /// Interrupt only the exact command generation owned by a Task. The
288    /// conservative default refuses generation-bound interruption.
289    fn interrupt_generation<'a>(
290        &'a self,
291        bash_state: &'a Arc<Mutex<Option<BashState>>>,
292        expected_generation: Option<u64>,
293    ) -> ShellRuntimeBoolFuture<'a> {
294        Box::pin(async move {
295            if expected_generation.is_some() {
296                return Ok(false);
297            }
298            self.interrupt(bash_state).await?;
299            Ok(true)
300        })
301    }
302
303    /// Interrupt only the complete execution identity. External runtimes keep
304    /// working through the generation-only conservative default.
305    fn interrupt_execution<'a>(
306        &'a self,
307        bash_state: &'a Arc<Mutex<Option<BashState>>>,
308        expected: Option<ShellExecutionToken>,
309    ) -> ShellRuntimeBoolFuture<'a> {
310        self.interrupt_generation(bash_state, expected.map(|token| token.generation))
311    }
312
313    /// Cancel a command that is reserved remotely but has not reached the PTY.
314    /// Embedded runtimes use the in-process cancellation flag and need no
315    /// additional reservation protocol.
316    fn cancel_pending_action<'a>(
317        &'a self,
318        _bash_state: &'a Arc<Mutex<Option<BashState>>>,
319        _cancellation_key: &'a str,
320    ) -> ShellRuntimeBoolFuture<'a> {
321        Box::pin(async { Ok(false) })
322    }
323
324    /// Release the runtime-owned resources for one logical session.
325    fn terminate_session<'a>(&'a self, thread_id: &'a str) -> ShellRuntimeUnitFuture<'a>;
326}
327
328/// Current in-process runtime. It intentionally delegates to the existing
329/// implementation so introducing the boundary does not change behavior.
330#[derive(Clone, Copy, Debug, Default)]
331pub struct EmbeddedShellRuntime;
332
333impl ShellRuntime for EmbeddedShellRuntime {
334    fn configure_session<'a>(
335        &'a self,
336        bash_state: &'a mut BashState,
337        transition: ShellSessionTransition,
338    ) -> ShellRuntimeConfigureFuture<'a> {
339        Box::pin(async move {
340            let operation_barrier =
341                lock_session_store().operation_barrier(&bash_state.current_thread_id);
342            let _operation = operation_barrier.write().await;
343            if matches!(
344                transition,
345                ShellSessionTransition::FirstCall | ShellSessionTransition::Reset
346            ) && (transition == ShellSessionTransition::Reset || bash_state.cwd.exists())
347            {
348                bash_state.init_pty_shell().await?;
349            }
350            let attach_hint = {
351                let shell = bash_state.pty_shell.lock().await;
352                shell.as_ref().and_then(|shell| shell.attach_hint.clone())
353            };
354            Ok(ShellSessionConfiguration { attach_hint, attached_existing: false })
355        })
356    }
357
358    fn attach_session<'a>(
359        &'a self,
360        _bash_state: &'a mut BashState,
361    ) -> ShellRuntimeConfigureFuture<'a> {
362        Box::pin(async {
363            Ok(ShellSessionConfiguration { attach_hint: None, attached_existing: true })
364        })
365    }
366
367    fn run_action<'a>(
368        &'a self,
369        bash_state: &'a Arc<Mutex<Option<BashState>>>,
370        command: BashCommand,
371    ) -> ShellRuntimeFuture<'a> {
372        Box::pin(crate::tools::bash_command::handle_embedded_tool_call(bash_state, command))
373    }
374
375    fn run_action_detailed<'a>(
376        &'a self,
377        bash_state: &'a Arc<Mutex<Option<BashState>>>,
378        command: BashCommand,
379        options: ShellActionOptions,
380    ) -> ShellRuntimeDetailedFuture<'a> {
381        Box::pin(crate::tools::bash_command::handle_embedded_tool_call_detailed(
382            bash_state, command, options,
383        ))
384    }
385
386    fn supports_generation_bound_actions(&self) -> ShellRuntimeBoolFuture<'_> {
387        Box::pin(async { Ok(true) })
388    }
389
390    fn interrupt<'a>(
391        &'a self,
392        bash_state: &'a Arc<Mutex<Option<BashState>>>,
393    ) -> ShellRuntimeUnitFuture<'a> {
394        Box::pin(interrupt_embedded(bash_state))
395    }
396
397    fn interrupt_generation<'a>(
398        &'a self,
399        bash_state: &'a Arc<Mutex<Option<BashState>>>,
400        expected_generation: Option<u64>,
401    ) -> ShellRuntimeBoolFuture<'a> {
402        Box::pin(interrupt_embedded_generation(bash_state, expected_generation))
403    }
404
405    fn interrupt_execution<'a>(
406        &'a self,
407        bash_state: &'a Arc<Mutex<Option<BashState>>>,
408        expected: Option<ShellExecutionToken>,
409    ) -> ShellRuntimeBoolFuture<'a> {
410        Box::pin(interrupt_embedded_execution(bash_state, expected))
411    }
412
413    fn terminate_session<'a>(&'a self, thread_id: &'a str) -> ShellRuntimeUnitFuture<'a> {
414        Box::pin(async move {
415            let operation_barrier = lock_session_store().operation_barrier(thread_id);
416            let _operation = operation_barrier.write().await;
417            let shell = lock_session_store().resolve(thread_id, &ShellTarget::Main);
418            if let Some(shell) = shell {
419                *shell.lock().await = None;
420            }
421            Ok(())
422        })
423    }
424}
425
426async fn interrupt_embedded(bash_state: &Arc<Mutex<Option<BashState>>>) -> Result<()> {
427    let _ = interrupt_embedded_generation(bash_state, None).await?;
428    Ok(())
429}
430
431const INTERRUPT_RECOVERY_TIMEOUT: Duration = Duration::from_secs(3);
432const INTERRUPT_RETRY_INTERVAL: Duration = Duration::from_millis(250);
433const INTERRUPT_POLL_INTERVAL: Duration = Duration::from_millis(50);
434const MAX_INTERRUPT_ATTEMPTS: u32 = 3;
435
436/// Interrupt one exact foreground generation and wait until its prompt is
437/// visible. PTYs can occasionally deliver Ctrl+C while a short-lived child is
438/// handing terminal control back to its parent, so retry the signal while the
439/// same generation is still running. The caller retains the session operation
440/// barrier, which prevents any retry from reaching a newer command.
441pub(crate) async fn interrupt_pty_until_recovered(
442    shell: &SharedPtyShell,
443    expected_generation: Option<u64>,
444) -> Result<bool> {
445    let deadline = Instant::now() + INTERRUPT_RECOVERY_TIMEOUT;
446    let mut next_interrupt = Instant::now();
447    let mut attempts = 0_u32;
448
449    loop {
450        let now = Instant::now();
451        let recovered = {
452            let mut guard = shell.lock().await;
453            let Some(pty) = guard.as_mut() else { return Ok(false) };
454            if expected_generation.is_some_and(|expected| pty.command_generation() != expected) {
455                return Ok(false);
456            }
457
458            if pty.poll_output_nonblocking() || !pty.command_running {
459                if attempts == 0 {
460                    return Ok(false);
461                }
462                true
463            } else {
464                if attempts < MAX_INTERRUPT_ATTEMPTS && now >= next_interrupt {
465                    pty.send_interrupt().map_err(|error| {
466                        WinxError::CommandExecutionError(format!(
467                            "failed to interrupt shell: {error}"
468                        ))
469                    })?;
470                    attempts += 1;
471                    next_interrupt = Instant::now() + INTERRUPT_RETRY_INTERVAL;
472                    if attempts > 1 {
473                        tracing::debug!(attempts, "retrying Ctrl+C for active PTY generation");
474                    }
475                }
476                false
477            }
478        };
479
480        if recovered {
481            return Ok(true);
482        }
483        if Instant::now() >= deadline {
484            break;
485        }
486        tokio::time::sleep(INTERRUPT_POLL_INTERVAL).await;
487    }
488
489    Err(WinxError::CommandExecutionError(
490        "interrupted shell did not return to a prompt within 3 seconds".to_string(),
491    ))
492}
493
494async fn interrupt_embedded_generation(
495    bash_state: &Arc<Mutex<Option<BashState>>>,
496    expected_generation: Option<u64>,
497) -> Result<bool> {
498    interrupt_embedded_guarded(bash_state, expected_generation, None).await
499}
500
501async fn interrupt_embedded_execution(
502    bash_state: &Arc<Mutex<Option<BashState>>>,
503    expected: Option<ShellExecutionToken>,
504) -> Result<bool> {
505    interrupt_embedded_guarded(
506        bash_state,
507        expected.as_ref().map(|token| token.generation),
508        expected.as_ref(),
509    )
510    .await
511}
512
513async fn interrupt_embedded_guarded(
514    bash_state: &Arc<Mutex<Option<BashState>>>,
515    expected_generation: Option<u64>,
516    expected_execution: Option<&ShellExecutionToken>,
517) -> Result<bool> {
518    // Match configure_session's BashState -> operation barrier lock order and
519    // retain the read side from incarnation capture through interruption.
520    let (shell, _operation) = {
521        let state = bash_state.lock().await;
522        let Some(state) = state.as_ref() else { return Ok(false) };
523        let operation_barrier = lock_session_store().operation_barrier(&state.current_thread_id);
524        let operation = operation_barrier.read_owned().await;
525        (state.pty_shell.clone(), operation)
526    };
527    {
528        let guard = shell.lock().await;
529        if let Some(pty) = guard.as_ref() {
530            if expected_execution.is_some_and(|expected| {
531                expected.guardian_epoch != "embedded"
532                    || expected.session_epoch != format!("{:016x}", pty.incarnation())
533            }) {
534                return Ok(false);
535            }
536        }
537    }
538    interrupt_pty_until_recovered(&shell, expected_generation).await
539}
winx_code_agent/runtime/mod.rs

winx_code_agent/runtime/
mod.rs
