1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
//! Process manager for iFlow CLI
//!
//! This module handles the lifecycle of the iFlow CLI process,
//! including starting, stopping, and managing stdio communication.
use crate::error::{IFlowError, Result};
use std::process::Stdio;
use std::time::Duration;
use tokio::process::Child;
use tokio::time::sleep;
/// Manages iFlow CLI process lifecycle
///
/// Handles starting and stopping the iFlow CLI process, as well as
/// providing access to its stdio streams for communication.
pub struct IFlowProcessManager {
pub process: Option<Child>, // Made public for access in Drop
start_port: u16,
port: Option<u16>,
debug: bool,
}
impl IFlowProcessManager {
/// Create a new process manager
///
/// # Arguments
/// * `start_port` - The port to start the process on
/// * `debug` - Whether to enable debug mode
///
/// # Returns
/// A new IFlowProcessManager instance
pub fn new(start_port: u16, debug: bool) -> Self {
Self {
process: None,
start_port,
port: None,
debug,
}
}
/// Check if a port is available for use
///
/// # Arguments
/// * `port` - Port number to check
///
/// # Returns
/// True if the port is available, False otherwise
fn is_port_available(port: u16) -> bool {
use std::net::TcpListener;
TcpListener::bind(("localhost", port)).is_ok()
}
/// Check if a port is listening (has a server running)
///
/// # Arguments
/// * `port` - Port number to check
///
/// # Returns
/// True if the port is listening, False otherwise
pub fn is_port_listening(port: u16) -> bool {
use std::net::TcpStream;
use std::time::Duration;
TcpStream::connect_timeout(
&format!("127.0.0.1:{}", port).parse().unwrap(),
Duration::from_millis(100),
)
.is_ok()
}
/// Find an available port starting from the given port
///
/// # Arguments
/// * `start_port` - Port to start searching from
/// * `max_attempts` - Maximum number of ports to try
///
/// # Returns
/// An available port number
///
/// # Errors
/// Returns an error if no available port is found
fn find_available_port(start_port: u16, max_attempts: u16) -> Result<u16> {
for i in 0..max_attempts {
let port = start_port + i;
if Self::is_port_available(port) {
tracing::debug!("Found available port: {}", port);
return Ok(port);
}
}
Err(IFlowError::ProcessManager(format!(
"No available port found in range {}-{}",
start_port,
start_port + max_attempts
)))
}
/// Start the iFlow process
///
/// Starts the iFlow CLI process with ACP support and WebSocket communication.
///
/// # Returns
/// * `Ok(String)` containing the WebSocket URL if the process was started successfully
/// * `Err(IFlowError)` if there was an error starting the process
pub async fn start(&mut self, use_websocket: bool) -> Result<Option<String>> {
if use_websocket {
tracing::debug!("Starting iFlow process with experimental ACP and WebSocket support");
// Find an available port
let port = Self::find_available_port(self.start_port, 100)?;
self.port = Some(port);
// Start iFlow process with WebSocket support
let mut cmd = tokio::process::Command::new("iflow");
cmd.arg("--experimental-acp");
cmd.arg("--port");
cmd.arg(port.to_string());
// Add debug flag if enabled
if self.debug {
cmd.arg("--debug");
}
// In WebSocket mode, set stdout/stderr to inherit to avoid blocking/exit when pipes are not consumed
cmd.stdout(Stdio::inherit());
cmd.stderr(Stdio::inherit());
cmd.stdin(Stdio::null()); // No stdin needed for WebSocket
let child = cmd
.spawn()
.map_err(|e| IFlowError::ProcessManager(format!("Failed to start iflow: {}", e)))?;
self.process = Some(child);
// Wait longer for process to start and WebSocket server to be ready
tracing::debug!("Waiting for iFlow process to start...");
sleep(Duration::from_secs(8)).await;
// Verify the port is actually listening with more retries and longer timeout
let mut attempts = 0;
let max_attempts = 30; // 30 attempts * 1 second = 30 seconds total
while attempts < max_attempts {
if Self::is_port_listening(port) {
tracing::debug!("iFlow WebSocket server is ready on port {}", port);
break;
}
attempts += 1;
if attempts % 5 == 0 {
tracing::debug!(
"Still waiting for iFlow to be ready... (attempt {}/{})",
attempts,
max_attempts
);
}
sleep(Duration::from_secs(1)).await;
}
if attempts >= max_attempts {
return Err(IFlowError::ProcessManager(format!(
"iFlow process failed to start WebSocket server on port {} after {} seconds",
port, max_attempts
)));
}
tracing::debug!(
"iFlow process started with WebSocket support on port {}",
port
);
// Return the WebSocket URL with peer parameter
Ok(Some(format!("ws://localhost:{}/acp?peer=iflow", port)))
} else {
tracing::debug!("Starting iFlow process with experimental ACP and stdio support");
// Start iFlow process with stdio support
let mut cmd = tokio::process::Command::new("iflow");
cmd.arg("--experimental-acp");
// Add debug flag if enabled
if self.debug {
cmd.arg("--debug");
}
cmd.stdout(Stdio::piped());
cmd.stderr(Stdio::piped());
cmd.stdin(Stdio::piped()); // stdin needed for stdio
tracing::debug!("Starting iFlow process with command: {:?}", cmd);
let child = cmd
.spawn()
.map_err(|e| IFlowError::ProcessManager(format!("Failed to start iflow: {}", e)))?;
self.process = Some(child);
// Wait for process to start
tracing::debug!("Waiting for iFlow process to start...");
sleep(Duration::from_secs(5)).await;
tracing::debug!("iFlow process should be started by now");
tracing::debug!("iFlow process started with stdio support");
// No WebSocket URL for stdio
Ok(None)
}
}
/// Stop the iFlow process
///
/// Attempts to gracefully stop the iFlow process if it's running.
///
/// # Returns
/// * `Ok(())` if the process was stopped successfully or wasn't running
/// * `Err(IFlowError)` if there was an error stopping the process
pub async fn stop(&mut self) -> Result<()> {
if let Some(mut process) = self.process.take() {
tracing::debug!("Stopping iFlow process");
// Try graceful shutdown first
match tokio::time::timeout(Duration::from_secs(5), process.kill()).await {
Ok(Ok(_)) => {
// Wait for the process to actually exit with a timeout
match tokio::time::timeout(Duration::from_secs(5), process.wait()).await {
Ok(Ok(_)) => tracing::debug!("iFlow process stopped gracefully"),
Ok(Err(e)) => tracing::warn!("Error waiting for iFlow process: {}", e),
Err(_) => {
tracing::warn!(
"Timeout waiting for iFlow process to exit, forcing termination"
);
// Force kill if it didn't exit in time
let _ = process.start_kill();
}
}
}
Ok(Err(e)) => {
tracing::warn!("Failed to kill iFlow process: {}, forcing termination", e);
let _ = process.start_kill();
}
Err(_) => {
tracing::warn!("Timeout killing iFlow process, forcing termination");
let _ = process.start_kill();
}
}
// Add a small delay to ensure all resources are released
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
tracing::debug!("iFlow process stopped");
}
// Clear the port when stopping
self.port = None;
Ok(())
}
/// Get the port the iFlow process is running on
///
/// # Returns
/// The port number, or None if not running
pub fn port(&self) -> Option<u16> {
self.port
}
/// Check if the iFlow process is running
///
/// # Returns
/// `true` if the process is running, `false` otherwise
pub fn is_running(&self) -> bool {
self.process.is_some()
}
/// Take ownership of the process's stdin
///
/// Takes ownership of the process's stdin stream for communication.
/// This method can only be called once, as it consumes the stream.
///
/// # Returns
/// `Some(ChildStdin)` if the process is running and has a stdin stream, `None` otherwise
pub fn take_stdin(&mut self) -> Option<tokio::process::ChildStdin> {
self.process.as_mut().and_then(|p| p.stdin.take())
}
/// Take ownership of the process's stdout
///
/// Takes ownership of the process's stdout stream for communication.
/// This method can only be called once, as it consumes the stream.
///
/// # Returns
/// `Some(ChildStdout)` if the process is running and has a stdout stream, `None` otherwise
pub fn take_stdout(&mut self) -> Option<tokio::process::ChildStdout> {
self.process.as_mut().and_then(|p| p.stdout.take())
}
}