app-server: inline sqlite feedback fallback

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

@@ -6719,7 +6719,22 @@ impl CodexMessageProcessor {
         let thread_id = snapshot.thread_id.clone();
         let sqlite_feedback_logs = if include_logs {
             let state_db_ctx = get_state_db(&self.config, None).await;
-            read_feedback_logs_from_state_db_context(state_db_ctx.as_ref(), conversation_id).await
+            match (state_db_ctx.as_ref(), conversation_id) {
+                (Some(state_db_ctx), Some(conversation_id)) => {
+                    let thread_id_text = conversation_id.to_string();
+                    match state_db_ctx.query_feedback_logs(&thread_id_text).await {
+                        Ok(logs) if logs.is_empty() => None,
+                        Ok(logs) => Some(logs),
+                        Err(err) => {
+                            warn!(
+                                "failed to query feedback logs from sqlite for thread_id={thread_id_text}: {err}"
+                            );
+                            None
+                        }
+                    }
+                }
+                _ => None,
+            }
         } else {
             None
         };
@@ -7307,26 +7322,6 @@ async fn read_summary_from_state_db_context_by_thread_id(
     ))
 }
 
-async fn read_feedback_logs_from_state_db_context(
-    state_db_ctx: Option<&StateDbHandle>,
-    thread_id: Option<ThreadId>,
-) -> Option<Vec<u8>> {
-    let state_db_ctx = state_db_ctx?;
-    let thread_id = thread_id?;
-    let thread_id_text = thread_id.to_string();
-
-    match state_db_ctx.query_feedback_logs(&thread_id_text).await {
-        Ok(logs) if logs.is_empty() => None,
-        Ok(logs) => Some(logs),
-        Err(err) => {
-            warn!(
-                "failed to query feedback logs from sqlite for thread_id={thread_id_text}: {err}"
-            );
-            None
-        }
-    }
-}
-
 async fn summary_from_thread_list_item(
     it: codex_core::ThreadItem,
     fallback_provider: &str,
@@ -7698,7 +7693,6 @@ mod tests {
     use anyhow::Result;
     use codex_protocol::protocol::SessionSource;
     use codex_protocol::protocol::SubAgentSource;
-    use codex_state::StateRuntime;
     use pretty_assertions::assert_eq;
     use serde_json::json;
     use std::path::PathBuf;
@@ -7925,23 +7919,6 @@ mod tests {
         Ok(())
     }
 
-    #[tokio::test]
-    async fn read_feedback_logs_from_state_db_context_returns_none_for_empty_logs() -> Result<()> {
-        let temp_dir = TempDir::new()?;
-        let state_db = StateRuntime::init(
-            temp_dir.path().to_path_buf(),
-            "test-provider".to_string(),
-            None,
-        )
-        .await?;
-        let thread_id = ThreadId::from_string("ad7f0408-99b8-4f6e-a46f-bd0eec433370")?;
-
-        let logs = read_feedback_logs_from_state_db_context(Some(&state_db), Some(thread_id)).await;
-
-        assert_eq!(logs, None);
-        Ok(())
-    }
-
     #[tokio::test]
     async fn removing_listeners_retains_thread_listener_when_last_subscriber_leaves() -> Result<()>
     {

codex-rs/state/src/log_db.rs

@@ -312,7 +312,7 @@ mod tests {
     use std::sync::Mutex;
     use std::time::Duration;
 
-    use pretty_assertions::assert_eq;
+    use tokio::time::Instant;
     use tracing_subscriber::filter::Targets;
     use tracing_subscriber::fmt::writer::MakeWriter;
     use tracing_subscriber::layer::SubscriberExt;
@@ -390,23 +390,31 @@ mod tests {
         });
         tracing::debug!("threadless-after");
 
-        tokio::time::sleep(Duration::from_millis(500)).await;
         drop(guard);
 
-        let sqlite_logs = String::from_utf8(
-            runtime
-                .query_feedback_logs("thread-1")
-                .await
-                .expect("query feedback logs"),
-        )
-        .expect("valid utf-8");
         // TODO(ccunningham): Store enough span metadata in SQLite to reproduce span
         // prefixes like `feedback-thread{thread_id="thread-1"}:` in feedback exports.
         let feedback_logs = writer
             .snapshot()
             .replace("feedback-thread{thread_id=\"thread-1\"}: ", "");
-
-        assert_eq!(sqlite_logs, feedback_logs);
+        let deadline = Instant::now() + Duration::from_secs(2);
+        loop {
+            let sqlite_logs = String::from_utf8(
+                runtime
+                    .query_feedback_logs("thread-1")
+                    .await
+                    .expect("query feedback logs"),
+            )
+            .expect("valid utf-8");
+            if sqlite_logs == feedback_logs {
+                break;
+            }
+            assert!(
+                Instant::now() < deadline,
+                "sqlite feedback logs did not match feedback formatter output before timeout\nsqlite:\n{sqlite_logs}\nfeedback:\n{feedback_logs}"
+            );
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
 
         let _ = tokio::fs::remove_dir_all(codex_home).await;
     }