Preserve diagnostic logs when SQLite logging fails (#47886)

## Why

SQLite write failures can leave persisted logs incomplete, while log snapshots taken before collection miss errors raised during that collection.

## What changed

- Report SQLite flush failures directly to the in-memory log buffer without re-entering tracing, using an error category and batch size instead of raw error details.
- Track whether the log writer has lost a batch and make app-server diagnostic uploads use the memory buffer when SQLite logs are incomplete.
- Refresh log bytes after flushing and querying SQLite so the same submission includes collection failures, while preserving captured metadata and thread identity.

## Testing

Add regression tests for failure reporting without recursion, subsequent write recovery, metadata preservation, and healthy and failing SQLite collection. Add an upload integration test that captures attachments locally and checks failure diagnostics, memory-buffer fallback, and omission of raw SQLite error details.

GitOrigin-RevId: 51b0c4b69c977d9f4e74f2f245a9fd90fdb13761

codex-rs/Cargo.lock

@@ -2126,6 +2126,7 @@ dependencies = [
  "opentelemetry",
  "opentelemetry_sdk",
  "pretty_assertions",
+ "rcgen",
  "rmcp",
  "serde",
  "serde_json",
@@ -2140,6 +2141,7 @@ dependencies = [
  "tikv-jemallocator",
  "time",
  "tokio",
+ "tokio-rustls",
  "tokio-tungstenite",
  "tokio-util",
  "toml 0.9.11+spec-1.1.0",
@@ -3358,6 +3360,7 @@ dependencies = [
  "codex-login",
  "codex-protocol",
  "codex-rollout",
+ "codex-state",
  "flate2",
  "http 1.4.0",
  "httpdate",

codex-rs/app-server/Cargo.toml

@@ -142,6 +142,7 @@ http = { workspace = true }
 opentelemetry = { workspace = true }
 opentelemetry_sdk = { workspace = true }
 pretty_assertions = { workspace = true }
+rcgen = { workspace = true }
 rmcp = { workspace = true, default-features = false, features = [
     "elicitation",
     "server",
@@ -152,6 +153,7 @@ shlex = { workspace = true }
 sqlx = { workspace = true }
 tar = { workspace = true }
 test-case = "3.3.1"
+tokio-rustls = { workspace = true }
 tokio-tungstenite = { workspace = true }
 tracing-opentelemetry = { workspace = true }
 wiremock = { workspace = true }

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

@@ -712,7 +712,9 @@ pub async fn run_main_with_transport_options(
 
     let feedback_layer = feedback.logger_layer();
     let feedback_metadata_layer = feedback.metadata_layer();
-    let log_db = state_db.clone().map(log_db::start);
+    let log_db = state_db
+        .clone()
+        .map(|state_db| log_db::start(state_db, std::sync::Arc::new(feedback.clone())));
     let log_db_layer = log_db
         .clone()
         .map(|layer| layer.with_filter(log_db::default_filter()));

codex-rs/app-server/src/request_processors/feedback_log_snapshot_tests.rs

@@ -0,0 +1,82 @@
+//! Exercises the real SQLite writer, feedback ring buffer, and snapshot ordering.
+use super::*;
+use codex_state::SqliteConfig;
+use codex_state::StateRuntime;
+use codex_utils_absolute_path::test_support::PathExt;
+use pretty_assertions::assert_eq;
+use tracing_subscriber::prelude::*;
+
+#[derive(PartialEq)]
+enum Scenario {
+    Healthy,
+    WriteFailure,
+    Corruption,
+}
+
+#[tokio::test]
+async fn feedback_includes_flush_and_query_failures_in_the_same_submission() -> anyhow::Result<()> {
+    for scenario in [
+        Scenario::Healthy,
+        Scenario::WriteFailure,
+        Scenario::Corruption,
+    ] {
+        let home = tempfile::tempdir()?;
+        let sqlite = SqliteConfig::new_for_testing(home.path().abs());
+        let runtime = StateRuntime::init(sqlite.clone(), "test-provider".to_string()).await?;
+        let pool = sqlite.open_read_write_pool(&sqlite.logs_db_path()).await?;
+        let feedback = CodexFeedback::new();
+        let writer =
+            codex_state::log_db::start(runtime.clone(), std::sync::Arc::new(feedback.clone()));
+        let _guard = tracing_subscriber::registry()
+            .with(feedback.logger_layer())
+            .with(writer.clone())
+            .set_default();
+        let thread_id = ThreadId::new();
+        tracing::info_span!("thread", thread_id = %thread_id)
+            .in_scope(|| tracing::info!("persisted log"));
+        writer.flush().await;
+        match scenario {
+            Scenario::WriteFailure => {
+                sqlx::query("CREATE TRIGGER fail_log_insert BEFORE INSERT ON logs BEGIN SELECT RAISE(ABORT, 'synthetic-secret'); END").execute(&pool).await?;
+            }
+            Scenario::Corruption => {
+                sqlx::raw_sql("PRAGMA writable_schema = ON; UPDATE sqlite_schema SET rootpage = 2147483647 WHERE name = 'logs'; PRAGMA schema_version = 1000000;").execute(&pool).await?;
+            }
+            Scenario::Healthy => {}
+        }
+        let snapshot = feedback.snapshot(Some(thread_id));
+        tracing::info_span!("thread", thread_id = %thread_id)
+            .in_scope(|| tracing::info!("latest log"));
+        let (snapshot, sqlite_logs) = collect_feedback_logs(
+            &feedback,
+            Some(&writer),
+            Some(&runtime),
+            snapshot,
+            &[thread_id],
+        )
+        .await;
+        assert_eq!(sqlite_logs.is_some(), scenario == Scenario::Healthy);
+        let logs = String::from_utf8(snapshot.log_attachment(sqlite_logs).buffer)?;
+        assert!(logs.contains("persisted log"));
+        assert!(logs.contains("latest log"));
+        let expected_error = match scenario {
+            Scenario::Healthy => None,
+            Scenario::WriteFailure => Some("constraint"),
+            Scenario::Corruption => Some("corrupt"),
+        };
+        assert_eq!(
+            logs.matches("failed to flush logs to SQLite").count(),
+            usize::from(expected_error.is_some())
+        );
+        if let Some(error) = expected_error {
+            assert!(logs.contains(&format!("failed to flush logs to SQLite error={error:?}")));
+        }
+        assert_eq!(
+            logs.contains("failed to query feedback logs from sqlite"),
+            scenario == Scenario::Corruption
+        );
+        pool.close().await;
+        runtime.close().await;
+    }
+    Ok(())
+}

codex-rs/app-server/src/request_processors/feedback_processor.rs

@@ -6,6 +6,7 @@ use codex_connectors::ConnectorDirectoryCacheKey;
 use codex_connectors::connector_runtime_cache_path;
 use codex_feedback::CODEX_APP_DIRECTORY_CACHE_ATTACHMENT_FILENAME;
 use codex_feedback::CODEX_APPS_TOOLS_CACHE_ATTACHMENT_FILENAME;
+use codex_feedback::FeedbackSnapshot;
 #[cfg(target_os = "windows")]
 use codex_feedback::WINDOWS_SANDBOX_LOG_ATTACHMENT_FILENAME;
 use codex_feedback::guardian_review_failures;
@@ -124,13 +125,9 @@ impl FeedbackRequestProcessor {
             tracing::info!(target: "feedback_tags", account_id);
         }
         let snapshot = self.feedback.snapshot(conversation_id);
-        let thread_id = snapshot.thread_id.clone();
         let mut extra_attachments = Vec::new();
         let mut feedback_index = None;
-        let (sqlite_feedback_logs, state_db_ctx) = if include_logs {
-            if let Some(log_db) = self.log_db.as_ref() {
-                log_db.flush().await;
-            }
+        let (snapshot, sqlite_feedback_logs, state_db_ctx) = if include_logs {
             let state_db_ctx = self.state_db.clone();
             let feedback_thread_ids = match conversation_id {
                 Some(conversation_id) => match self
@@ -161,38 +158,19 @@ impl FeedbackRequestProcessor {
                 feedback_index = Some(index);
             }
             extra_attachments.extend(failures.attachment);
-            let sqlite_feedback_logs = if let Some(state_db_ctx) = state_db_ctx.as_ref()
-                && !feedback_thread_ids.is_empty()
-            {
-                let thread_id_texts = feedback_thread_ids
-                    .iter()
-                    .map(ToString::to_string)
-                    .collect::<Vec<_>>();
-                let thread_id_refs = thread_id_texts
-                    .iter()
-                    .map(String::as_str)
-                    .collect::<Vec<_>>();
-                match state_db_ctx
-                    .query_feedback_logs_for_threads(&thread_id_refs)
-                    .await
-                {
-                    Ok(logs) if logs.is_empty() => None,
-                    Ok(logs) => Some(logs),
-                    Err(err) => {
-                        let thread_ids = thread_id_texts.join(", ");
-                        warn!(
-                            "failed to query feedback logs from sqlite for thread_ids=[{thread_ids}]: {err}"
-                        );
-                        None
-                    }
-                }
-            } else {
-                None
-            };
-            (sqlite_feedback_logs, state_db_ctx)
+            let (snapshot, sqlite_feedback_logs) = collect_feedback_logs(
+                &self.feedback,
+                self.log_db.as_ref(),
+                state_db_ctx.as_ref(),
+                snapshot,
+                &feedback_thread_ids,
+            )
+            .await;
+            (snapshot, sqlite_feedback_logs, state_db_ctx)
         } else {
-            (None, None)
+            (snapshot, None, None)
         };
+        let thread_id = snapshot.thread_id.clone();
 
         let mut attachment_paths = Vec::new();
         let mut seen_attachment_paths = HashSet::new();
@@ -503,6 +481,49 @@ fn windows_sandbox_log_attachment(_codex_home: &Path) -> Option<FeedbackAttachme
     None
 }
 
+// Refresh logs after SQLite collection without changing the earlier metadata snapshot.
+async fn collect_feedback_logs(
+    feedback: &CodexFeedback,
+    log_db: Option<&LogDbLayer>,
+    state_db: Option<&StateDbHandle>,
+    mut snapshot: FeedbackSnapshot,
+    thread_ids: &[ThreadId],
+) -> (FeedbackSnapshot, Option<Vec<u8>>) {
+    if let Some(log_db) = log_db {
+        log_db.flush().await;
+    }
+    let mut sqlite_logs = None;
+    if let Some(state_db) = state_db
+        && !thread_ids.is_empty()
+    {
+        let thread_ids = thread_ids
+            .iter()
+            .map(ToString::to_string)
+            .collect::<Vec<_>>();
+        let thread_id_refs = thread_ids.iter().map(String::as_str).collect::<Vec<_>>();
+        match state_db
+            .query_feedback_logs_for_threads(&thread_id_refs)
+            .await
+        {
+            Ok(logs) if logs.is_empty() => {}
+            Ok(logs) => sqlite_logs = Some(logs),
+            Err(err) => {
+                let thread_ids = thread_ids.join(", ");
+                tracing::warn!(
+                    "failed to query feedback logs from sqlite for thread_ids=[{thread_ids}]: {err}"
+                );
+            }
+        }
+    }
+    if log_db.is_some_and(LogDbLayer::has_write_failure) {
+        sqlite_logs = None;
+    }
+
+    // new logs might arrive to the buffer while collecting metadata e.g. if log db flush fails
+    snapshot.refresh_logs(feedback);
+    (snapshot, sqlite_logs)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -885,3 +906,7 @@ mod tests {
         );
     }
 }
+
+#[cfg(test)]
+#[path = "feedback_log_snapshot_tests.rs"]
+mod feedback_log_snapshot_tests;

codex-rs/app-server/tests/suite/v2/feedback.rs

@@ -90,3 +90,180 @@ async fn feedback_upload_limits_concurrency_and_releases_failed_uploads() -> Res
     assert_eq!(error.error.code, -32603);
     Ok(())
 }
+
+#[tokio::test]
+async fn feedback_upload_includes_sqlite_flush_and_query_failures() -> Result<()> {
+    use std::io::Read;
+    use std::sync::Arc;
+
+    use app_test_support::MockResponsesConfig;
+    use codex_app_server_protocol::ClientNotification;
+    use codex_app_server_protocol::ThreadStartParams;
+    use codex_state::SqliteConfig;
+    use codex_utils_absolute_path::test_support::PathExt;
+    use tokio::io::AsyncReadExt;
+    use tokio_rustls::TlsAcceptor;
+    use tokio_rustls::rustls;
+
+    // Terminate HTTPS locally and capture real Sentry envelopes. Never forward uploads.
+    let home = tempfile::tempdir()?;
+    let cert = rcgen::generate_simple_self_signed(vec!["o33249.ingest.us.sentry.io".to_string()])?;
+    let cert_path = home.path().join("feedback-ca.pem");
+    std::fs::write(&cert_path, cert.cert.pem())?;
+    let tls = rustls::ServerConfig::builder_with_provider(Arc::new(
+        rustls::crypto::aws_lc_rs::default_provider(),
+    ))
+    .with_safe_default_protocol_versions()?
+    .with_no_client_auth()
+    .with_single_cert(
+        vec![cert.cert.der().clone()],
+        rustls::pki_types::PrivatePkcs8KeyDer::from(cert.signing_key.serialize_der()).into(),
+    )?;
+    let acceptor = TlsAcceptor::from(Arc::new(tls));
+    let listener = TcpListener::bind("127.0.0.1:0").await?;
+    let proxy_uri = format!("http://{}", listener.local_addr()?);
+    let (uploads_tx, mut uploads_rx) = tokio::sync::mpsc::unbounded_channel();
+    let _proxy = tokio_util::task::AbortOnDropHandle::new(tokio::spawn(async move {
+        while let Ok((stream, _)) = listener.accept().await {
+            let mut stream = BufReader::new(stream);
+            loop {
+                let mut line = String::new();
+                anyhow::ensure!(stream.read_line(&mut line).await? > 0, "missing CONNECT");
+                if line == "\r\n" {
+                    break;
+                }
+            }
+            stream
+                .write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n")
+                .await?;
+            let mut stream = BufReader::new(acceptor.accept(stream.into_inner()).await?);
+            let mut length = None;
+            loop {
+                let mut line = String::new();
+                anyhow::ensure!(
+                    stream.read_line(&mut line).await? > 0,
+                    "missing HTTP headers"
+                );
+                if line == "\r\n" {
+                    break;
+                }
+                if let Some((name, value)) = line.split_once(':')
+                    && name.eq_ignore_ascii_case("content-length")
+                {
+                    length = Some(value.trim().parse::<usize>()?);
+                }
+            }
+            let length = length.ok_or_else(|| anyhow::anyhow!("missing content length"))?;
+            anyhow::ensure!(length < 16 * 1024 * 1024, "unexpected envelope size");
+            let mut body = vec![0; length];
+            stream.read_exact(&mut body).await?;
+            if body.starts_with(&[0x1f, 0x8b]) {
+                let mut decoded = Vec::new();
+                flate2::read::GzDecoder::new(body.as_slice()).read_to_end(&mut decoded)?;
+                body = decoded;
+            }
+            uploads_tx.send(body)?;
+            stream
+                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{}")
+                .await?;
+            stream.flush().await?;
+        }
+        Ok::<(), anyhow::Error>(())
+    }));
+
+    let models = wiremock::MockServer::start().await;
+    MockResponsesConfig::new(&models.uri()).write(home.path())?;
+    let mut app_server = TestAppServer::builder()
+        .with_codex_home(home.path())
+        .with_env_overrides(&[
+            ("HTTPS_PROXY", Some(proxy_uri.as_str())),
+            ("https_proxy", Some(proxy_uri.as_str())),
+            ("NO_PROXY", Some("127.0.0.1,localhost")),
+            ("no_proxy", Some("127.0.0.1,localhost")),
+            ("SSL_CERT_FILE", cert_path.to_str()),
+        ])
+        .build_initialized()
+        .await?;
+    let request = app_server
+        .send_thread_start_request_with_auto_env(ThreadStartParams {
+            ephemeral: Some(true),
+            ..Default::default()
+        })
+        .await?;
+    let response = app_server
+        .read_stream_until_response_message(RequestId::Integer(request))
+        .await?;
+    let thread_id = response.result["thread"]["id"].as_str().unwrap();
+    let sqlite = SqliteConfig::new_for_testing(home.path().abs());
+    let pool = sqlite.open_read_write_pool(&sqlite.logs_db_path()).await?;
+    // This old history exists only in SQLite, as if it had already left the memory ring.
+    sqlx::query("INSERT INTO logs (ts, ts_nanos, level, target, feedback_log_body, thread_id) VALUES (1, 0, 'INFO', 'test', 'sqlite-only-history', ?)")
+        .bind(thread_id).execute(&pool).await?;
+    sqlx::query("CREATE TRIGGER fail_log_insert BEFORE INSERT ON logs BEGIN SELECT RAISE(ABORT, 'synthetic-secret'); END")
+        .execute(&pool).await?;
+
+    #[derive(Debug, PartialEq)]
+    enum Phase {
+        WriteFailure,
+        Corrupt,
+    }
+    for phase in [Phase::WriteFailure, Phase::Corrupt] {
+        match phase {
+            Phase::Corrupt => {
+                sqlx::raw_sql("PRAGMA writable_schema = ON; UPDATE sqlite_schema SET rootpage = 2147483647 WHERE name = 'logs'; PRAGMA schema_version = 1000000;").execute(&pool).await?;
+            }
+            Phase::WriteFailure => {}
+        }
+        // Notifications emit an INFO log, ensuring feedback/upload has a batch to flush.
+        app_server
+            .send_notification(ClientNotification::Initialized)
+            .await?;
+        let request = app_server
+            .send_raw_request(
+                "feedback/upload",
+                Some(json!({
+                    "classification": "bug", "includeLogs": true, "threadId": thread_id
+                })),
+            )
+            .await?;
+        timeout(
+            Duration::from_secs(/*secs*/ 30),
+            app_server.read_stream_until_response_message(RequestId::Integer(request)),
+        )
+        .await??;
+        let mut attachments = std::collections::BTreeMap::new();
+        while let Ok(envelope) = uploads_rx.try_recv() {
+            let mut bytes = envelope.as_slice();
+            // Skip the envelope header, then decode each length-delimited item.
+            let header_end = bytes.iter().position(|byte| *byte == b'\n').unwrap();
+            bytes = &bytes[header_end + 1..];
+            while !bytes.is_empty() {
+                let header_end = bytes.iter().position(|byte| *byte == b'\n').unwrap();
+                let header: serde_json::Value = serde_json::from_slice(&bytes[..header_end])?;
+                let length = header["length"].as_u64().unwrap() as usize;
+                bytes = &bytes[header_end + 1..];
+                if let Some(filename) = header["filename"].as_str() {
+                    attachments.insert(
+                        filename.to_string(),
+                        String::from_utf8_lossy(&bytes[..length]).into_owned(),
+                    );
+                }
+                bytes = &bytes[length..];
+                if bytes.starts_with(b"\n") {
+                    bytes = &bytes[1..];
+                }
+            }
+        }
+        let logs = &attachments["codex-logs.log"];
+        if phase == Phase::Corrupt {
+            assert!(logs.contains("failed to flush logs to SQLite error=\"corrupt\""));
+            assert!(logs.contains("failed to query feedback logs from sqlite"));
+        } else {
+            assert!(logs.contains("failed to flush logs to SQLite error=\"constraint\""));
+            assert!(!logs.contains("sqlite-only-history"));
+            assert!(!logs.contains("synthetic-secret"));
+        }
+    }
+    pool.close().await;
+    Ok(())
+}

codex-rs/core/tests/suite/sqlite_state.rs

@@ -987,7 +987,8 @@ async fn tool_call_logs_include_thread_id() -> Result<()> {
 
     test.submit_turn("run a shell command").await?;
 
-    let log_db_layer = codex_state::log_db::start(db.clone());
+    let log_db_layer =
+        codex_state::log_db::start(db.clone(), Arc::new(codex_feedback::CodexFeedback::new()));
     let subscriber = tracing_subscriber::registry().with(log_db_layer.clone());
     let dispatch = tracing::Dispatch::new(subscriber);
     tracing::dispatcher::with_default(&dispatch, || {

codex-rs/feedback/Cargo.toml

@@ -14,6 +14,7 @@ codex-http-client = { workspace = true }
 codex-login = { workspace = true }
 codex-protocol = { workspace = true }
 codex-rollout = { workspace = true }
+codex-state = { workspace = true }
 flate2 = { workspace = true }
 http = { workspace = true }
 httpdate = { workspace = true }

codex-rs/feedback/src/lib.rs

@@ -21,6 +21,7 @@ use codex_http_client::RouteAwareClientPool;
 use codex_login::AuthEnvTelemetry;
 use codex_protocol::ThreadId;
 use codex_protocol::protocol::SessionSource;
+use codex_state::LogWriteFailureReporter;
 use tracing::Event;
 use tracing::Level;
 use tracing::field::Visit;
@@ -196,6 +197,17 @@ pub struct CodexFeedback {
     inner: Arc<FeedbackInner>,
 }
 
+impl LogWriteFailureReporter for CodexFeedback {
+    fn report_failure(&self, diagnostic: &str) {
+        // Bypass tracing so this diagnostic cannot return to the SQLite writer.
+        self.inner
+            .ring
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .push_bytes(diagnostic.as_bytes());
+    }
+}
+
 impl Default for CodexFeedback {
     fn default() -> Self {
         Self::new()
@@ -489,6 +501,16 @@ pub struct FeedbackUploadOptions<'a> {
 }
 
 impl FeedbackSnapshot {
+    /// Refreshes log bytes while preserving the captured metadata and thread identity.
+    pub fn refresh_logs(&mut self, feedback: &CodexFeedback) {
+        self.bytes = feedback
+            .inner
+            .ring
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .snapshot_bytes();
+    }
+
     fn feedback_event(
         &self,
         classification: &str,

codex-rs/feedback/src/metadata_tests.rs

@@ -8,6 +8,32 @@ use sentry::protocol::EnvelopeItem;
 use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::util::SubscriberInitExt;
 
+#[test]
+fn refreshing_logs_preserves_captured_metadata() {
+    let feedback = CodexFeedback::new();
+    let _guard = tracing_subscriber::registry()
+        .with(feedback.metadata_layer())
+        .set_default();
+    let thread_id = ThreadId::new();
+    tracing::info!(target: FEEDBACK_TAGS_TARGET, approval_policy = "on-request");
+    feedback.report_failure("initial log\n");
+    let mut snapshot = feedback.snapshot(Some(thread_id));
+
+    // Another conversation changes the process-wide tags while log collection runs.
+    tracing::info!(target: FEEDBACK_TAGS_TARGET, approval_policy = "never");
+    feedback.report_failure("flush failed\n");
+    snapshot.refresh_logs(&feedback);
+
+    assert_eq!(
+        (snapshot.tags, snapshot.thread_id, snapshot.bytes),
+        (
+            BTreeMap::from([("approval_policy".to_string(), "on-request".to_string())]),
+            thread_id.to_string(),
+            b"initial log\nflush failed\n".to_vec(),
+        )
+    );
+}
+
 #[test]
 fn dynamic_tags_survive_upload_alongside_static_fields() {
     let feedback = CodexFeedback::new();

codex-rs/state/src/lib.rs

@@ -19,6 +19,7 @@ mod runtime;
 mod sqlite;
 mod telemetry;
 
+pub use log_db::LogWriteFailureReporter;
 pub use model::CreatedProject;
 pub use model::LogEntry;
 pub use model::LogQuery;

codex-rs/state/src/log_db.rs

@@ -4,8 +4,8 @@
 //! formats each one into a `LogEntry`, and sends entries to a bounded background
 //! queue. The background task inserts into the dedicated `logs` SQLite database
 //! in batches to keep logging overhead low.
-//! SQLx diagnostics are always excluded so database writes cannot generate more
-//! database writes, even without an external subscriber filter.
+//! SQLx events are excluded so database writes cannot generate more writes.
+//! Flush failures go to the supplied failure reporter without re-entering tracing.
 //!
 //! ## Usage
 //!
@@ -13,8 +13,8 @@
 //! use codex_state::log_db;
 //! use tracing_subscriber::prelude::*;
 //!
-//! # async fn example(state_db: std::sync::Arc<codex_state::StateRuntime>) {
-//! let layer = log_db::start(state_db);
+//! # async fn example(state_db: std::sync::Arc<codex_state::StateRuntime>, reporter: std::sync::Arc<dyn codex_state::LogWriteFailureReporter>) {
+//! let layer = log_db::start(state_db, reporter);
 //! let _ = tracing_subscriber::registry()
 //!     .with(layer)
 //!     .try_init();
@@ -22,7 +22,10 @@
 //! ```
 
 use std::future::Future;
+use std::sync::Arc;
 use std::sync::OnceLock;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
 use std::time::Duration;
 use std::time::Instant;
 use std::time::SystemTime;
@@ -124,42 +127,71 @@ where
     fn flush(&self) -> impl Future<Output = ()> + Send + '_;
 }
 
+/// Receives a formatted, redacted diagnostic when a SQLite log batch is lost
+/// this propagatest to telemetry and feedback memory buffer to avoid lost reports
+pub trait LogWriteFailureReporter: Send + Sync {
+    fn report_failure(&self, diagnostic: &str);
+}
+
 pub struct LogDbLayer {
     sender: mpsc::Sender<LogDbCommand>,
+    has_write_failure: Arc<AtomicBool>,
     process_uuid: String,
 }
 
-pub fn start(state_db: std::sync::Arc<StateRuntime>) -> LogDbLayer {
-    LogDbLayer::start(state_db)
+/// Starts the SQLite log writer with a shared, independent failure reporter.
+pub fn start(
+    state_db: std::sync::Arc<StateRuntime>,
+    failure_reporter: Arc<dyn LogWriteFailureReporter>,
+) -> LogDbLayer {
+    LogDbLayer::start(state_db, failure_reporter)
 }
 
 impl Clone for LogDbLayer {
     fn clone(&self) -> Self {
         Self {
             sender: self.sender.clone(),
+            has_write_failure: self.has_write_failure.clone(),
             process_uuid: self.process_uuid.clone(),
         }
     }
 }
 
 impl LogDbLayer {
-    pub fn start(state_db: std::sync::Arc<StateRuntime>) -> Self {
-        Self::start_with_config(state_db, LogSinkQueueConfig::default())
+    pub fn start(
+        state_db: std::sync::Arc<StateRuntime>,
+        failure_reporter: Arc<dyn LogWriteFailureReporter>,
+    ) -> Self {
+        Self::start_with_config(state_db, failure_reporter, LogSinkQueueConfig::default())
     }
 
     pub fn start_with_config(
         state_db: std::sync::Arc<StateRuntime>,
+        failure_reporter: Arc<dyn LogWriteFailureReporter>,
         config: LogSinkQueueConfig,
     ) -> Self {
         let config = config.normalized();
         let (sender, receiver) = mpsc::channel(config.queue_capacity);
-        tokio::spawn(run_inserter(state_db, receiver, config));
+        let has_write_failure = Arc::new(AtomicBool::new(false));
+        tokio::spawn(run_inserter(
+            state_db,
+            receiver,
+            config,
+            failure_reporter,
+            has_write_failure.clone(),
+        ));
         Self {
             sender,
+            has_write_failure,
             process_uuid: current_process_log_uuid().to_string(),
         }
     }
 
+    /// 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)
+    }
+
     pub async fn flush(&self) {
         let (tx, rx) = oneshot::channel();
         if self.sender.send(LogDbCommand::Flush(tx)).await.is_ok() {
@@ -447,6 +479,8 @@ async fn run_inserter(
     state_db: std::sync::Arc<StateRuntime>,
     mut receiver: mpsc::Receiver<LogDbCommand>,
     config: LogSinkQueueConfig,
+    failure_reporter: Arc<dyn LogWriteFailureReporter>,
+    has_write_failure: Arc<AtomicBool>,
 ) {
     let mut buffer = Vec::with_capacity(config.batch_size);
     let mut ticker = tokio::time::interval(config.flush_interval);
@@ -459,36 +493,56 @@ async fn run_inserter(
                     Some(LogDbCommand::Entry(entry)) => {
                         buffer.push(*entry);
                         if buffer.len() >= config.batch_size {
-                            flush(&state_db, &mut buffer).await;
+                            flush(&state_db, &mut buffer, failure_reporter.as_ref(), &has_write_failure).await;
                         }
                     }
                     Some(LogDbCommand::Flush(reply)) => {
-                        flush(&state_db, &mut buffer).await;
+                        flush(&state_db, &mut buffer, failure_reporter.as_ref(), &has_write_failure).await;
                         let _ = reply.send(());
                     }
                     None => {
-                        flush(&state_db, &mut buffer).await;
+                        flush(&state_db, &mut buffer, failure_reporter.as_ref(), &has_write_failure).await;
                         break;
                     }
                 }
             }
             _ = ticker.tick() => {
-                flush(&state_db, &mut buffer).await;
+                flush(&state_db, &mut buffer, failure_reporter.as_ref(), &has_write_failure).await;
             }
         }
     }
 }
 
-async fn flush(state_db: &StateRuntime, buffer: &mut Vec<LogEntry>) {
+async fn flush(
+    state_db: &StateRuntime,
+    buffer: &mut Vec<LogEntry>,
+    failure_reporter: &dyn LogWriteFailureReporter,
+    has_write_failure: &AtomicBool,
+) {
     if buffer.is_empty() {
         return;
     }
     let entries = buffer.split_off(0);
     let started = Instant::now();
     let result = state_db.insert_logs(entries.as_slice()).await;
+    let duration = started.elapsed();
+    // if log flushing failed it means that likely something is wrong with the logs database,
+    // so store the logs in the memory buffer to avoid them being lost during /feedback
+    // the entries are dropped because the in memory buffer has it's own space-constrainted rules for preserving logs
+    if let Err(error) = &result {
+        let error = crate::telemetry::classify_error(error);
+        let timestamp =
+            chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, /*use_z*/ true);
+        let entry_count = entries.len();
+        failure_reporter.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(
         /*telemetry*/ None,
-        started.elapsed(),
+        duration,
         entries.as_slice(),
         &result,
     );
@@ -545,11 +599,16 @@ impl Visit for MessageVisitor {
 #[path = "log_db_filter_tests.rs"]
 mod filter_tests;
 
+#[cfg(test)]
+#[path = "log_db_failure_tests.rs"]
+mod failure_tests;
+
 #[cfg(test)]
 mod tests {
     use std::io;
     use std::sync::Arc;
     use std::sync::Mutex;
+    use std::sync::atomic::AtomicBool;
 
     use codex_utils_absolute_path::test_support::PathExt;
     use pretty_assertions::assert_eq;
@@ -600,21 +659,28 @@ mod tests {
     }
 
     #[derive(Clone, Default)]
-    struct SharedWriter {
+    pub(super) struct SharedWriter {
         bytes: Arc<Mutex<Vec<u8>>>,
     }
 
     impl SharedWriter {
-        fn snapshot(&self) -> String {
+        pub(super) fn snapshot(&self) -> String {
             String::from_utf8(self.bytes.lock().expect("writer mutex poisoned").clone())
                 .expect("valid utf-8")
         }
     }
 
-    struct SharedWriterGuard {
+    pub(super) struct SharedWriterGuard {
         bytes: Arc<Mutex<Vec<u8>>>,
     }
 
+    impl LogWriteFailureReporter for SharedWriter {
+        fn report_failure(&self, diagnostic: &str) {
+            std::io::Write::write_all(&mut self.make_writer(), diagnostic.as_bytes())
+                .expect("write diagnostic to test buffer");
+        }
+    }
+
     impl<'a> MakeWriter<'a> for SharedWriter {
         type Writer = SharedWriterGuard;
 
@@ -649,7 +715,7 @@ mod tests {
         .await
         .expect("initialize runtime");
         let writer = SharedWriter::default();
-        let layer = start(runtime.clone());
+        let layer = start(runtime.clone(), Arc::new(SharedWriter::default()));
 
         let subscriber = tracing_subscriber::registry()
             .with(
@@ -709,7 +775,7 @@ mod tests {
         )
         .await
         .expect("initialize runtime");
-        let layer = start(runtime.clone());
+        let layer = start(runtime.clone(), Arc::new(SharedWriter::default()));
 
         let guard = tracing_subscriber::registry()
             .with(
@@ -745,6 +811,7 @@ mod tests {
         .expect("initialize runtime");
         let layer = LogDbLayer::start_with_config(
             runtime.clone(),
+            Arc::new(SharedWriter::default()),
             LogSinkQueueConfig {
                 queue_capacity: 8,
                 batch_size: 2,
@@ -797,6 +864,7 @@ mod tests {
         .expect("initialize runtime");
         let layer = LogDbLayer::start_with_config(
             runtime.clone(),
+            Arc::new(SharedWriter::default()),
             LogSinkQueueConfig {
                 queue_capacity: 8,
                 batch_size: 128,
@@ -827,6 +895,7 @@ mod tests {
         let (sender, mut receiver) = mpsc::channel(1);
         let layer = LogDbLayer {
             sender,
+            has_write_failure: Arc::new(AtomicBool::new(false)),
             process_uuid: "process-1".to_string(),
         };
 
@@ -847,6 +916,7 @@ mod tests {
         let (sender, mut receiver) = mpsc::channel(1);
         let layer = LogDbLayer {
             sender,
+            has_write_failure: Arc::new(AtomicBool::new(false)),
             process_uuid: "process-1".to_string(),
         };
 

codex-rs/state/src/log_db_failure_tests.rs

@@ -0,0 +1,86 @@
+//! Regression coverage for flush diagnostics, recovery, and sink recursion prevention.
+
+use codex_utils_absolute_path::test_support::PathExt;
+use pretty_assertions::assert_eq;
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
+
+use super::tests::SharedWriter;
+use super::*;
+
+#[tokio::test]
+async fn failed_batch_is_reported_once_and_later_writes_recover() {
+    let codex_home =
+        std::env::temp_dir().join(format!("codex-state-log-failure-{}", Uuid::new_v4()));
+    let _cleanup = scopeguard::guard(codex_home.clone(), |codex_home| {
+        let _ = std::fs::remove_dir_all(codex_home);
+    });
+    let sqlite = crate::SqliteConfig::new_for_testing(codex_home.as_path().abs());
+    let runtime = StateRuntime::init(sqlite.clone(), "test-provider".to_string())
+        .await
+        .expect("initialize runtime");
+    let pool = sqlite
+        .open_read_write_pool(&sqlite.logs_db_path())
+        .await
+        .expect("open logs database");
+    sqlx::query("CREATE TRIGGER fail_log_insert BEFORE INSERT ON logs BEGIN SELECT RAISE(ABORT, 'synthetic-secret'); END")
+        .execute(&pool)
+        .await
+        .expect("install failure trigger");
+    let diagnostics = SharedWriter::default();
+    let layer = start(runtime.clone(), Arc::new(diagnostics.clone()));
+    assert_eq!(layer.has_write_failure(), false);
+    let writer = SharedWriter::default();
+    let guard = tracing_subscriber::registry()
+        .with(
+            tracing_subscriber::fmt::layer()
+                .with_writer(writer.clone())
+                .without_time()
+                .with_level(/*display_level*/ false)
+                .with_target(/*display_target*/ false)
+                .with_ansi(/*ansi*/ false)
+                .with_filter(LevelFilter::ERROR),
+        )
+        // No diagnostic filter is needed: the diagnostic writer bypasses tracing.
+        .with(layer.clone())
+        .set_default();
+
+    tracing::info!("sensitive log body");
+    layer.flush().await;
+    assert_eq!(layer.has_write_failure(), true);
+    let diagnostic = diagnostics.snapshot();
+    assert_eq!(
+        diagnostic.split_once(" ERROR ").unwrap().1,
+        "failed to flush logs to SQLite error=\"constraint\" entries=1\n"
+    );
+    assert_eq!(writer.snapshot(), "");
+
+    // Reporting the first failure must not enqueue a second failing batch.
+    layer.flush().await;
+    assert_eq!(diagnostics.snapshot(), diagnostic);
+    assert_eq!(writer.snapshot(), "");
+
+    sqlx::query("DROP TRIGGER fail_log_insert")
+        .execute(&pool)
+        .await
+        .expect("remove failure trigger");
+    tracing::info!("recovered log");
+    layer.flush().await;
+    assert_eq!(diagnostics.snapshot(), diagnostic);
+    assert_eq!(writer.snapshot(), "");
+    assert_eq!(layer.has_write_failure(), true);
+    drop(guard);
+
+    let logs = runtime
+        .query_logs(&crate::LogQuery::default())
+        .await
+        .expect("query logs");
+    assert_eq!(
+        logs.iter()
+            .map(|log| log.message.as_deref())
+            .collect::<Vec<_>>(),
+        vec![Some("recovered log")]
+    );
+    pool.close().await;
+    runtime.close().await;
+}

codex-rs/state/src/log_db_filter_tests.rs

@@ -19,7 +19,10 @@ async fn sqlite_sink_filters_noisy_targets_without_dropping_useful_diagnostics()
     )
     .await
     .expect("initialize runtime");
-    let layer = start(runtime.clone());
+    let layer = start(
+        runtime.clone(),
+        Arc::new(super::tests::SharedWriter::default()),
+    );
 
     let guard = tracing_subscriber::registry()
         .with(layer.clone().with_filter(default_filter()))

codex-rs/state/src/telemetry.rs

@@ -177,7 +177,7 @@ impl DbOutcomeTags {
     }
 }
 
-fn classify_error(err: &anyhow::Error) -> &'static str {
+pub(crate) fn classify_error(err: &anyhow::Error) -> &'static str {
     for cause in err.chain() {
         if let Some(sqlx_err) = cause.downcast_ref::<sqlx::Error>() {
             return classify_sqlx_error(sqlx_err);

codex-rs/tui/src/startup_orchestration.rs

@@ -828,7 +828,9 @@ pub(super) async fn run_main_inner(
 
     let otel_tracing_layer = otel.as_ref().and_then(|o| o.tracing_layer());
 
-    let log_db = state_db.clone().map(log_db::start);
+    let log_db = state_db
+        .clone()
+        .map(|state_db| log_db::start(state_db, std::sync::Arc::new(feedback.clone())));
     let log_db_layer = log_db
         .clone()
         .map(|layer| layer.with_filter(log_db::default_filter()));