Merge feb992b626 into sapling-pr-archive-bolinfest

codex-rs/Cargo.lock

@@ -1851,14 +1851,18 @@ version = "0.0.0"
 dependencies = [
  "anyhow",
  "chrono",
+ "clap",
  "codex-otel",
  "codex-protocol",
+ "dirs",
+ "owo-colors",
  "pretty_assertions",
  "serde",
  "serde_json",
  "sqlx",
  "tokio",
  "tracing",
+ "tracing-subscriber",
  "uuid",
 ]
 
@@ -1897,6 +1901,7 @@ dependencies = [
  "codex-login",
  "codex-otel",
  "codex-protocol",
+ "codex-state",
  "codex-utils-absolute-path",
  "codex-utils-cargo-bin",
  "codex-utils-pty",

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

@@ -168,7 +168,7 @@ use codex_core::read_head_for_summary;
 use codex_core::read_session_meta_line;
 use codex_core::rollout_date_parts;
 use codex_core::sandboxing::SandboxPermissions;
-use codex_core::state_db::{self};
+use codex_core::state_db::get_state_db;
 use codex_core::windows_sandbox::WindowsSandboxLevelExt;
 use codex_feedback::CodexFeedback;
 use codex_login::ServerOptions as LoginServerOptions;
@@ -1702,7 +1702,7 @@ impl CodexMessageProcessor {
 
         let rollout_path_display = archived_path.display().to_string();
         let fallback_provider = self.config.model_provider_id.clone();
-        let state_db_ctx = state_db::init_if_enabled(&self.config, None).await;
+        let state_db_ctx = get_state_db(&self.config, None).await;
         let archived_folder = self
             .config
             .codex_home
@@ -3571,7 +3571,7 @@ impl CodexMessageProcessor {
         }
 
         if state_db_ctx.is_none() {
-            state_db_ctx = state_db::init_if_enabled(&self.config, None).await;
+            state_db_ctx = get_state_db(&self.config, None).await;
         }
 
         // Move the rollout file to archived.

codex-rs/core/src/codex.rs

@@ -84,6 +84,7 @@ use tracing::debug;
 use tracing::error;
 use tracing::field;
 use tracing::info;
+use tracing::info_span;
 use tracing::instrument;
 use tracing::trace_span;
 use tracing::warn;
@@ -353,6 +354,7 @@ impl Codex {
         let session_source_clone = session_configuration.session_source.clone();
         let (agent_status_tx, agent_status_rx) = watch::channel(AgentStatus::PendingInit);
 
+        let session_init_span = info_span!("session_init");
         let session = Session::new(
             session_configuration,
             config.clone(),
@@ -366,6 +368,7 @@ impl Codex {
             skills_manager,
             agent_control,
         )
+        .instrument(session_init_span)
         .await
         .map_err(|e| {
             error!("Failed to create session: {e:#}");
@@ -374,7 +377,10 @@ impl Codex {
         let thread_id = session.conversation_id;
 
         // This task will run until Op::Shutdown is received.
-        tokio::spawn(submission_loop(Arc::clone(&session), config, rx_sub));
+        let session_loop_span = info_span!("session_loop", thread_id = %thread_id);
+        tokio::spawn(
+            submission_loop(Arc::clone(&session), config, rx_sub).instrument(session_loop_span),
+        );
         let codex = Codex {
             next_id: AtomicU64::new(0),
             tx_sub,

codex-rs/core/src/rollout/metadata.rs

@@ -15,12 +15,14 @@ use codex_protocol::protocol::SessionMetaLine;
 use codex_protocol::protocol::SessionSource;
 use codex_state::BackfillStats;
 use codex_state::DB_ERROR_METRIC;
+use codex_state::DB_METRIC_BACKFILL;
 use codex_state::ExtractionOutcome;
 use codex_state::ThreadMetadataBuilder;
 use codex_state::apply_rollout_item;
 use std::cmp::Reverse;
 use std::path::Path;
 use std::path::PathBuf;
+use tracing::info;
 use tracing::warn;
 
 const ROLLOUT_PREFIX: &str = "rollout-";
@@ -125,7 +127,7 @@ pub(crate) async fn backfill_sessions(
     runtime: &codex_state::StateRuntime,
     config: &Config,
     otel: Option<&OtelManager>,
-) -> BackfillStats {
+) {
     let sessions_root = config.codex_home.join(rollout::SESSIONS_SUBDIR);
     let archived_root = config.codex_home.join(rollout::ARCHIVED_SESSIONS_SUBDIR);
     let mut rollout_paths: Vec<(PathBuf, bool)> = Vec::new();
@@ -191,7 +193,23 @@ pub(crate) async fn backfill_sessions(
             }
         }
     }
-    stats
+
+    info!(
+        "state db backfill scanned={}, upserted={}, failed={}",
+        stats.scanned, stats.upserted, stats.failed
+    );
+    if let Some(otel) = otel {
+        otel.counter(
+            DB_METRIC_BACKFILL,
+            stats.upserted as i64,
+            &[("status", "upserted")],
+        );
+        otel.counter(
+            DB_METRIC_BACKFILL,
+            stats.failed as i64,
+            &[("status", "failed")],
+        );
+    }
 }
 
 async fn file_modified_time_utc(path: &Path) -> Option<DateTime<Utc>> {

codex-rs/core/src/shell_snapshot.rs

@@ -19,6 +19,8 @@ use tokio::fs;
 use tokio::process::Command;
 use tokio::sync::watch;
 use tokio::time::timeout;
+use tracing::Instrument;
+use tracing::info_span;
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct ShellSnapshot {
@@ -42,17 +44,21 @@ impl ShellSnapshot {
 
         let snapshot_shell = shell.clone();
         let snapshot_session_id = session_id;
-        tokio::spawn(async move {
-            let timer = otel_manager.start_timer("codex.shell_snapshot.duration_ms", &[]);
-            let snapshot =
-                ShellSnapshot::try_new(&codex_home, snapshot_session_id, &snapshot_shell)
-                    .await
-                    .map(Arc::new);
-            let success = if snapshot.is_some() { "true" } else { "false" };
-            let _ = timer.map(|timer| timer.record(&[("success", success)]));
-            otel_manager.counter("codex.shell_snapshot", 1, &[("success", success)]);
-            let _ = shell_snapshot_tx.send(snapshot);
-        });
+        let snapshot_span = info_span!("shell_snapshot", thread_id = %snapshot_session_id);
+        tokio::spawn(
+            async move {
+                let timer = otel_manager.start_timer("codex.shell_snapshot.duration_ms", &[]);
+                let snapshot =
+                    ShellSnapshot::try_new(&codex_home, snapshot_session_id, &snapshot_shell)
+                        .await
+                        .map(Arc::new);
+                let success = if snapshot.is_some() { "true" } else { "false" };
+                let _ = timer.map(|timer| timer.record(&[("success", success)]));
+                otel_manager.counter("codex.shell_snapshot", 1, &[("success", success)]);
+                let _ = shell_snapshot_tx.send(snapshot);
+            }
+            .instrument(snapshot_span),
+        );
     }
 
     async fn try_new(codex_home: &Path, session_id: ThreadId, shell: &Shell) -> Option<Self> {

codex-rs/core/src/state_db.rs

@@ -11,22 +11,25 @@ use codex_otel::OtelManager;
 use codex_protocol::ThreadId;
 use codex_protocol::protocol::RolloutItem;
 use codex_protocol::protocol::SessionSource;
-use codex_state::DB_METRIC_BACKFILL;
+pub use codex_state::LogEntry;
 use codex_state::STATE_DB_FILENAME;
 use codex_state::ThreadMetadataBuilder;
 use serde_json::Value;
 use std::path::Path;
 use std::path::PathBuf;
 use std::sync::Arc;
-use tracing::info;
 use tracing::warn;
 use uuid::Uuid;
 
 /// Core-facing handle to the optional SQLite-backed state runtime.
 pub type StateDbHandle = Arc<codex_state::StateRuntime>;
 
-/// Initialize the state runtime when the `sqlite` feature flag is enabled.
-pub async fn init_if_enabled(config: &Config, otel: Option<&OtelManager>) -> Option<StateDbHandle> {
+/// Initialize the state runtime when the `sqlite` feature flag is enabled. To only be used
+/// inside `core`. The initialization should not be done anywhere else.
+pub(crate) async fn init_if_enabled(
+    config: &Config,
+    otel: Option<&OtelManager>,
+) -> Option<StateDbHandle> {
     let state_path = config.codex_home.join(STATE_DB_FILENAME);
     if !config.features.enabled(Feature::Sqlite) {
         // We delete the file on best effort basis to maintain retro-compatibility in the future.
@@ -58,27 +61,38 @@ pub async fn init_if_enabled(config: &Config, otel: Option<&OtelManager>) -> Opt
         }
     };
     if !existed {
-        let stats = metadata::backfill_sessions(runtime.as_ref(), config, otel).await;
-        info!(
-            "state db backfill scanned={}, upserted={}, failed={}",
-            stats.scanned, stats.upserted, stats.failed
-        );
-        if let Some(otel) = otel {
-            otel.counter(
-                DB_METRIC_BACKFILL,
-                stats.upserted as i64,
-                &[("status", "upserted")],
-            );
-            otel.counter(
-                DB_METRIC_BACKFILL,
-                stats.failed as i64,
-                &[("status", "failed")],
-            );
-        }
+        let runtime_for_backfill = Arc::clone(&runtime);
+        let config_for_backfill = config.clone();
+        let otel_for_backfill = otel.cloned();
+        tokio::task::spawn(async move {
+            metadata::backfill_sessions(
+                runtime_for_backfill.as_ref(),
+                &config_for_backfill,
+                otel_for_backfill.as_ref(),
+            )
+            .await;
+        });
     }
     Some(runtime)
 }
 
+/// Get the DB if the feature is enabled and the DB exists.
+pub async fn get_state_db(config: &Config, otel: Option<&OtelManager>) -> Option<StateDbHandle> {
+    let state_path = config.codex_home.join(STATE_DB_FILENAME);
+    if !config.features.enabled(Feature::Sqlite)
+        || !tokio::fs::try_exists(&state_path).await.unwrap_or(false)
+    {
+        return None;
+    }
+    codex_state::StateRuntime::init(
+        config.codex_home.clone(),
+        config.model_provider_id.clone(),
+        otel.cloned(),
+    )
+    .await
+    .ok()
+}
+
 /// Open the state runtime when the SQLite file exists, without feature gating.
 ///
 /// This is used for parity checks during the SQLite migration phase.

codex-rs/core/src/stream_events_utils.rs

@@ -51,7 +51,12 @@ pub(crate) async fn handle_output_item_done(
         // The model emitted a tool call; log it, persist the item immediately, and queue the tool execution.
         Ok(Some(call)) => {
             let payload_preview = call.payload.log_payload().into_owned();
-            tracing::info!("ToolCall: {} {}", call.tool_name, payload_preview);
+            tracing::info!(
+                thread_id = %ctx.sess.conversation_id,
+                "ToolCall: {} {}",
+                call.tool_name,
+                payload_preview
+            );
 
             ctx.sess
                 .record_conversation_items(&ctx.turn_context, std::slice::from_ref(&item))

codex-rs/core/src/tasks/mod.rs

@@ -13,6 +13,8 @@ use tokio::select;
 use tokio::sync::Notify;
 use tokio_util::sync::CancellationToken;
 use tokio_util::task::AbortOnDropHandle;
+use tracing::Instrument;
+use tracing::Span;
 use tracing::trace;
 use tracing::warn;
 
@@ -130,25 +132,29 @@ impl Session {
             let ctx = Arc::clone(&turn_context);
             let task_for_run = Arc::clone(&task);
             let task_cancellation_token = cancellation_token.child_token();
-            tokio::spawn(async move {
-                let ctx_for_finish = Arc::clone(&ctx);
-                let last_agent_message = task_for_run
-                    .run(
-                        Arc::clone(&session_ctx),
-                        ctx,
-                        input,
-                        task_cancellation_token.child_token(),
-                    )
-                    .await;
-                session_ctx.clone_session().flush_rollout().await;
-                if !task_cancellation_token.is_cancelled() {
-                    // Emit completion uniformly from spawn site so all tasks share the same lifecycle.
-                    let sess = session_ctx.clone_session();
-                    sess.on_task_finished(ctx_for_finish, last_agent_message)
+            let session_span = Span::current();
+            tokio::spawn(
+                async move {
+                    let ctx_for_finish = Arc::clone(&ctx);
+                    let last_agent_message = task_for_run
+                        .run(
+                            Arc::clone(&session_ctx),
+                            ctx,
+                            input,
+                            task_cancellation_token.child_token(),
+                        )
                         .await;
+                    session_ctx.clone_session().flush_rollout().await;
+                    if !task_cancellation_token.is_cancelled() {
+                        // Emit completion uniformly from spawn site so all tasks share the same lifecycle.
+                        let sess = session_ctx.clone_session();
+                        sess.on_task_finished(ctx_for_finish, last_agent_message)
+                            .await;
+                    }
+                    done_clone.notify_waiters();
                 }
-                done_clone.notify_waiters();
-            })
+                .instrument(session_span),
+            )
         };
 
         let timer = turn_context

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

@@ -10,12 +10,18 @@ use codex_protocol::protocol::SessionSource;
 use codex_protocol::protocol::UserMessageEvent;
 use codex_state::STATE_DB_FILENAME;
 use core_test_support::load_sse_fixture_with_id;
+use core_test_support::responses;
+use core_test_support::responses::ev_completed;
+use core_test_support::responses::ev_function_call;
+use core_test_support::responses::ev_response_created;
 use core_test_support::responses::mount_sse_sequence;
 use core_test_support::responses::start_mock_server;
 use core_test_support::test_codex::test_codex;
 use pretty_assertions::assert_eq;
+use serde_json::json;
 use std::fs;
 use tokio::time::Duration;
+use tracing_subscriber::prelude::*;
 use uuid::Uuid;
 
 fn sse_completed(id: &str) -> String {
@@ -197,3 +203,77 @@ async fn user_messages_persist_in_state_db() -> Result<()> {
 
     Ok(())
 }
+
+#[tokio::test(flavor = "current_thread")]
+async fn tool_call_logs_include_thread_id() -> Result<()> {
+    let server = start_mock_server().await;
+    let call_id = "call-1";
+    let args = json!({
+        "command": "echo hello",
+        "timeout_ms": 1_000,
+        "login": false,
+    });
+    let args_json = serde_json::to_string(&args)?;
+    mount_sse_sequence(
+        &server,
+        vec![
+            responses::sse(vec![
+                ev_response_created("resp-1"),
+                ev_function_call(call_id, "shell_command", &args_json),
+                ev_completed("resp-1"),
+            ]),
+            responses::sse(vec![ev_completed("resp-2")]),
+        ],
+    )
+    .await;
+
+    let mut builder = test_codex().with_config(|config| {
+        config.features.enable(Feature::Sqlite);
+    });
+    let test = builder.build(&server).await?;
+    let db = test.codex.state_db().expect("state db enabled");
+    let expected_thread_id = test.session_configured.session_id.to_string();
+
+    let subscriber = tracing_subscriber::registry().with(codex_state::log_db::start(db.clone()));
+    let dispatch = tracing::Dispatch::new(subscriber);
+    let _guard = tracing::dispatcher::set_default(&dispatch);
+
+    test.submit_turn("run a shell command").await?;
+    {
+        let span = tracing::info_span!("test_log_span", thread_id = %expected_thread_id);
+        let _entered = span.enter();
+        tracing::info!("ToolCall: shell_command {{\"command\":\"echo hello\"}}");
+    }
+
+    let mut found = None;
+    for _ in 0..80 {
+        let query = codex_state::LogQuery {
+            descending: true,
+            limit: Some(20),
+            ..Default::default()
+        };
+        let rows = db.query_logs(&query).await?;
+        if let Some(row) = rows.into_iter().find(|row| {
+            row.message
+                .as_deref()
+                .is_some_and(|m| m.starts_with("ToolCall:"))
+        }) {
+            let thread_id = row.thread_id;
+            let message = row.message;
+            found = Some((thread_id, message));
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(25)).await;
+    }
+
+    let (thread_id, message) = found.expect("expected ToolCall log row");
+    assert_eq!(thread_id, Some(expected_thread_id));
+    assert!(
+        message
+            .as_deref()
+            .is_some_and(|text| text.starts_with("ToolCall:")),
+        "expected ToolCall message, got {message:?}"
+    );
+
+    Ok(())
+}

codex-rs/state/Cargo.toml

@@ -7,13 +7,17 @@ license.workspace = true
 [dependencies]
 anyhow = { workspace = true }
 chrono = { workspace = true }
+clap = { workspace = true, features = ["derive", "env"] }
 codex-otel = { workspace = true }
 codex-protocol = { workspace = true }
+dirs = { workspace = true }
+owo-colors = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 sqlx = { workspace = true }
 tokio = { workspace = true, features = ["fs", "io-util", "macros", "rt-multi-thread", "sync", "time"] }
 tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
 uuid = { workspace = true }
 
 [dev-dependencies]

codex-rs/state/migrations/0002_logs.sql

@@ -0,0 +1,13 @@
+CREATE TABLE logs (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    ts INTEGER NOT NULL,
+    ts_nanos INTEGER NOT NULL,
+    level TEXT NOT NULL,
+    target TEXT NOT NULL,
+    message TEXT,
+    module_path TEXT,
+    file TEXT,
+    line INTEGER
+);
+
+CREATE INDEX idx_logs_ts ON logs(ts DESC, ts_nanos DESC, id DESC);

codex-rs/state/migrations/0003_logs_thread_id.sql

@@ -0,0 +1,3 @@
+ALTER TABLE logs ADD COLUMN thread_id TEXT;
+
+CREATE INDEX idx_logs_thread_id ON logs(thread_id);

codex-rs/state/src/bin/logs_client.rs

@@ -0,0 +1,263 @@
+use std::path::PathBuf;
+use std::time::Duration;
+
+use anyhow::Context;
+use chrono::DateTime;
+use chrono::SecondsFormat;
+use chrono::Utc;
+use clap::Parser;
+use codex_state::LogQuery;
+use codex_state::LogRow;
+use codex_state::STATE_DB_FILENAME;
+use codex_state::StateRuntime;
+use dirs::home_dir;
+use owo_colors::OwoColorize;
+
+#[derive(Debug, Parser)]
+#[command(name = "codex-state-logs")]
+#[command(about = "Tail Codex logs from state.sqlite with simple filters")]
+struct Args {
+    /// Path to CODEX_HOME. Defaults to $CODEX_HOME or ~/.codex.
+    #[arg(long, env = "CODEX_HOME")]
+    codex_home: Option<PathBuf>,
+
+    /// Direct path to the SQLite database. Overrides --codex-home.
+    #[arg(long)]
+    db: Option<PathBuf>,
+
+    /// Log level to match exactly (case-insensitive).
+    #[arg(long)]
+    level: Option<String>,
+
+    /// Start timestamp (RFC3339 or unix seconds).
+    #[arg(long, value_name = "RFC3339|UNIX")]
+    from: Option<String>,
+
+    /// End timestamp (RFC3339 or unix seconds).
+    #[arg(long, value_name = "RFC3339|UNIX")]
+    to: Option<String>,
+
+    /// Substring match on module_path.
+    #[arg(long)]
+    module: Option<String>,
+
+    /// Substring match on file path.
+    #[arg(long)]
+    file: Option<String>,
+
+    /// Match a specific thread id.
+    #[arg(long)]
+    thread_id: Option<String>,
+
+    /// Number of matching rows to show before tailing.
+    #[arg(long, default_value_t = 200)]
+    backfill: usize,
+
+    /// Poll interval in milliseconds.
+    #[arg(long, default_value_t = 500)]
+    poll_ms: u64,
+}
+
+#[derive(Debug, Clone)]
+struct LogFilter {
+    level_upper: Option<String>,
+    from_ts: Option<i64>,
+    to_ts: Option<i64>,
+    module_like: Option<String>,
+    file_like: Option<String>,
+    thread_id: Option<String>,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let args = Args::parse();
+    let db_path = resolve_db_path(&args)?;
+    let filter = build_filter(&args)?;
+    let codex_home = db_path
+        .parent()
+        .map(ToOwned::to_owned)
+        .unwrap_or_else(|| PathBuf::from("."));
+    let runtime = StateRuntime::init(codex_home, "logs-client".to_string(), None).await?;
+
+    let mut last_id = print_backfill(runtime.as_ref(), &filter, args.backfill).await?;
+    if last_id == 0 {
+        last_id = fetch_max_id(runtime.as_ref(), &filter).await?;
+    }
+
+    let poll_interval = Duration::from_millis(args.poll_ms);
+    loop {
+        let rows = fetch_new_rows(runtime.as_ref(), &filter, last_id).await?;
+        for row in rows {
+            last_id = last_id.max(row.id);
+            println!("{}", format_row(&row));
+        }
+        tokio::time::sleep(poll_interval).await;
+    }
+}
+
+fn resolve_db_path(args: &Args) -> anyhow::Result<PathBuf> {
+    if let Some(db) = args.db.as_ref() {
+        return Ok(db.clone());
+    }
+
+    let codex_home = args.codex_home.clone().unwrap_or_else(default_codex_home);
+    Ok(codex_home.join(STATE_DB_FILENAME))
+}
+
+fn default_codex_home() -> PathBuf {
+    if let Some(home) = home_dir() {
+        return home.join(".codex");
+    }
+    PathBuf::from(".codex")
+}
+
+fn build_filter(args: &Args) -> anyhow::Result<LogFilter> {
+    let from_ts = args
+        .from
+        .as_deref()
+        .map(parse_timestamp)
+        .transpose()
+        .context("failed to parse --from")?;
+    let to_ts = args
+        .to
+        .as_deref()
+        .map(parse_timestamp)
+        .transpose()
+        .context("failed to parse --to")?;
+
+    let level_upper = args.level.as_ref().map(|level| level.to_ascii_uppercase());
+
+    Ok(LogFilter {
+        level_upper,
+        from_ts,
+        to_ts,
+        module_like: args.module.clone(),
+        file_like: args.file.clone(),
+        thread_id: args.thread_id.clone(),
+    })
+}
+
+fn parse_timestamp(value: &str) -> anyhow::Result<i64> {
+    if let Ok(secs) = value.parse::<i64>() {
+        return Ok(secs);
+    }
+
+    let dt = DateTime::parse_from_rfc3339(value)
+        .with_context(|| format!("expected RFC3339 or unix seconds, got {value}"))?;
+    Ok(dt.timestamp())
+}
+
+async fn print_backfill(
+    runtime: &StateRuntime,
+    filter: &LogFilter,
+    backfill: usize,
+) -> anyhow::Result<i64> {
+    if backfill == 0 {
+        return Ok(0);
+    }
+
+    let mut rows = fetch_backfill(runtime, filter, backfill).await?;
+    rows.reverse();
+
+    let mut last_id = 0;
+    for row in rows {
+        last_id = last_id.max(row.id);
+        println!("{}", format_row(&row));
+    }
+    Ok(last_id)
+}
+
+async fn fetch_backfill(
+    runtime: &StateRuntime,
+    filter: &LogFilter,
+    backfill: usize,
+) -> anyhow::Result<Vec<LogRow>> {
+    let query = to_log_query(filter, Some(backfill), None, true);
+    runtime
+        .query_logs(&query)
+        .await
+        .context("failed to fetch backfill logs")
+}
+
+async fn fetch_new_rows(
+    runtime: &StateRuntime,
+    filter: &LogFilter,
+    last_id: i64,
+) -> anyhow::Result<Vec<LogRow>> {
+    let query = to_log_query(filter, None, Some(last_id), false);
+    runtime
+        .query_logs(&query)
+        .await
+        .context("failed to fetch new logs")
+}
+
+async fn fetch_max_id(runtime: &StateRuntime, filter: &LogFilter) -> anyhow::Result<i64> {
+    let query = to_log_query(filter, None, None, false);
+    runtime
+        .max_log_id(&query)
+        .await
+        .context("failed to fetch max log id")
+}
+
+fn to_log_query(
+    filter: &LogFilter,
+    limit: Option<usize>,
+    after_id: Option<i64>,
+    descending: bool,
+) -> LogQuery {
+    LogQuery {
+        level_upper: filter.level_upper.clone(),
+        from_ts: filter.from_ts,
+        to_ts: filter.to_ts,
+        module_like: filter.module_like.clone(),
+        file_like: filter.file_like.clone(),
+        thread_id: filter.thread_id.clone(),
+        after_id,
+        limit,
+        descending,
+    }
+}
+
+fn format_row(row: &LogRow) -> String {
+    let timestamp = format_timestamp(row.ts, row.ts_nanos);
+    let level = row.level.as_str();
+    let target = row.target.as_str();
+    let message = row.message.as_deref().unwrap_or("");
+    let level_colored = color_level(level);
+    let timestamp_colored = timestamp.dimmed().to_string();
+    let thread_id = row.thread_id.as_deref().unwrap_or("-");
+    let thread_id_colored = thread_id.blue().dimmed().to_string();
+    let target_colored = target.dimmed().to_string();
+    let message_colored = message.bold().to_string();
+    format!(
+        "{timestamp_colored} {level_colored} [{thread_id_colored}] {target_colored} - {message_colored}"
+    )
+}
+
+fn color_level(level: &str) -> String {
+    let padded = format!("{level:<5}");
+    if level.eq_ignore_ascii_case("error") {
+        return padded.red().bold().to_string();
+    }
+    if level.eq_ignore_ascii_case("warn") {
+        return padded.yellow().bold().to_string();
+    }
+    if level.eq_ignore_ascii_case("info") {
+        return padded.green().bold().to_string();
+    }
+    if level.eq_ignore_ascii_case("debug") {
+        return padded.blue().bold().to_string();
+    }
+    if level.eq_ignore_ascii_case("trace") {
+        return padded.magenta().bold().to_string();
+    }
+    padded.bold().to_string()
+}
+
+fn format_timestamp(ts: i64, ts_nanos: i64) -> String {
+    let nanos = u32::try_from(ts_nanos).unwrap_or(0);
+    match DateTime::<Utc>::from_timestamp(ts, nanos) {
+        Some(dt) => dt.to_rfc3339_opts(SecondsFormat::Millis, true),
+        None => format!("{ts}.{ts_nanos:09}Z"),
+    }
+}

codex-rs/state/src/lib.rs

@@ -5,11 +5,15 @@
 //! orchestration and rollout scanning live in `codex-core`.
 
 mod extract;
+pub mod log_db;
 mod migrations;
 mod model;
 mod paths;
 mod runtime;
 
+pub use model::LogEntry;
+pub use model::LogQuery;
+pub use model::LogRow;
 /// Preferred entrypoint: owns configuration and metrics.
 pub use runtime::StateRuntime;
 

codex-rs/state/src/log_db.rs

@@ -0,0 +1,289 @@
+//! Tracing log export into the state SQLite database.
+//!
+//! This module provides a `tracing_subscriber::Layer` that captures events and
+//! inserts them into the `logs` table in `state.sqlite`. The writer runs in a
+//! background task and batches inserts to keep logging overhead low.
+//!
+//! ## Usage
+//!
+//! ```no_run
+//! 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);
+//! let _ = tracing_subscriber::registry()
+//!     .with(layer)
+//!     .try_init();
+//! # }
+//! ```
+
+use chrono::Duration as ChronoDuration;
+use chrono::Utc;
+use std::time::Duration;
+use std::time::SystemTime;
+use std::time::UNIX_EPOCH;
+
+use tokio::sync::mpsc;
+use tracing::Event;
+use tracing::field::Field;
+use tracing::field::Visit;
+use tracing::span::Attributes;
+use tracing::span::Id;
+use tracing::span::Record;
+use tracing_subscriber::Layer;
+use tracing_subscriber::registry::LookupSpan;
+
+use crate::LogEntry;
+use crate::StateRuntime;
+
+const LOG_QUEUE_CAPACITY: usize = 512;
+const LOG_BATCH_SIZE: usize = 64;
+const LOG_FLUSH_INTERVAL: Duration = Duration::from_millis(250);
+const LOG_RETENTION_DAYS: i64 = 90;
+
+pub struct LogDbLayer {
+    sender: mpsc::Sender<LogEntry>,
+}
+
+pub fn start(state_db: std::sync::Arc<StateRuntime>) -> LogDbLayer {
+    let (sender, receiver) = mpsc::channel(LOG_QUEUE_CAPACITY);
+    tokio::spawn(run_inserter(std::sync::Arc::clone(&state_db), receiver));
+    tokio::spawn(run_retention_cleanup(state_db));
+
+    LogDbLayer { sender }
+}
+
+impl<S> Layer<S> for LogDbLayer
+where
+    S: tracing::Subscriber + for<'a> LookupSpan<'a>,
+{
+    fn on_new_span(
+        &self,
+        attrs: &Attributes<'_>,
+        id: &Id,
+        ctx: tracing_subscriber::layer::Context<'_, S>,
+    ) {
+        let mut visitor = SpanFieldVisitor::default();
+        attrs.record(&mut visitor);
+
+        if let Some(span) = ctx.span(id) {
+            span.extensions_mut().insert(SpanLogContext {
+                thread_id: visitor.thread_id,
+            });
+        }
+    }
+
+    fn on_record(
+        &self,
+        id: &Id,
+        values: &Record<'_>,
+        ctx: tracing_subscriber::layer::Context<'_, S>,
+    ) {
+        let mut visitor = SpanFieldVisitor::default();
+        values.record(&mut visitor);
+
+        if visitor.thread_id.is_none() {
+            return;
+        }
+
+        if let Some(span) = ctx.span(id) {
+            let mut extensions = span.extensions_mut();
+            if let Some(log_context) = extensions.get_mut::<SpanLogContext>() {
+                log_context.thread_id = visitor.thread_id;
+            } else {
+                extensions.insert(SpanLogContext {
+                    thread_id: visitor.thread_id,
+                });
+            }
+        }
+    }
+
+    fn on_event(&self, event: &Event<'_>, ctx: tracing_subscriber::layer::Context<'_, S>) {
+        let metadata = event.metadata();
+        let mut visitor = MessageVisitor::default();
+        event.record(&mut visitor);
+        let thread_id = visitor
+            .thread_id
+            .clone()
+            .or_else(|| event_thread_id(event, &ctx));
+
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap_or_else(|_| Duration::from_secs(0));
+        let entry = LogEntry {
+            ts: now.as_secs() as i64,
+            ts_nanos: now.subsec_nanos() as i64,
+            level: metadata.level().as_str().to_string(),
+            target: metadata.target().to_string(),
+            message: visitor.message,
+            thread_id,
+            module_path: metadata.module_path().map(ToString::to_string),
+            file: metadata.file().map(ToString::to_string),
+            line: metadata.line().map(|line| line as i64),
+        };
+
+        let _ = self.sender.try_send(entry);
+    }
+}
+
+#[derive(Clone, Debug, Default)]
+struct SpanLogContext {
+    thread_id: Option<String>,
+}
+
+#[derive(Default)]
+struct SpanFieldVisitor {
+    thread_id: Option<String>,
+}
+
+impl SpanFieldVisitor {
+    fn record_field(&mut self, field: &Field, value: String) {
+        if field.name() == "thread_id" && self.thread_id.is_none() {
+            self.thread_id = Some(value);
+        }
+    }
+}
+
+impl Visit for SpanFieldVisitor {
+    fn record_i64(&mut self, field: &Field, value: i64) {
+        self.record_field(field, value.to_string());
+    }
+
+    fn record_u64(&mut self, field: &Field, value: u64) {
+        self.record_field(field, value.to_string());
+    }
+
+    fn record_bool(&mut self, field: &Field, value: bool) {
+        self.record_field(field, value.to_string());
+    }
+
+    fn record_f64(&mut self, field: &Field, value: f64) {
+        self.record_field(field, value.to_string());
+    }
+
+    fn record_str(&mut self, field: &Field, value: &str) {
+        self.record_field(field, value.to_string());
+    }
+
+    fn record_error(&mut self, field: &Field, value: &(dyn std::error::Error + 'static)) {
+        self.record_field(field, value.to_string());
+    }
+
+    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
+        self.record_field(field, format!("{value:?}"));
+    }
+}
+
+fn event_thread_id<S>(
+    event: &Event<'_>,
+    ctx: &tracing_subscriber::layer::Context<'_, S>,
+) -> Option<String>
+where
+    S: tracing::Subscriber + for<'a> LookupSpan<'a>,
+{
+    let mut thread_id = None;
+    if let Some(scope) = ctx.event_scope(event) {
+        for span in scope.from_root() {
+            let extensions = span.extensions();
+            if let Some(log_context) = extensions.get::<SpanLogContext>()
+                && log_context.thread_id.is_some()
+            {
+                thread_id = log_context.thread_id.clone();
+            }
+        }
+    }
+    thread_id
+}
+
+async fn run_inserter(
+    state_db: std::sync::Arc<StateRuntime>,
+    mut receiver: mpsc::Receiver<LogEntry>,
+) {
+    let mut buffer = Vec::with_capacity(LOG_BATCH_SIZE);
+    let mut ticker = tokio::time::interval(LOG_FLUSH_INTERVAL);
+    loop {
+        tokio::select! {
+            maybe_entry = receiver.recv() => {
+                match maybe_entry {
+                    Some(entry) => {
+                        buffer.push(entry);
+                        if buffer.len() >= LOG_BATCH_SIZE {
+                            flush(&state_db, &mut buffer).await;
+                        }
+                    }
+                    None => {
+                        flush(&state_db, &mut buffer).await;
+                        break;
+                    }
+                }
+            }
+            _ = ticker.tick() => {
+                flush(&state_db, &mut buffer).await;
+            }
+        }
+    }
+}
+
+async fn flush(state_db: &std::sync::Arc<StateRuntime>, buffer: &mut Vec<LogEntry>) {
+    if buffer.is_empty() {
+        return;
+    }
+    let entries = buffer.split_off(0);
+    let _ = state_db.insert_logs(entries.as_slice()).await;
+}
+
+async fn run_retention_cleanup(state_db: std::sync::Arc<StateRuntime>) {
+    let Some(cutoff) = Utc::now().checked_sub_signed(ChronoDuration::days(LOG_RETENTION_DAYS))
+    else {
+        return;
+    };
+    let _ = state_db.delete_logs_before(cutoff.timestamp()).await;
+}
+
+#[derive(Default)]
+struct MessageVisitor {
+    message: Option<String>,
+    thread_id: Option<String>,
+}
+
+impl MessageVisitor {
+    fn record_field(&mut self, field: &Field, value: String) {
+        if field.name() == "message" && self.message.is_none() {
+            self.message = Some(value.clone());
+        }
+        if field.name() == "thread_id" && self.thread_id.is_none() {
+            self.thread_id = Some(value);
+        }
+    }
+}
+
+impl Visit for MessageVisitor {
+    fn record_i64(&mut self, field: &Field, value: i64) {
+        self.record_field(field, value.to_string());
+    }
+
+    fn record_u64(&mut self, field: &Field, value: u64) {
+        self.record_field(field, value.to_string());
+    }
+
+    fn record_bool(&mut self, field: &Field, value: bool) {
+        self.record_field(field, value.to_string());
+    }
+
+    fn record_f64(&mut self, field: &Field, value: f64) {
+        self.record_field(field, value.to_string());
+    }
+
+    fn record_str(&mut self, field: &Field, value: &str) {
+        self.record_field(field, value.to_string());
+    }
+
+    fn record_error(&mut self, field: &Field, value: &(dyn std::error::Error + 'static)) {
+        self.record_field(field, value.to_string());
+    }
+
+    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
+        self.record_field(field, format!("{value:?}"));
+    }
+}

codex-rs/state/src/model/log.rs

@@ -0,0 +1,41 @@
+use serde::Serialize;
+use sqlx::FromRow;
+
+#[derive(Clone, Debug, Serialize)]
+pub struct LogEntry {
+    pub ts: i64,
+    pub ts_nanos: i64,
+    pub level: String,
+    pub target: String,
+    pub message: Option<String>,
+    pub thread_id: Option<String>,
+    pub module_path: Option<String>,
+    pub file: Option<String>,
+    pub line: Option<i64>,
+}
+
+#[derive(Clone, Debug, FromRow)]
+pub struct LogRow {
+    pub id: i64,
+    pub ts: i64,
+    pub ts_nanos: i64,
+    pub level: String,
+    pub target: String,
+    pub message: Option<String>,
+    pub thread_id: Option<String>,
+    pub file: Option<String>,
+    pub line: Option<i64>,
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct LogQuery {
+    pub level_upper: Option<String>,
+    pub from_ts: Option<i64>,
+    pub to_ts: Option<i64>,
+    pub module_like: Option<String>,
+    pub file_like: Option<String>,
+    pub thread_id: Option<String>,
+    pub after_id: Option<i64>,
+    pub limit: Option<usize>,
+    pub descending: bool,
+}

codex-rs/state/src/model/mod.rs

@@ -0,0 +1,17 @@
+mod log;
+mod thread_metadata;
+
+pub use log::LogEntry;
+pub use log::LogQuery;
+pub use log::LogRow;
+pub use thread_metadata::Anchor;
+pub use thread_metadata::BackfillStats;
+pub use thread_metadata::ExtractionOutcome;
+pub use thread_metadata::SortKey;
+pub use thread_metadata::ThreadMetadata;
+pub use thread_metadata::ThreadMetadataBuilder;
+pub use thread_metadata::ThreadsPage;
+
+pub(crate) use thread_metadata::ThreadRow;
+pub(crate) use thread_metadata::anchor_from_item;
+pub(crate) use thread_metadata::datetime_to_epoch_seconds;

codex-rs/state/src/runtime.rs

@@ -1,4 +1,7 @@
 use crate::DB_ERROR_METRIC;
+use crate::LogEntry;
+use crate::LogQuery;
+use crate::LogRow;
 use crate::SortKey;
 use crate::ThreadMetadata;
 use crate::ThreadMetadataBuilder;
@@ -201,6 +204,75 @@ FROM threads
         })
     }
 
+    /// Insert one log entry into the logs table.
+    pub async fn insert_log(&self, entry: &LogEntry) -> anyhow::Result<()> {
+        self.insert_logs(std::slice::from_ref(entry)).await
+    }
+
+    /// Insert a batch of log entries into the logs table.
+    pub async fn insert_logs(&self, entries: &[LogEntry]) -> anyhow::Result<()> {
+        if entries.is_empty() {
+            return Ok(());
+        }
+
+        let mut builder = QueryBuilder::<Sqlite>::new(
+            "INSERT INTO logs (ts, ts_nanos, level, target, message, thread_id, module_path, file, line) ",
+        );
+        builder.push_values(entries, |mut row, entry| {
+            row.push_bind(entry.ts)
+                .push_bind(entry.ts_nanos)
+                .push_bind(&entry.level)
+                .push_bind(&entry.target)
+                .push_bind(&entry.message)
+                .push_bind(&entry.thread_id)
+                .push_bind(&entry.module_path)
+                .push_bind(&entry.file)
+                .push_bind(entry.line);
+        });
+        builder.build().execute(self.pool.as_ref()).await?;
+        Ok(())
+    }
+
+    pub(crate) async fn delete_logs_before(&self, cutoff_ts: i64) -> anyhow::Result<u64> {
+        let result = sqlx::query("DELETE FROM logs WHERE ts < ?")
+            .bind(cutoff_ts)
+            .execute(self.pool.as_ref())
+            .await?;
+        Ok(result.rows_affected())
+    }
+
+    /// Query logs with optional filters.
+    pub async fn query_logs(&self, query: &LogQuery) -> anyhow::Result<Vec<LogRow>> {
+        let mut builder = QueryBuilder::<Sqlite>::new(
+            "SELECT id, ts, ts_nanos, level, target, message, thread_id, file, line FROM logs WHERE 1 = 1",
+        );
+        push_log_filters(&mut builder, query);
+        if query.descending {
+            builder.push(" ORDER BY id DESC");
+        } else {
+            builder.push(" ORDER BY id ASC");
+        }
+        if let Some(limit) = query.limit {
+            builder.push(" LIMIT ").push_bind(limit as i64);
+        }
+
+        let rows = builder
+            .build_query_as::<LogRow>()
+            .fetch_all(self.pool.as_ref())
+            .await?;
+        Ok(rows)
+    }
+
+    /// Return the max log id matching optional filters.
+    pub async fn max_log_id(&self, query: &LogQuery) -> anyhow::Result<i64> {
+        let mut builder =
+            QueryBuilder::<Sqlite>::new("SELECT MAX(id) AS max_id FROM logs WHERE 1 = 1");
+        push_log_filters(&mut builder, query);
+        let row = builder.build().fetch_one(self.pool.as_ref()).await?;
+        let max_id: Option<i64> = row.try_get("max_id")?;
+        Ok(max_id.unwrap_or(0))
+    }
+
     /// List thread ids using the underlying database (no rollout scanning).
     pub async fn list_thread_ids(
         &self,
@@ -373,6 +445,40 @@ ON CONFLICT(id) DO UPDATE SET
     }
 }
 
+fn push_log_filters<'a>(builder: &mut QueryBuilder<'a, Sqlite>, query: &'a LogQuery) {
+    if let Some(level_upper) = query.level_upper.as_ref() {
+        builder
+            .push(" AND UPPER(level) = ")
+            .push_bind(level_upper.as_str());
+    }
+    if let Some(from_ts) = query.from_ts {
+        builder.push(" AND ts >= ").push_bind(from_ts);
+    }
+    if let Some(to_ts) = query.to_ts {
+        builder.push(" AND ts <= ").push_bind(to_ts);
+    }
+    if let Some(module_like) = query.module_like.as_ref() {
+        builder
+            .push(" AND module_path LIKE '%' || ")
+            .push_bind(module_like.as_str())
+            .push(" || '%'");
+    }
+    if let Some(file_like) = query.file_like.as_ref() {
+        builder
+            .push(" AND file LIKE '%' || ")
+            .push_bind(file_like.as_str())
+            .push(" || '%'");
+    }
+    if let Some(thread_id) = query.thread_id.as_ref() {
+        builder
+            .push(" AND thread_id = ")
+            .push_bind(thread_id.as_str());
+    }
+    if let Some(after_id) = query.after_id {
+        builder.push(" AND id > ").push_bind(after_id);
+    }
+}
+
 async fn open_sqlite(path: &Path) -> anyhow::Result<SqlitePool> {
     let options = SqliteConnectOptions::new()
         .filename(path)

codex-rs/tui/Cargo.toml

@@ -42,6 +42,7 @@ codex-file-search = { workspace = true }
 codex-login = { workspace = true }
 codex-otel = { workspace = true }
 codex-protocol = { workspace = true }
+codex-state = { workspace = true }
 codex-utils-absolute-path = { workspace = true }
 color-eyre = { workspace = true }
 crossterm = { workspace = true, features = ["bracketed-paste", "event-stream"] }

codex-rs/tui/src/lib.rs

@@ -36,6 +36,7 @@ use codex_protocol::config_types::SandboxMode;
 use codex_protocol::config_types::WindowsSandboxLevel;
 use codex_protocol::protocol::RolloutItem;
 use codex_protocol::protocol::RolloutLine;
+use codex_state::log_db;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use cwd_prompt::CwdPromptAction;
 use cwd_prompt::CwdSelection;
@@ -305,7 +306,10 @@ pub async fn run_main(
         // grep for a specific module/target while troubleshooting.
         .with_target(true)
         .with_ansi(false)
-        .with_span_events(tracing_subscriber::fmt::format::FmtSpan::FULL)
+        .with_span_events(
+            tracing_subscriber::fmt::format::FmtSpan::NEW
+                | tracing_subscriber::fmt::format::FmtSpan::CLOSE,
+        )
         .with_filter(env_filter());
 
     let feedback = codex_feedback::CodexFeedback::new();
@@ -351,10 +355,15 @@ pub async fn run_main(
 
     let otel_tracing_layer = otel.as_ref().and_then(|o| o.tracing_layer());
 
+    let log_db_layer = codex_core::state_db::get_state_db(&config, None)
+        .await
+        .map(|db| log_db::start(db).with_filter(env_filter()));
+
     let _ = tracing_subscriber::registry()
         .with(file_layer)
         .with(feedback_layer)
         .with(feedback_metadata_layer)
+        .with(log_db_layer)
         .with(otel_logger_layer)
         .with(otel_tracing_layer)
         .try_init();

codex-rs/tui/tooltips.txt

@@ -9,6 +9,7 @@ Use /status to see the current model, approvals, and token usage.
 Use /fork to branch the current chat into a new thread.
 Use /init to create an AGENTS.md with project-specific guidance.
 Use /mcp to list configured MCP tools.
+Use the OpenAI docs MCP for API questions; enable it with `codex mcp add openaiDeveloperDocs --url https://developers.openai.com/mcp`.
 You can run any shell command from Codex using `!` (e.g. `!ls`)
 Type / to open the command popup; Tab autocompletes slash commands.
 When the composer is empty, press Esc to step back and edit your last message; Enter confirms.

justfile

@@ -67,3 +67,7 @@ mcp-server-run *args:
 # Regenerate the json schema for config.toml from the current config types.
 write-config-schema:
     cargo run -p codex-core --bin codex-write-config-schema
+
+# Tail logs from the state SQLite database
+log *args:
+    if [ "${1:-}" = "--" ]; then shift; fi; cargo run -p codex-state --bin logs_client -- "$@"