app-server: support daemon-safe restart handling (#21831)

## Why The app-server daemon work needs two app-server behaviors to be safe when lifecycle management is driven by a helper process: - a readiness probe must not become the process-wide client identity just because it connects first - a graceful reload signal needs to keep draining active turns even if it is delivered more than once ## What changed - Treat `codex_app_server_daemon` initialization as a probe-only client for process-global originator and user-agent suffix state. - Distinguish forceable shutdown signals from graceful-only ones, and treat Unix `SIGHUP` as graceful-only while leaving `SIGTERM` and Ctrl-C forceable. - Add regression coverage for daemon probe initialization and repeated `SIGHUP` delivery while a turn is still running. ## Testing - `cargo test -p codex-app-server` - The new daemon-probe and repeated-`SIGHUP` coverage passed. - The run still failed in the existing `suite::conversation_summary::get_conversation_summary_by_relative_rollout_path_resolves_from_codex_home` and `suite::conversation_summary::get_conversation_summary_by_thread_id_reads_rollout` tests because their initialize handshake timed out. - `cargo test -p codex-app-server --test all suite::conversation_summary::` - Reproduced the same two existing initialize-timeout failures in isolation.
2026-07-01 00:31:56 +08:00 · 2026-05-08 15:47:51 -07:00
parent dac108f2f1
commit 1b86906fa1
4 changed files with 111 additions and 26 deletions
@@ -160,22 +160,33 @@ enum ShutdownAction {
    Finish,
 }

-async fn shutdown_signal() -> IoResult<()> {
+#[derive(Clone, Copy)]
+enum ShutdownSignal {
+    Forceable,
+    #[cfg(unix)]
+    GracefulOnly,
+}
+
+async fn shutdown_signal() -> IoResult<ShutdownSignal> {
    #[cfg(unix)]
    {
        use tokio::signal::unix::SignalKind;
        use tokio::signal::unix::signal;

        let mut term = signal(SignalKind::terminate())?;
+        let mut hangup = signal(SignalKind::hangup())?;
        tokio::select! {
-            ctrl_c_result = tokio::signal::ctrl_c() => ctrl_c_result,
-            _ = term.recv() => Ok(()),
+            ctrl_c_result = tokio::signal::ctrl_c() => ctrl_c_result.map(|_| ShutdownSignal::Forceable),
+            _ = term.recv() => Ok(ShutdownSignal::Forceable),
+            _ = hangup.recv() => Ok(ShutdownSignal::GracefulOnly),
        }
    }

    #[cfg(not(unix))]
    {
-        tokio::signal::ctrl_c().await
+        tokio::signal::ctrl_c()
+            .await
+            .map(|_| ShutdownSignal::Forceable)
    }
 }

@@ -188,9 +199,16 @@ impl ShutdownState {
        self.forced
    }

-    fn on_signal(&mut self, connection_count: usize, running_turn_count: usize) {
+    fn on_signal(
+        &mut self,
+        signal: ShutdownSignal,
+        connection_count: usize,
+        running_turn_count: usize,
+    ) {
        if self.requested {
-            self.forced = true;
+            if matches!(signal, ShutdownSignal::Forceable) {
+                self.forced = true;
+            }
            return;
        }

@@ -814,11 +832,15 @@ pub async fn run_main_with_transport_options(

                tokio::select! {
                    shutdown_signal_result = shutdown_signal(), if graceful_signal_restart_enabled && !shutdown_state.forced() => {
-                        if let Err(err) = shutdown_signal_result {
-                            warn!("failed to listen for shutdown signal during graceful restart drain: {err}");
-                        }
+                        let signal = match shutdown_signal_result {
+                            Ok(signal) => signal,
+                            Err(err) => {
+                                warn!("failed to listen for shutdown signal during graceful restart drain: {err}");
+                                continue;
+                            }
+                        };
                        let running_turn_count = *running_turn_count_rx.borrow();
-                        shutdown_state.on_signal(connections.len(), running_turn_count);
+                        shutdown_state.on_signal(signal, connections.len(), running_turn_count);
                    }
                    changed = running_turn_count_rx.changed(), if graceful_signal_restart_enabled && shutdown_state.requested() => {
                        if changed.is_err() {
@@ -13,6 +13,8 @@ use super::*;
 use crate::message_processor::ConnectionSessionState;
 use crate::message_processor::InitializedConnectionSessionState;

+const DAEMON_PROBE_CLIENT_NAME: &str = "codex_app_server_daemon";
+
 #[derive(Clone)]
 pub(crate) struct InitializeRequestProcessor {
    outgoing: Arc<OutgoingMessageSender>,
@@ -90,6 +92,7 @@ impl InitializeRequestProcessor {
        }
        let originator = name.clone();
        let user_agent_suffix = format!("{name}; {version}");
+        let mutates_global_identity = name != DAEMON_PROBE_CLIENT_NAME;
        let codex_home = self.config.codex_home.clone();
        if session
            .initialize(InitializedConnectionSessionState {
@@ -104,21 +107,22 @@ impl InitializeRequestProcessor {
            return Err(invalid_request("Already initialized"));
        }

-        // Only the request that wins session initialization may mutate
-        // process-global client metadata.
-        if let Err(error) = set_default_originator(originator.clone()) {
-            match error {
-                SetOriginatorError::InvalidHeaderValue => {
-                    tracing::warn!(
-                        client_info_name = %name,
-                        "validated clientInfo.name was rejected while setting originator"
-                    );
-                }
-                SetOriginatorError::AlreadyInitialized => {
-                    // No-op. This is expected to happen if the originator is already set via env var.
-                    // TODO(owen): Once we remove support for CODEX_INTERNAL_ORIGINATOR_OVERRIDE,
-                    // this will be an unexpected state and we can return a JSON-RPC error indicating
-                    // internal server error.
+        if mutates_global_identity {
+            // Only real client initialization may mutate process-global client metadata.
+            if let Err(error) = set_default_originator(originator.clone()) {
+                match error {
+                    SetOriginatorError::InvalidHeaderValue => {
+                        tracing::warn!(
+                            client_info_name = %name,
+                            "validated clientInfo.name was rejected while setting originator"
+                        );
+                    }
+                    SetOriginatorError::AlreadyInitialized => {
+                        // No-op. This is expected to happen if the originator is already set via env var.
+                        // TODO(owen): Once we remove support for CODEX_INTERNAL_ORIGINATOR_OVERRIDE,
+                        // this will be an unexpected state and we can return a JSON-RPC error indicating
+                        // internal server error.
+                    }
                }
            }
        }
@@ -129,7 +133,7 @@ impl InitializeRequestProcessor {
            self.rpc_transport,
        );
        set_default_client_residency_requirement(self.config.enforce_residency.value());
-        if let Ok(mut suffix) = USER_AGENT_SUFFIX.lock() {
+        if mutates_global_identity && let Ok(mut suffix) = USER_AGENT_SUFFIX.lock() {
            *suffix = Some(user_agent_suffix);
        }

@@ -133,6 +133,34 @@ async fn websocket_transport_second_sigterm_forces_exit_while_turn_running() ->
    Ok(())
 }

+#[tokio::test]
+async fn websocket_transport_repeated_sighup_keeps_waiting_for_running_turn() -> Result<()> {
+    let GracefulCtrlCFixture {
+        _codex_home,
+        _server,
+        mut process,
+        mut ws,
+    } = start_ctrl_c_restart_fixture(Duration::from_secs(3)).await?;
+
+    send_sighup(&process)?;
+    assert_process_does_not_exit_within(&mut process, Duration::from_millis(300)).await?;
+
+    send_sighup(&process)?;
+    assert_process_does_not_exit_within(&mut process, Duration::from_millis(300)).await?;
+
+    let status = wait_for_process_exit_within(
+        &mut process,
+        Duration::from_secs(10),
+        "timed out waiting for graceful repeated SIGHUP restart shutdown",
+    )
+    .await?;
+    assert!(status.success(), "expected graceful exit, got {status}");
+
+    expect_websocket_disconnect(&mut ws).await?;
+
+    Ok(())
+}
+
 struct GracefulCtrlCFixture {
    _codex_home: TempDir,
    _server: wiremock::MockServer,
@@ -236,6 +264,10 @@ fn send_sigterm(process: &Child) -> Result<()> {
    send_signal(process, "-TERM")
 }

+fn send_sighup(process: &Child) -> Result<()> {
+    send_signal(process, "-HUP")
+}
+
 fn send_signal(process: &Child, signal: &str) -> Result<()> {
    let pid = process
        .id()
@@ -62,6 +62,33 @@ async fn initialize_uses_client_info_name_as_originator() -> Result<()> {
    Ok(())
 }

+#[tokio::test]
+async fn initialize_probe_does_not_override_originator() -> Result<()> {
+    let responses = Vec::new();
+    let server = create_mock_responses_server_sequence_unchecked(responses).await;
+    let codex_home = TempDir::new()?;
+    create_config_toml(codex_home.path(), &server.uri(), "never")?;
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+
+    let message = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.initialize_with_client_info(ClientInfo {
+            name: "codex_app_server_daemon".to_string(),
+            title: Some("Codex App Server Daemon".to_string()),
+            version: "0.1.0".to_string(),
+        }),
+    )
+    .await??;
+
+    let JSONRPCMessage::Response(response) = message else {
+        anyhow::bail!("expected initialize response, got {message:?}");
+    };
+    let InitializeResponse { user_agent, .. } = to_response::<InitializeResponse>(response)?;
+
+    assert!(user_agent.starts_with("codex_cli_rs/"));
+    Ok(())
+}
+
 #[tokio::test]
 async fn initialize_respects_originator_override_env_var() -> Result<()> {
    let responses = Vec::new();