Warn users when SQLite diagnostic log writes fail (#47887)

## Why

Failed SQLite writes leave diagnostic logs incomplete. Notify users before they close Codex so they can submit logs or run `codex doctor`.

## What changed

- Add a best-effort warning on the first log write failure in standalone and in-process app servers, while continuing to retain failure diagnostics in memory.
- Handle failures that occurred before the in-process warning reporter was attached.
- Tailor the warning's guidance to whether submitting logs is enabled.

## Testing

Add coverage for warning deduplication, both guidance variants, retained diagnostics after the notification sender is dropped, and SQLite flush failures before and after reporter attachment. Extend log upload integration coverage and add a TUI warnings snapshot.

GitOrigin-RevId: 12a74126d13c3568430e539a180cca47fa931fa9

codex-rs/app-server/src/in_process.rs

@@ -456,6 +456,18 @@ async fn start_uninitialized(mut args: InProcessStartArgs) -> IoResult<InProcess
             analytics_events_client.clone(),
         ));
 
+        let log_write_warning = crate::log_write_warning::LogWriteWarningReporter::new(
+            args.feedback.clone(),
+            &outgoing_message_sender,
+            &args.config,
+        );
+        if let Some(log_db) = &args.log_db {
+            log_db.set_failure_reporter(log_write_warning.clone());
+            if log_db.has_write_failure() {
+                log_write_warning.notify_failure();
+            }
+        }
+
         let (writer_tx, mut writer_rx) = mpsc::channel::<QueuedOutgoingMessage>(channel_capacity);
         let outbound_initialized = Arc::new(AtomicBool::new(false));
         let outbound_experimental_api_enabled = Arc::new(AtomicBool::new(false));

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 {

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<OutgoingMessageSender>,
+    /// 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<OutgoingMessageSender>,
+        config: &Config,
+    ) -> Arc<Self> {
+        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;

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<dyn LogWriteFailureReporter> = 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(())
+}

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],

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<codex_login::GatewayLoginControl>,
         outgoing: Arc<OutgoingMessageSender>,

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 {