imagegen_bridge_codex_app_server/
process.rs

1//! Supervised `codex app-server` child process.
2
3use std::{
4    path::PathBuf,
5    process::Stdio,
6    sync::{
7        Arc,
8        atomic::{AtomicBool, AtomicU64, Ordering},
9    },
10    time::{Duration, Instant},
11};
12
13use imagegen_bridge_core::{BridgeError, ErrorCode};
14use tokio::{
15    process::{Child, Command},
16    sync::Mutex,
17};
18
19use crate::{AppServerRpc, RpcConfig};
20
21/// Child process configuration.
22#[derive(Debug, Clone)]
23pub struct CodexProcessConfig {
24    /// Codex executable path or command name.
25    pub executable: PathBuf,
26    /// Extra arguments after `app-server`.
27    pub args: Vec<String>,
28    /// Optional process working directory.
29    pub cwd: Option<PathBuf>,
30    /// JSONL connection limits.
31    pub rpc: RpcConfig,
32    /// Grace period before forceful termination.
33    pub shutdown_timeout: Duration,
34    /// Minimum interval between child starts to bound restart storms.
35    pub restart_backoff: Duration,
36}
37
38impl Default for CodexProcessConfig {
39    fn default() -> Self {
40        Self {
41            executable: PathBuf::from("codex"),
42            args: Vec::new(),
43            cwd: None,
44            rpc: RpcConfig::default(),
45            shutdown_timeout: Duration::from_secs(5),
46            restart_backoff: Duration::from_millis(250),
47        }
48    }
49}
50
51/// Initialized RPC client plus owned child lifecycle.
52#[derive(Debug)]
53pub struct CodexProcess {
54    config: CodexProcessConfig,
55    running: Mutex<Option<RunningProcess>>,
56    shutting_down: AtomicBool,
57    generation: AtomicU64,
58}
59
60#[derive(Debug)]
61struct RunningProcess {
62    rpc: Arc<AppServerRpc>,
63    child: Child,
64    started_at: Instant,
65}
66
67impl CodexProcess {
68    /// Spawns `codex app-server` using inherited OAuth/config state.
69    pub async fn spawn(config: CodexProcessConfig) -> Result<Self, BridgeError> {
70        if config.shutdown_timeout.is_zero() {
71            return Err(BridgeError::new(
72                ErrorCode::Configuration,
73                "codex app-server shutdown timeout must be greater than zero",
74            ));
75        }
76        if config.restart_backoff > Duration::from_secs(30) {
77            return Err(BridgeError::new(
78                ErrorCode::Configuration,
79                "codex app-server restart backoff must not exceed 30 seconds",
80            ));
81        }
82        let running = spawn_running(&config).await?;
83        Ok(Self {
84            config,
85            running: Mutex::new(Some(running)),
86            shutting_down: AtomicBool::new(false),
87            generation: AtomicU64::new(1),
88        })
89    }
90
91    /// Returns a healthy initialized connection, restarting a failed child once.
92    pub async fn rpc(&self) -> Result<Arc<AppServerRpc>, BridgeError> {
93        if self.shutting_down.load(Ordering::Acquire) {
94            return Err(BridgeError::new(
95                ErrorCode::Cancelled,
96                "codex app-server supervisor is shutting down",
97            ));
98        }
99        let mut running = self.running.lock().await;
100        let healthy = match running.as_mut() {
101            Some(process) if !process.rpc.is_closed() => match process.child.try_wait() {
102                Ok(None) => true,
103                Ok(Some(_)) | Err(_) => false,
104            },
105            Some(_) | None => false,
106        };
107        if healthy {
108            return Ok(Arc::clone(
109                &running.as_ref().ok_or_else(supervisor_state_error)?.rpc,
110            ));
111        }
112        let previous_started_at = running.as_ref().map(|process| process.started_at);
113        if let Some(process) = running.take() {
114            stop_running(process, self.config.shutdown_timeout).await?;
115        }
116        if self.shutting_down.load(Ordering::Acquire) {
117            return Err(BridgeError::new(
118                ErrorCode::Cancelled,
119                "codex app-server supervisor is shutting down",
120            ));
121        }
122        if let Some(started_at) = previous_started_at {
123            tokio::time::sleep(
124                self.config
125                    .restart_backoff
126                    .saturating_sub(started_at.elapsed()),
127            )
128            .await;
129        }
130        if self.shutting_down.load(Ordering::Acquire) {
131            return Err(BridgeError::new(
132                ErrorCode::Cancelled,
133                "codex app-server supervisor is shutting down",
134            ));
135        }
136        let replacement = spawn_running(&self.config).await?;
137        let rpc = Arc::clone(&replacement.rpc);
138        *running = Some(replacement);
139        self.generation.fetch_add(1, Ordering::AcqRel);
140        Ok(rpc)
141    }
142
143    /// Current one-based child generation, useful for safe operational metrics.
144    #[must_use]
145    pub fn generation(&self) -> u64 {
146        self.generation.load(Ordering::Acquire)
147    }
148
149    /// Terminates and reaps the owned child process. Repeated calls are safe.
150    pub async fn shutdown(&self) -> Result<(), BridgeError> {
151        self.shutting_down.store(true, Ordering::Release);
152        let Some(process) = self.running.lock().await.take() else {
153            return Ok(());
154        };
155        stop_running(process, self.config.shutdown_timeout).await
156    }
157}
158
159async fn spawn_running(config: &CodexProcessConfig) -> Result<RunningProcess, BridgeError> {
160    let mut command = Command::new(&config.executable);
161    command
162        .arg("app-server")
163        .args(&config.args)
164        .stdin(Stdio::piped())
165        .stdout(Stdio::piped())
166        .stderr(Stdio::null())
167        .kill_on_drop(true);
168    if let Some(cwd) = &config.cwd {
169        command.current_dir(cwd);
170    }
171    let mut child = command.spawn().map_err(|_| {
172        BridgeError::new(ErrorCode::Configuration, "could not spawn codex app-server")
173    })?;
174    let stdin = child.stdin.take().ok_or_else(|| {
175        BridgeError::new(ErrorCode::Internal, "codex app-server stdin is unavailable")
176    })?;
177    let stdout = child.stdout.take().ok_or_else(|| {
178        BridgeError::new(
179            ErrorCode::Internal,
180            "codex app-server stdout is unavailable",
181        )
182    })?;
183    let rpc = match AppServerRpc::connect(stdout, stdin, config.rpc).await {
184        Ok(rpc) => rpc,
185        Err(error) => {
186            let _ = child.kill().await;
187            return Err(error);
188        }
189    };
190    Ok(RunningProcess {
191        rpc,
192        child,
193        started_at: Instant::now(),
194    })
195}
196
197async fn stop_running(
198    mut process: RunningProcess,
199    shutdown_timeout: Duration,
200) -> Result<(), BridgeError> {
201    match process.child.try_wait() {
202        Ok(Some(_)) => return Ok(()),
203        Ok(None) => {}
204        Err(_) => {
205            return Err(BridgeError::new(
206                ErrorCode::Internal,
207                "could not inspect codex app-server process",
208            ));
209        }
210    }
211    process
212        .child
213        .start_kill()
214        .map_err(|_| BridgeError::new(ErrorCode::Internal, "could not stop codex app-server"))?;
215    tokio::time::timeout(shutdown_timeout, process.child.wait())
216        .await
217        .map_err(|_| BridgeError::new(ErrorCode::Timeout, "codex app-server did not stop"))?
218        .map_err(|_| BridgeError::new(ErrorCode::Internal, "could not reap codex app-server"))?;
219    Ok(())
220}
221
222fn supervisor_state_error() -> BridgeError {
223    BridgeError::new(
224        ErrorCode::Internal,
225        "codex app-server supervisor has no active process",
226    )
227}
imagegen_bridge_codex_app_server/process.rs

imagegen_bridge_codex_app_server/
process.rs
