diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index 1fb7b51f17..b467a33ec2 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -456,6 +456,18 @@ async fn start_uninitialized(mut args: InProcessStartArgs) -> IoResult(channel_capacity); let outbound_initialized = Arc::new(AtomicBool::new(false)); let outbound_experimental_api_enabled = Arc::new(AtomicBool::new(false)); diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 4b7e526085..ae748d401e 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -120,6 +120,7 @@ mod fuzzy_file_search; mod gateway_oauth_notifications; mod image_url; pub mod in_process; +mod log_write_warning; mod mcp_refresh; mod message_processor; mod model_catalog; @@ -691,6 +692,12 @@ pub async fn run_main_with_transport_options( }); } + let analytics_events_client = + analytics_events_client_from_config(Arc::clone(&auth_manager), &config); + let outgoing_message_sender = Arc::new(OutgoingMessageSender::new( + outgoing_tx, + analytics_events_client.clone(), + )); let feedback = CodexFeedback::new(); // Install a simple subscriber so `tracing` output is visible. Users can @@ -710,11 +717,16 @@ pub async fn run_main_with_transport_options( .boxed(), }; + let log_write_warning = log_write_warning::LogWriteWarningReporter::new( + feedback.clone(), + &outgoing_message_sender, + &config, + ); let feedback_layer = feedback.logger_layer(); let feedback_metadata_layer = feedback.metadata_layer(); let log_db = state_db .clone() - .map(|state_db| log_db::start(state_db, std::sync::Arc::new(feedback.clone()))); + .map(|state_db| log_db::start(state_db, log_write_warning.clone())); let log_db_layer = log_db .clone() .map(|layer| layer.with_filter(log_db::default_filter())); @@ -947,12 +959,6 @@ pub async fn run_main_with_transport_options( let recovery_file = daemon_recovery_file_path(&config.codex_home); let processor_handle = tokio::spawn({ let auth_manager = Arc::clone(&auth_manager); - let analytics_events_client = - analytics_events_client_from_config(Arc::clone(&auth_manager), &config); - let outgoing_message_sender = Arc::new(OutgoingMessageSender::new( - outgoing_tx, - analytics_events_client.clone(), - )); let initialize_notification_sender = outgoing_message_sender.clone(); let outbound_control_tx = outbound_control_tx; let processor = Arc::new(MessageProcessor::new(MessageProcessorArgs { diff --git a/codex-rs/app-server/src/log_write_warning.rs b/codex-rs/app-server/src/log_write_warning.rs new file mode 100644 index 0000000000..f282b1a1e0 --- /dev/null +++ b/codex-rs/app-server/src/log_write_warning.rs @@ -0,0 +1,67 @@ +use std::sync::Arc; +use std::sync::Weak; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; + +use codex_app_server_protocol::ServerNotification; +use codex_app_server_protocol::WarningNotification; +use codex_core::config::Config; +use codex_feedback::CodexFeedback; +use codex_state::LogWriteFailureReporter; + +use crate::outgoing_message::OutgoingMessageSender; + +const LOG_WRITE_WARNING_WITH_FEEDBACK: &str = "Codex couldn't save diagnostic logs to its local database. Use /feedback with logs included before closing Codex, or run `codex doctor` for diagnostics."; +const LOG_WRITE_WARNING_WITHOUT_FEEDBACK: &str = "Codex couldn't save diagnostic logs to its local database. Run `codex doctor` for diagnostics."; + +pub(crate) struct LogWriteWarningReporter { + feedback: CodexFeedback, + outgoing: Weak, + /// Permanently set by the first failure; limits warning delivery to one attempt + /// for this reporter, even if later SQLite writes succeed. + has_failure: AtomicBool, + message: &'static str, +} + +impl LogWriteWarningReporter { + pub(crate) fn new( + feedback: CodexFeedback, + outgoing: &Arc, + config: &Config, + ) -> Arc { + Arc::new(Self { + feedback, + outgoing: Arc::downgrade(outgoing), + has_failure: AtomicBool::new(false), + message: if config.feedback_enabled { + LOG_WRITE_WARNING_WITH_FEEDBACK + } else { + LOG_WRITE_WARNING_WITHOUT_FEEDBACK + }, + }) + } + + pub(crate) fn notify_failure(&self) { + if !self.has_failure.swap(true, Ordering::Relaxed) + && let Some(outgoing) = self.outgoing.upgrade() + { + outgoing.try_send_server_notification(ServerNotification::Warning( + WarningNotification { + thread_id: None, + message: self.message.to_string(), + }, + )); + } + } +} + +impl LogWriteFailureReporter for LogWriteWarningReporter { + fn report_failure(&self, diagnostic: &str) { + self.feedback.report_failure(diagnostic); + self.notify_failure(); + } +} + +#[cfg(test)] +#[path = "log_write_warning_tests.rs"] +mod tests; diff --git a/codex-rs/app-server/src/log_write_warning_tests.rs b/codex-rs/app-server/src/log_write_warning_tests.rs new file mode 100644 index 0000000000..cbccc8d6da --- /dev/null +++ b/codex-rs/app-server/src/log_write_warning_tests.rs @@ -0,0 +1,171 @@ +use super::*; +use crate::outgoing_message::OutgoingEnvelope; +use crate::outgoing_message::OutgoingMessage; +use codex_analytics::AnalyticsEventsClient; +use codex_core::config::ConfigBuilder; +use codex_feedback::CodexFeedback; +use codex_state::LogWriteFailureReporter; +use pretty_assertions::assert_eq; +use std::sync::Arc; +use tokio::sync::mpsc; + +#[tokio::test] +async fn log_write_warning_is_broadcast_once() -> anyhow::Result<()> { + let home = tempfile::tempdir()?; + let mut config = ConfigBuilder::default() + .codex_home(home.path().to_path_buf()) + .build() + .await?; + for feedback_enabled in [true, false] { + config.feedback_enabled = feedback_enabled; + let feedback = CodexFeedback::new(); + let (tx, mut rx) = mpsc::channel(/*buffer*/ 4); + let outgoing = Arc::new(OutgoingMessageSender::new( + tx, + AnalyticsEventsClient::disabled(), + )); + let reporter = LogWriteWarningReporter::new(feedback.clone(), &outgoing, &config); + reporter.report_failure("SQLite flush failed"); + let OutgoingEnvelope::Broadcast { + message: OutgoingMessage::AppServerNotification(envelope), + } = rx.try_recv()? + else { + panic!("expected a broadcast notification"); + }; + let expected = if feedback_enabled { + "Codex couldn't save diagnostic logs to its local database. Use /feedback with logs included before closing Codex, or run `codex doctor` for diagnostics." + } else { + "Codex couldn't save diagnostic logs to its local database. Run `codex doctor` for diagnostics." + }; + let ServerNotification::Warning(notification) = envelope.notification else { + panic!("expected a warning"); + }; + assert_eq!( + notification, + WarningNotification { + thread_id: None, + message: expected.to_string(), + } + ); + reporter.report_failure("another SQLite flush failed"); + assert!(rx.try_recv().is_err()); + assert_eq!( + feedback + .snapshot(/*session_id*/ None) + .log_attachment(/*logs_override*/ None) + .buffer, + b"SQLite flush failedanother SQLite flush failed" + ); + } + Ok(()) +} + +#[tokio::test] +async fn log_write_warning_keeps_feedback_when_sender_is_gone() -> anyhow::Result<()> { + let home = tempfile::tempdir()?; + let config = ConfigBuilder::default() + .codex_home(home.path().to_path_buf()) + .build() + .await?; + let feedback = CodexFeedback::new(); + let (tx, mut rx) = mpsc::channel(/*buffer*/ 4); + let outgoing = Arc::new(OutgoingMessageSender::new( + tx, + AnalyticsEventsClient::disabled(), + )); + let reporter = LogWriteWarningReporter::new(feedback.clone(), &outgoing, &config); + drop(outgoing); + reporter.report_failure("SQLite flush failed"); + assert_eq!( + feedback + .snapshot(/*session_id*/ None) + .log_attachment(/*logs_override*/ None) + .buffer, + b"SQLite flush failed" + ); + assert!(reporter.outgoing.upgrade().is_none()); + reporter.report_failure("second failure"); + assert_eq!( + feedback + .snapshot(/*session_id*/ None) + .log_attachment(/*logs_override*/ None) + .buffer, + b"SQLite flush failedsecond failure" + ); + assert!(rx.try_recv().is_err()); + Ok(()) +} + +#[tokio::test] +async fn log_write_warning_reports_failed_sqlite_flush() -> anyhow::Result<()> { + use tracing_subscriber::layer::SubscriberExt; + + for attach_after_failure in [false, true] { + let home = tempfile::tempdir()?; + let config = ConfigBuilder::default() + .codex_home(home.path().to_path_buf()) + .build() + .await?; + let feedback = CodexFeedback::new(); + let runtime = codex_state::StateRuntime::init( + config.sqlite_config().clone(), + "test-provider".to_string(), + ) + .await?; + let (tx, mut rx) = mpsc::channel(/*buffer*/ 4); + let outgoing = Arc::new(OutgoingMessageSender::new( + tx, + AnalyticsEventsClient::disabled(), + )); + let reporter = LogWriteWarningReporter::new(feedback.clone(), &outgoing, &config); + let initial_reporter: Arc = if attach_after_failure { + Arc::new(feedback.clone()) + } else { + reporter.clone() + }; + let layer = codex_state::log_db::start(runtime.clone(), initial_reporter); + runtime.close().await; + tracing::subscriber::with_default( + tracing_subscriber::registry().with(layer.clone()), + || { + tracing::info!("first failed log write"); + }, + ); + layer.flush().await; + assert!(layer.has_write_failure()); + if attach_after_failure { + layer.set_failure_reporter(reporter.clone()); + if layer.has_write_failure() { + reporter.notify_failure(); + } + } + + assert!(matches!( + rx.try_recv(), + Ok(OutgoingEnvelope::Broadcast { + message: OutgoingMessage::AppServerNotification(_), + .. + }) + )); + tracing::subscriber::with_default( + tracing_subscriber::registry().with(layer.clone()), + || { + tracing::info!("second failed log write"); + }, + ); + layer.flush().await; + assert!(rx.try_recv().is_err()); + assert_eq!( + String::from_utf8( + feedback + .snapshot(/*session_id*/ None) + .log_attachment(/*logs_override*/ None) + .buffer + )? + .matches("failed to flush logs to SQLite") + .count(), + 2 + ); + } + Ok(()) +} diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index e6a109ceb3..961eccc2aa 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -750,6 +750,14 @@ impl OutgoingMessageSender { .await; } + pub(crate) fn try_send_server_notification(&self, notification: ServerNotification) { + if let Err(error) = self.sender.try_send(OutgoingEnvelope::Broadcast { + message: timestamped_server_notification(notification), + }) { + warn!("failed to queue server notification to clients: {error}"); + } + } + pub(crate) async fn send_server_notification_to_connections( &self, connection_ids: &[ConnectionId], diff --git a/codex-rs/app-server/src/request_processors/initialize_processor.rs b/codex-rs/app-server/src/request_processors/initialize_processor.rs index 9ee058aef7..d68a29df9f 100644 --- a/codex-rs/app-server/src/request_processors/initialize_processor.rs +++ b/codex-rs/app-server/src/request_processors/initialize_processor.rs @@ -31,6 +31,7 @@ pub(crate) struct InitializeRequestProcessor { } impl InitializeRequestProcessor { + #[allow(clippy::too_many_arguments)] pub(crate) fn new( gateway_login_control: Arc, outgoing: Arc, diff --git a/codex-rs/app-server/tests/suite/v2/feedback.rs b/codex-rs/app-server/tests/suite/v2/feedback.rs index c594d6237f..fbe609df32 100644 --- a/codex-rs/app-server/tests/suite/v2/feedback.rs +++ b/codex-rs/app-server/tests/suite/v2/feedback.rs @@ -218,6 +218,20 @@ async fn feedback_upload_includes_sqlite_flush_and_query_failures() -> Result<() app_server .send_notification(ClientNotification::Initialized) .await?; + // The periodic writer must notify the user before they choose to submit feedback. + if phase == Phase::WriteFailure { + let notification = timeout( + Duration::from_secs(/*secs*/ 20), + app_server.read_stream_until_notification_message("warning"), + ) + .await??; + let warning: codex_app_server_protocol::WarningNotification = + serde_json::from_value(notification.params.unwrap())?; + assert_eq!(warning, codex_app_server_protocol::WarningNotification { + thread_id: None, + message: "Codex couldn't save diagnostic logs to its local database. Use /feedback with logs included before closing Codex, or run `codex doctor` for diagnostics.".to_string(), + }); + } let request = app_server .send_raw_request( "feedback/upload", @@ -256,6 +270,12 @@ async fn feedback_upload_includes_sqlite_flush_and_query_failures() -> Result<() } let logs = &attachments["codex-logs.log"]; if phase == Phase::Corrupt { + assert!( + !app_server + .pending_notification_methods() + .iter() + .any(|method| method == "warning") + ); assert!(logs.contains("failed to flush logs to SQLite error=\"corrupt\"")); assert!(logs.contains("failed to query feedback logs from sqlite")); } else { diff --git a/codex-rs/state/src/log_db.rs b/codex-rs/state/src/log_db.rs index 4b2d8a6c90..0361312cab 100644 --- a/codex-rs/state/src/log_db.rs +++ b/codex-rs/state/src/log_db.rs @@ -24,6 +24,7 @@ use std::future::Future; use std::sync::Arc; use std::sync::OnceLock; +use std::sync::RwLock; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; use std::time::Duration; @@ -136,6 +137,7 @@ pub trait LogWriteFailureReporter: Send + Sync { pub struct LogDbLayer { sender: mpsc::Sender, has_write_failure: Arc, + failure_reporter: Arc>>, process_uuid: String, } @@ -152,6 +154,7 @@ impl Clone for LogDbLayer { Self { sender: self.sender.clone(), has_write_failure: self.has_write_failure.clone(), + failure_reporter: self.failure_reporter.clone(), process_uuid: self.process_uuid.clone(), } } @@ -173,20 +176,30 @@ impl LogDbLayer { let config = config.normalized(); let (sender, receiver) = mpsc::channel(config.queue_capacity); let has_write_failure = Arc::new(AtomicBool::new(false)); + let failure_reporter = Arc::new(RwLock::new(failure_reporter)); tokio::spawn(run_inserter( state_db, receiver, config, - failure_reporter, + failure_reporter.clone(), has_write_failure.clone(), )); Self { sender, has_write_failure, + failure_reporter, process_uuid: current_process_log_uuid().to_string(), } } + /// Replaces the startup reporter once the client notification destination is available. + pub fn set_failure_reporter(&self, reporter: Arc) { + *self + .failure_reporter + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) = reporter; + } + /// Whether this writer has lost logs, making SQLite incomplete for feedback. pub fn has_write_failure(&self) -> bool { self.has_write_failure.load(Ordering::Relaxed) @@ -479,7 +492,7 @@ async fn run_inserter( state_db: std::sync::Arc, mut receiver: mpsc::Receiver, config: LogSinkQueueConfig, - failure_reporter: Arc, + failure_reporter: Arc>>, has_write_failure: Arc, ) { let mut buffer = Vec::with_capacity(config.batch_size); @@ -516,7 +529,7 @@ async fn run_inserter( async fn flush( state_db: &StateRuntime, buffer: &mut Vec, - failure_reporter: &dyn LogWriteFailureReporter, + failure_reporter: &RwLock>, has_write_failure: &AtomicBool, ) { if buffer.is_empty() { @@ -534,10 +547,10 @@ async fn flush( let timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, /*use_z*/ true); let entry_count = entries.len(); - failure_reporter.report_failure(&format!( + has_write_failure.store(true, Ordering::Relaxed); + failure_reporter.read().unwrap_or_else(std::sync::PoisonError::into_inner).report_failure(&format!( "{timestamp} ERROR failed to flush logs to SQLite error={error:?} entries={entry_count}\n" )); - has_write_failure.store(true, Ordering::Relaxed); } crate::telemetry::record_log_write( @@ -896,6 +909,7 @@ mod tests { let layer = LogDbLayer { sender, has_write_failure: Arc::new(AtomicBool::new(false)), + failure_reporter: Arc::new(RwLock::new(Arc::new(SharedWriter::default()))), process_uuid: "process-1".to_string(), }; @@ -917,6 +931,7 @@ mod tests { let layer = LogDbLayer { sender, has_write_failure: Arc::new(AtomicBool::new(false)), + failure_reporter: Arc::new(RwLock::new(Arc::new(SharedWriter::default()))), process_uuid: "process-1".to_string(), }; diff --git a/codex-rs/tui/src/chatwidget/tests/app_server.rs b/codex-rs/tui/src/chatwidget/tests/app_server.rs index 9779268c92..7eeed4e626 100644 --- a/codex-rs/tui/src/chatwidget/tests/app_server.rs +++ b/codex-rs/tui/src/chatwidget/tests/app_server.rs @@ -2013,3 +2013,28 @@ async fn permission_discovery_invalidates_on_thread_settings_and_uses_updated_cw }; assert_eq!(thread_cwd, Some(test_path_buf("/tmp/thread-settings"))); } + +#[tokio::test] +async fn sqlite_log_write_warning_is_visible_in_warnings() { + let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.warning_display_state.startup_complete = true; + chat.handle_server_notification( + ServerNotification::Warning(WarningNotification { + thread_id: None, + message: "Codex couldn't save diagnostic logs to its local database. Use /feedback with logs included before closing Codex, or run `codex doctor` for diagnostics.".to_string(), + }), + /*replay_kind*/ None, + ); + let cells: Vec> = std::iter::from_fn(|| rx.try_recv().ok()) + .filter_map(|event| match event { + AppEvent::InsertHistoryCell(cell) => Some(Arc::from(cell)), + _ => None, + }) + .collect(); + assert_eq!(cells.len(), 1); + chat.open_warnings(&cells); + insta::assert_snapshot!( + "sqlite_log_write_warning", + super::helpers::render_bottom_popup(&chat, /*width*/ 80) + ); +} diff --git a/codex-rs/tui/src/chatwidget/tests/snapshots/codex_tui__chatwidget__tests__app_server__sqlite_log_write_warning.snap b/codex-rs/tui/src/chatwidget/tests/snapshots/codex_tui__chatwidget__tests__app_server__sqlite_log_write_warning.snap new file mode 100644 index 0000000000..14f59f8ef9 --- /dev/null +++ b/codex-rs/tui/src/chatwidget/tests/snapshots/codex_tui__chatwidget__tests__app_server__sqlite_log_write_warning.snap @@ -0,0 +1,16 @@ +--- +source: tui/src/chatwidget/tests/app_server.rs +expression: "super::helpers::render_bottom_popup(&chat, 80)" +--- + Warnings · 1 of 1 · Warning + + Codex couldn't save diagnostic logs to its local database. Use /feedback + with logs included before closing Codex, or run `codex doctor` for + diagnostics. + + + + + + + esc back · ctrl+o copy · ←/→ warning · ↓ scroll