diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 923eaefa2e..cd2a7e99f9 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2077,6 +2077,7 @@ dependencies = [ "clap", "codex-agent-extension", "codex-analytics", + "codex-app-server-daemon", "codex-app-server-protocol", "codex-app-server-transport", "codex-arg0", @@ -2233,6 +2234,8 @@ dependencies = [ "tokio", "tokio-tungstenite", "tracing", + "tracing-appender", + "tracing-subscriber", "windows-sys 0.61.2", ] diff --git a/codex-rs/app-server-daemon/Cargo.toml b/codex-rs/app-server-daemon/Cargo.toml index 3721f88b7d..788f92f69c 100644 --- a/codex-rs/app-server-daemon/Cargo.toml +++ b/codex-rs/app-server-daemon/Cargo.toml @@ -40,10 +40,12 @@ tokio = { workspace = true, features = [ tempfile = { workspace = true } tokio-tungstenite = { workspace = true } tracing = { workspace = true } +tracing-appender = { workspace = true } [dev-dependencies] codex-utils-cargo-bin = { workspace = true } pretty_assertions = { workspace = true } +tracing-subscriber = { workspace = true } [target.'cfg(windows)'.dependencies] -windows-sys = { workspace = true, features = ["Win32_Foundation", "Win32_Storage_FileSystem", "Win32_System_IO", "Win32_System_JobObjects", "Win32_Security", "Win32_Security_Authorization", "Win32_System_Threading"] } +windows-sys = { workspace = true, features = ["Win32_Foundation", "Win32_Storage_FileSystem", "Win32_System_Console", "Win32_System_IO", "Win32_System_JobObjects", "Win32_Security", "Win32_Security_Authorization", "Win32_System_Threading"] } diff --git a/codex-rs/app-server-daemon/src/backend/mod.rs b/codex-rs/app-server-daemon/src/backend/mod.rs index 19622b7e4c..1d2f4ecfac 100644 --- a/codex-rs/app-server-daemon/src/backend/mod.rs +++ b/codex-rs/app-server-daemon/src/backend/mod.rs @@ -1,5 +1,5 @@ mod pid; -mod stderr_log; +pub(crate) mod stderr_log; #[cfg(windows)] pub(crate) mod windows; diff --git a/codex-rs/app-server-daemon/src/backend/pid_windows.rs b/codex-rs/app-server-daemon/src/backend/pid_windows.rs index 8487fa2bc4..eaae6a6ec5 100644 --- a/codex-rs/app-server-daemon/src/backend/pid_windows.rs +++ b/codex-rs/app-server-daemon/src/backend/pid_windows.rs @@ -38,14 +38,33 @@ impl PidBackend { .context("failed to acknowledge updater startup") } + // Keep pruning excluded until successor startup and any ownership rollback finish. + #[allow(clippy::await_holding_invalid_type)] pub(crate) async fn replace_current_updater(&self) -> Result<()> { + // Retry busy pruning instead of waiting on the logging worker. Keep it paused + // before open_stderr_log() truncates the shared log and throughout startup. + let mut pruning_enabled = super::super::stderr_log::PRUNING_ENABLED + .try_lock() + .context("stderr pruning is busy; retry updater handoff")?; let record = PidRecord { pid: std::process::id(), process_start_time: read_process_start_time(std::process::id()).await?, process_identity: None, executable_identity: None, }; - self.start_inner(Some(record)).await?; + *pruning_enabled = false; + let result = self.start_inner(Some(record.clone())).await; + // Resume only after ownership is restored. Failed cleanup may leave the + // successor alive, and successful handoff leaves it in charge of retention. + if result.is_err() + && matches!( + self.read_pid_file_state().await, + Ok(PidFileState::Running(previous)) if previous == record + ) + { + *pruning_enabled = true; + } + result?; Ok(()) } diff --git a/codex-rs/app-server-daemon/src/backend/stderr_log.rs b/codex-rs/app-server-daemon/src/backend/stderr_log.rs index 4e619735f1..9c8968d512 100644 --- a/codex-rs/app-server-daemon/src/backend/stderr_log.rs +++ b/codex-rs/app-server-daemon/src/backend/stderr_log.rs @@ -1,53 +1,191 @@ -//! Saves a bounded previous-launch stderr tail before a detached launch truncates its log. -//! Preservation is best effort; missing or empty logs leave the last useful tail intact. +//! Bounds managed stderr with the shared nonblocking logging worker and tail retention. +//! The worker serializes writes with pruning and pauses pruning during Windows handoff. +//! Native writers remain best effort. use std::io; +use std::io::Read; +use std::io::Seek; +use std::io::Write; use std::path::Path; +use std::path::PathBuf; +use std::sync::OnceLock; use std::time::Instant; - -use tokio::io::AsyncReadExt; -use tokio::io::AsyncSeekExt; +use tokio::sync::Mutex; +use tracing_appender::non_blocking::NonBlocking; +use tracing_appender::non_blocking::NonBlockingBuilder; +use tracing_appender::non_blocking::WorkerGuard; const MAX_PREVIOUS_LOG_BYTES: u64 = 256 * 1024; +const MAX_CURRENT_LOG_BYTES: u64 = 1024 * 1024; + +const QUEUED_RECORDS: usize = 64; +const MAX_RECORD_BYTES: usize = 16 * 1024; +// Process lifetime: dropping WorkerGuard would wait for a potentially stalled disk. +static LOGGER: OnceLock<(NonBlocking, WorkerGuard)> = OnceLock::new(); +// Handoff holds this lock through startup and leaves predecessor pruning disabled on success. +// Both callers use try_lock so a stalled worker cannot make handoff wait on disk. +pub(super) static PRUNING_ENABLED: Mutex = Mutex::const_new(true); + +pub(crate) fn start(path: PathBuf) { + LOGGER.get_or_init(|| { + nonblocking(LogFile { + path, + file: None, + can_prune: None, + }) + }); +} + +fn nonblocking(writer: impl Write + Send + 'static) -> (NonBlocking, WorkerGuard) { + NonBlockingBuilder::default() + .buffered_lines_limit(QUEUED_RECORDS) + .lossy(/* is_lossy */ true) + .finish(writer) +} + +/// Writes complete records through the bounded managed queue, or ordinary stderr otherwise. +/// Each write must contain one complete record; oversized records are discarded before copying. +pub fn stderr_writer() -> impl Write { + StderrWriter +} + +struct StderrWriter; + +impl Write for StderrWriter { + fn write(&mut self, bytes: &[u8]) -> io::Result { + if let Some((writer, _guard)) = LOGGER.get() { + if bytes.len() <= MAX_RECORD_BYTES { + writer.clone().write_all(bytes)?; + } + Ok(bytes.len()) + } else { + io::stderr().write(bytes) + } + } + + fn flush(&mut self) -> io::Result<()> { + if LOGGER.get().is_none() { + io::stderr().flush()?; + } + Ok(()) + } +} + +struct LogFile { + path: PathBuf, + file: Option, + can_prune: Option, +} + +impl Write for LogFile { + fn write(&mut self, bytes: &[u8]) -> io::Result { + // All filesystem work, including first open and legacy handle setup, stays on the worker. + if *self + .can_prune + .get_or_insert_with(|| enable_append().is_ok()) + { + let _ = prune(&self.path, &PRUNING_ENABLED); + } + let file = match self.file { + Some(ref mut file) => file, + None => self.file.insert( + std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&self.path)?, + ), + }; + file.write_all(bytes)?; + Ok(bytes.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +#[cfg(unix)] +fn enable_append() -> io::Result<()> { + use std::os::fd::AsRawFd; + + let fd = std::io::stderr().as_raw_fd(); + // SAFETY: F_GETFL reads flags from the borrowed descriptor and takes no pointers. + let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) }; + if flags == -1 { + return Err(io::Error::last_os_error()); + } + // SAFETY: F_SETFL changes status flags on the same descriptor without taking ownership. + if unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_APPEND) } == -1 { + return Err(io::Error::last_os_error()); + } + Ok(()) +} + +#[cfg(windows)] +fn enable_append() -> io::Result<()> { + use std::os::windows::io::AsRawHandle; + use std::os::windows::io::FromRawHandle; + use std::os::windows::io::IntoRawHandle; + use std::os::windows::io::OwnedHandle; + use windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE; + use windows_sys::Win32::Storage::FileSystem::FILE_GENERIC_WRITE; + use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_DELETE; + use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_READ; + use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_WRITE; + use windows_sys::Win32::Storage::FileSystem::FILE_WRITE_DATA; + use windows_sys::Win32::Storage::FileSystem::ReOpenFile; + use windows_sys::Win32::System::Console::STD_ERROR_HANDLE; + use windows_sys::Win32::System::Console::SetStdHandle; + + let stderr = std::io::stderr(); + // SAFETY: stderr is borrowed and remains open; reopening does not truncate it. + let handle = unsafe { + ReOpenFile( + stderr.as_raw_handle(), + // Append-only access makes writes follow EOF after truncation. + FILE_GENERIC_WRITE & !FILE_WRITE_DATA, + FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE, + /*dwflagsandattributes*/ 0, + ) + }; + if handle == INVALID_HANDLE_VALUE { + return Err(io::Error::last_os_error()); + } + // SAFETY: ReOpenFile returned a new, valid handle owned by this function. + let handle = unsafe { OwnedHandle::from_raw_handle(handle) }; + // SAFETY: the append handle is valid and retained for the process lifetime on success. + if unsafe { SetStdHandle(STD_ERROR_HANDLE, handle.as_raw_handle()) } == 0 { + return Err(io::Error::last_os_error()); + } + // The process owns its new stderr. Keep the borrowed original valid for cached users. + let _ = handle.into_raw_handle(); + Ok(()) +} + +fn prune(path: &Path, pruning_enabled: &Mutex) -> io::Result<()> { + let Ok(enabled) = pruning_enabled.try_lock() else { + return Ok(()); + }; + if !*enabled { + return Ok(()); + } + let metadata = std::fs::symlink_metadata(path)?; + if !metadata.is_file() || metadata.len() <= MAX_CURRENT_LOG_BYTES { + return Ok(()); + } + preserve_tail(path)?; + // Keep inherited handles connected, holding the pruning lock through truncation. + std::fs::OpenOptions::new() + .write(true) + .open(path)? + .set_len(0) +} pub(super) async fn preserve(path: &Path) { let started = Instant::now(); + let path = path.to_path_buf(); let result = async { - match tokio::fs::symlink_metadata(path).await { - Ok(metadata) if !metadata.is_file() => return Ok(()), - Ok(_) => {} - Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()), - Err(error) => return Err(error.into()), - } - let mut file = match tokio::fs::File::open(path).await { - Ok(file) => file, - Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()), - Err(error) => return Err(error.into()), - }; - let metadata = file.metadata().await?; - let len = metadata.len(); - if !metadata.is_file() || len == 0 { - return Ok(()); - } - file.seek(io::SeekFrom::Start( - len.saturating_sub(MAX_PREVIOUS_LOG_BYTES), - )) - .await?; - let mut bytes = Vec::new(); - file.take(MAX_PREVIOUS_LOG_BYTES) - .read_to_end(&mut bytes) - .await?; - let previous = path.with_extension("log.previous"); - let temporary = path.with_extension("log.previous.tmp"); - let result = async { - tokio::fs::write(&temporary, bytes).await?; - tokio::fs::rename(&temporary, previous).await - } - .await; - if result.is_err() { - let _ = tokio::fs::remove_file(&temporary).await; - } - result?; + tokio::task::spawn_blocking(move || preserve_tail(&path)).await??; anyhow::Ok(()) } .await; @@ -56,6 +194,38 @@ pub(super) async fn preserve(path: &Path) { } } +fn preserve_tail(path: &Path) -> io::Result<()> { + match std::fs::symlink_metadata(path) { + Ok(metadata) if !metadata.is_file() => return Ok(()), + Ok(_) => {} + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()), + Err(error) => return Err(error), + } + let mut file = match std::fs::File::open(path) { + Ok(file) => file, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()), + Err(error) => return Err(error), + }; + let metadata = file.metadata()?; + let len = metadata.len(); + if !metadata.is_file() || len == 0 { + return Ok(()); + } + file.seek(io::SeekFrom::Start( + len.saturating_sub(MAX_PREVIOUS_LOG_BYTES), + ))?; + let mut bytes = Vec::new(); + file.take(MAX_PREVIOUS_LOG_BYTES).read_to_end(&mut bytes)?; + let previous = path.with_extension("log.previous"); + let temporary = path.with_extension("log.previous.tmp"); + let result = + std::fs::write(&temporary, bytes).and_then(|()| std::fs::rename(&temporary, previous)); + if result.is_err() { + let _ = std::fs::remove_file(&temporary); + } + result +} + #[cfg(test)] #[path = "stderr_log_tests.rs"] mod tests; diff --git a/codex-rs/app-server-daemon/src/backend/stderr_log_tests.rs b/codex-rs/app-server-daemon/src/backend/stderr_log_tests.rs index 85603d2f5d..1dbe1e3dfe 100644 --- a/codex-rs/app-server-daemon/src/backend/stderr_log_tests.rs +++ b/codex-rs/app-server-daemon/src/backend/stderr_log_tests.rs @@ -1,7 +1,51 @@ +use super::MAX_CURRENT_LOG_BYTES; use super::MAX_PREVIOUS_LOG_BYTES; use super::preserve; +use super::prune; use pretty_assertions::assert_eq; +#[cfg(any(unix, windows))] +#[tokio::test] +async fn pruning_resets_the_write_position_of_a_legacy_nonappend_writer() -> anyhow::Result<()> { + use std::io::Write; + + const CHILD_LOG: &str = "CODEX_TEST_LEGACY_STDERR_LOG"; + if let Some(path) = std::env::var_os(CHILD_LOG) { + std::io::stderr().write_all(b"before pruning\n")?; + let mut writer = super::LogFile { + path: path.into(), + file: None, + can_prune: None, + }; + writer.write_all(b"queued logging\n")?; + std::io::stderr().write_all(b"still logging\n")?; + return Ok(()); + } + + let home = tempfile::tempdir()?; + let path = home.path().join("daemon.stderr.log"); + let mut writer = std::fs::File::create(&path)?; + writer.write_all(&vec![b'x'; MAX_CURRENT_LOG_BYTES as usize + 1])?; + // Isolate the process-wide stderr replacement and inherit the legacy writer's offset. + let status = tokio::process::Command::new(std::env::current_exe()?) + .args([ + "--exact", + "backend::stderr_log::tests::pruning_resets_the_write_position_of_a_legacy_nonappend_writer", + "--nocapture", + ]) + .env(CHILD_LOG, &path) + .stderr(writer) + .kill_on_drop(true) + .status() + .await?; + assert!(status.success(), "legacy stderr child failed: {status}"); + assert_eq!( + tokio::fs::read(&path).await?, + b"queued logging\nstill logging\n" + ); + Ok(()) +} + #[tokio::test] async fn preserves_bounded_tail_and_keeps_it_after_empty_launch() -> anyhow::Result<()> { let home = tempfile::tempdir()?; @@ -28,3 +72,146 @@ async fn preserves_bounded_tail_and_keeps_it_after_empty_launch() -> anyhow::Res assert_eq!(tokio::fs::read(&previous).await?, b"next launch"); Ok(()) } + +#[tokio::test] +#[allow(clippy::await_holding_invalid_type)] // Exercise pruning exclusion during handoff I/O. +async fn prunes_live_log_without_disconnecting_inherited_append_handles() -> anyhow::Result<()> { + use tokio::io::AsyncWriteExt; + + let pruning_enabled = tokio::sync::Mutex::new(true); + let home = tempfile::tempdir()?; + let path = home.path().join("daemon.stderr.log"); + let mut writer = tokio::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&path) + .await?; + writer.write_all(b"startup\n").await?; + writer.flush().await?; + prune(&path, &pruning_enabled)?; + assert_eq!(tokio::fs::read(&path).await?, b"startup\n"); + + let mut bytes = vec![b'x'; MAX_CURRENT_LOG_BYTES as usize]; + bytes.extend_from_slice(b"latest failure\n"); + writer.write_all(&bytes).await?; + writer.flush().await?; + + // A failed tail publication must leave the live log untouched. + let previous = path.with_extension("log.previous"); + tokio::fs::create_dir(&previous).await?; + assert!(prune(&path, &pruning_enabled).is_err()); + assert_eq!( + tokio::fs::read(&path).await?, + [b"startup\n".as_slice(), bytes.as_slice()].concat() + ); + tokio::fs::remove_dir(&previous).await?; + + // Handoff excludes pruning even while the shared log exceeds the limit. + let mut handoff = pruning_enabled.try_lock()?; + *handoff = false; + prune(&path, &pruning_enabled)?; + assert_eq!( + tokio::fs::read(&path).await?, + [b"startup\n".as_slice(), bytes.as_slice()].concat() + ); + + // Failed handoff restores ownership, so retention resumes after unlocking. + *handoff = true; + drop(handoff); + prune(&path, &pruning_enabled)?; + assert_eq!( + tokio::fs::read(previous).await?, + bytes[bytes.len() - MAX_PREVIOUS_LOG_BYTES as usize..] + ); + writer.write_all(b"still logging\n").await?; + writer.flush().await?; + assert_eq!(tokio::fs::read(&path).await?, b"still logging\n"); + + // Successful handoff keeps predecessor pruning disabled after unlocking. + let mut handoff = pruning_enabled.try_lock()?; + *handoff = false; + writer.write_all(&bytes).await?; + writer.write_all(b"successor startup\n").await?; + writer.flush().await?; + drop(handoff); + prune(&path, &pruning_enabled)?; + assert_eq!( + tokio::fs::read(&path).await?, + [ + b"still logging\n".as_slice(), + bytes.as_slice(), + b"successor startup\n", + ] + .concat() + ); + Ok(()) +} + +#[test] +fn logging_does_not_wait_for_a_stalled_worker() -> anyhow::Result<()> { + use std::io::Write; + use std::sync::Barrier; + use std::sync::mpsc; + + const CHILD: &str = "CODEX_TEST_STALLED_LOG_WORKER"; + if std::env::var_os(CHILD).is_none() { + let output = std::process::Command::new(std::env::current_exe()?) + .args([ + "--exact", + "backend::stderr_log::tests::logging_does_not_wait_for_a_stalled_worker", + ]) + .env(CHILD, "1") + .output()?; + assert!( + output.status.success(), + "logging child failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + assert_eq!(output.stderr, Vec::::new()); + return Ok(()); + } + + struct StalledWriter(mpsc::Sender<()>); + impl Write for StalledWriter { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.send(()).unwrap(); + // No second participant: the worker stays stalled until process exit. + Barrier::new(/* n */ 2).wait(); + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + let (entered_tx, entered_rx) = mpsc::channel(); + assert!( + super::LOGGER + .set(super::nonblocking(StalledWriter(entered_tx))) + .is_ok() + ); + // The updater's diagnostic starts the stalled write. + crate::diagnostics::event("stalled-worker", serde_json::json!({})); + entered_rx.recv()?; + // Exercise the same tracing formatter/writer combination used by app-server. + let subscriber = tracing_subscriber::fmt() + .with_writer(super::stderr_writer) + .finish(); + tracing::subscriber::with_default(subscriber, || tracing::warn!("app-server event")); + super::stderr_writer().write_all(&vec![b'x'; super::MAX_RECORD_BYTES + 1])?; + for _ in 1..super::QUEUED_RECORDS { + super::stderr_writer().write_all(&vec![b'x'; super::MAX_RECORD_BYTES])?; + } + super::stderr_writer().write_all(b"queue is full\n")?; + super::stderr_writer().flush()?; + assert_eq!( + super::LOGGER + .get() + .unwrap() + .0 + .error_counter() + .dropped_lines(), + 1 + ); + // Returning also proves process shutdown does not wait for the stalled worker. + Ok(()) +} diff --git a/codex-rs/app-server-daemon/src/diagnostics.rs b/codex-rs/app-server-daemon/src/diagnostics.rs index a000bce055..3910338025 100644 --- a/codex-rs/app-server-daemon/src/diagnostics.rs +++ b/codex-rs/app-server-daemon/src/diagnostics.rs @@ -100,7 +100,7 @@ fn emit(payload: Payload) { process_pid: std::process::id(), payload, }; - let Ok(record) = serde_json::to_string(&record) else { + let Ok(mut record) = serde_json::to_string(&record) else { return; }; if tracing::dispatcher::has_been_set() { @@ -114,7 +114,8 @@ fn emit(payload: Payload) { // The detached updater has redirected stderr and no tracing subscriber. // A TUI can own a live terminal before installing its subscriber. // Never let a log write failure interrupt the lifecycle operation. - let _ = writeln!(std::io::stderr().lock(), "{record}"); + record.push('\n'); + let _ = crate::stderr_writer().write_all(record.as_bytes()); } } diff --git a/codex-rs/app-server-daemon/src/lib.rs b/codex-rs/app-server-daemon/src/lib.rs index 88a662486d..0c4f95ff2f 100644 --- a/codex-rs/app-server-daemon/src/lib.rs +++ b/codex-rs/app-server-daemon/src/lib.rs @@ -26,6 +26,7 @@ mod thread_recovery; mod update_loop; use std::collections::BTreeMap; +use std::io::Write; use std::path::Path; use std::path::PathBuf; use std::time::Duration; @@ -298,6 +299,20 @@ pub async fn run_pid_update_loop( update_loop::run(http_client_factory, restore_release).await } +pub use backend::stderr_log::stderr_writer; + +/// Routes managed app-server logs through a bounded queue to a dedicated file worker. +/// Full queues and oversized records are dropped; shutdown never waits for disk I/O. +/// Native and other-process stderr writes remain best effort during pruning. +pub fn maintain_stderr_log() -> Result<()> { + backend::stderr_log::start( + Daemon::from_environment()? + .pid_file + .with_extension("stderr.log"), + ); + Ok(()) +} + pub async fn update( http_client_factory: codex_http_client::HttpClientFactory, ) -> Result { @@ -575,9 +590,9 @@ impl Daemon { #[cfg(windows)] backend::windows::ensure_detached_launch(managed_codex_bin)?; if let Err(err) = thread_recovery::discard_pending(self) { - eprintln!( - "warning: failed to clear stale daemon recovery before update: {err}" - ); + let _ = stderr_writer().write_all(format!( + "warning: failed to clear stale daemon recovery before update: {err}\n" + ).as_bytes()); } diagnostics::event( "restart_requested", diff --git a/codex-rs/app-server-daemon/src/update_loop.rs b/codex-rs/app-server-daemon/src/update_loop.rs index ced80c7567..344aba56e5 100644 --- a/codex-rs/app-server-daemon/src/update_loop.rs +++ b/codex-rs/app-server-daemon/src/update_loop.rs @@ -1,6 +1,7 @@ //! Owns scheduled and manual installs, daemon restarts, and updater replacement. use std::collections::VecDeque; +use std::io::Write; use std::path::Path; #[cfg(unix)] use std::process::Command as StdCommand; @@ -79,13 +80,15 @@ pub(crate) async fn run( http_client_factory: HttpClientFactory, restore_release: Option, ) -> Result<()> { + let daemon = Daemon::from_environment()?; + crate::backend::stderr_log::start(daemon.update_pid_file.with_extension("stderr.log")); let http = RouteAwareClientPool::new_without_request_logging( http_client_factory, ClientRouteClass::Other, ); run_with_http( &http, - &Daemon::from_environment()?, + &daemon, ¤t_updater_identity().await?, restore_release, ) @@ -228,7 +231,7 @@ async fn run_with_http( if listener.is_none() { return Err(err); } - eprintln!("warning: failed to refresh managed updater: {err:#}"); + let _ = crate::stderr_writer().write_all(format!("warning: failed to refresh managed updater: {err:#}\n").as_bytes()); next_check = Instant::now() + Duration::from_secs(30); } } diff --git a/codex-rs/app-server/Cargo.toml b/codex-rs/app-server/Cargo.toml index 650c717e69..d0cf6a4e0c 100644 --- a/codex-rs/app-server/Cargo.toml +++ b/codex-rs/app-server/Cargo.toml @@ -35,6 +35,7 @@ axum = { workspace = true, default-features = false, features = [ "ws", ] } codex-analytics = { workspace = true } +codex-app-server-daemon = { workspace = true } codex-agent-extension = { workspace = true } codex-arg0 = { workspace = true } codex-aws-auth = { workspace = true } diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index a121cfddf1..ed787e10f9 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -17,6 +17,7 @@ use std::collections::HashMap; use std::collections::HashSet; use std::io::ErrorKind; use std::io::Result as IoResult; +use std::io::Write; use std::path::Path; use std::sync::Arc; use std::sync::RwLock; @@ -733,12 +734,12 @@ pub async fn run_main_with_transport_options( let stderr_fmt: StderrLogLayer = match log_format_from_env() { LogFormat::Json => tracing_subscriber::fmt::layer() .json() - .with_writer(std::io::stderr) + .with_writer(codex_app_server_daemon::stderr_writer) .with_span_events(stderr_span_events()) .with_filter(EnvFilter::from_default_env()) .boxed(), LogFormat::Default => tracing_subscriber::fmt::layer() - .with_writer(std::io::stderr) + .with_writer(codex_app_server_daemon::stderr_writer) .with_span_events(stderr_span_events()) .with_filter(EnvFilter::from_default_env()) .boxed(), @@ -1521,10 +1522,8 @@ fn sqlite_recovery_notice( fn emit_state_db_backup_warning(message: &str) { warn!("{message}"); if !tracing::dispatcher::has_been_set() { - #[allow(clippy::print_stderr)] - { - eprintln!("{message}"); - } + let _ = + codex_app_server_daemon::stderr_writer().write_all(format!("{message}\n").as_bytes()); } } diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index 10095f3a46..67bb615f2a 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -1285,6 +1285,14 @@ async fn cli_main( listen }; let auth = auth.try_into_settings()?; + if managed_daemon + && matches!( + &transport, + codex_app_server::AppServerTransport::UnixSocket { .. } + ) + { + let _ = codex_app_server_daemon::maintain_stderr_log(); + } let runtime_options = codex_app_server::AppServerRuntimeOptions { cloud_skills_snapshot, code_mode_host_transport: code_mode_host.into(),