From 780482da84e2bb14b9900e75b04ca2e4f65dfddb Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 29 Jan 2026 10:23:03 +0100 Subject: [PATCH 01/13] feat: add log db (#10086) Add a log DB. The goal is just to store our logs in a `.sqlite` DB to make it easier to crawl them and drop the oldest ones. --- codex-rs/Cargo.lock | 2 + codex-rs/core/src/state_db.rs | 1 + codex-rs/state/Cargo.toml | 1 + codex-rs/state/migrations/0002_logs.sql | 14 ++ codex-rs/state/src/lib.rs | 2 + codex-rs/state/src/log_db.rs | 162 ++++++++++++++++++ codex-rs/state/src/model/log.rs | 14 ++ codex-rs/state/src/model/mod.rs | 15 ++ .../{model.rs => model/thread_metadata.rs} | 0 codex-rs/state/src/runtime.rs | 30 ++++ codex-rs/tui/Cargo.toml | 1 + codex-rs/tui/src/lib.rs | 6 + 12 files changed, 248 insertions(+) create mode 100644 codex-rs/state/migrations/0002_logs.sql create mode 100644 codex-rs/state/src/log_db.rs create mode 100644 codex-rs/state/src/model/log.rs create mode 100644 codex-rs/state/src/model/mod.rs rename codex-rs/state/src/{model.rs => model/thread_metadata.rs} (100%) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 08cc850524..0ae6c60d42 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1859,6 +1859,7 @@ dependencies = [ "sqlx", "tokio", "tracing", + "tracing-subscriber", "uuid", ] @@ -1897,6 +1898,7 @@ dependencies = [ "codex-login", "codex-otel", "codex-protocol", + "codex-state", "codex-utils-absolute-path", "codex-utils-cargo-bin", "codex-utils-pty", diff --git a/codex-rs/core/src/state_db.rs b/codex-rs/core/src/state_db.rs index d859328e1c..74c9ed7383 100644 --- a/codex-rs/core/src/state_db.rs +++ b/codex-rs/core/src/state_db.rs @@ -12,6 +12,7 @@ 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; diff --git a/codex-rs/state/Cargo.toml b/codex-rs/state/Cargo.toml index 810e250ca7..b8c7fce0e3 100644 --- a/codex-rs/state/Cargo.toml +++ b/codex-rs/state/Cargo.toml @@ -14,6 +14,7 @@ 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] diff --git a/codex-rs/state/migrations/0002_logs.sql b/codex-rs/state/migrations/0002_logs.sql new file mode 100644 index 0000000000..4a193500b9 --- /dev/null +++ b/codex-rs/state/migrations/0002_logs.sql @@ -0,0 +1,14 @@ +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, + fields_json TEXT NOT NULL, + module_path TEXT, + file TEXT, + line INTEGER +); + +CREATE INDEX idx_logs_ts ON logs(ts DESC, ts_nanos DESC, id DESC); diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index 67533a3791..91ebacea71 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -5,11 +5,13 @@ //! 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; /// Preferred entrypoint: owns configuration and metrics. pub use runtime::StateRuntime; diff --git a/codex-rs/state/src/log_db.rs b/codex-rs/state/src/log_db.rs new file mode 100644 index 0000000000..c03b7977d3 --- /dev/null +++ b/codex-rs/state/src/log_db.rs @@ -0,0 +1,162 @@ +//! 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) { +//! let layer = log_db::start(state_db); +//! let _ = tracing_subscriber::registry() +//! .with(layer) +//! .try_init(); +//! # } +//! ``` + +use std::time::Duration; +use std::time::SystemTime; +use std::time::UNIX_EPOCH; + +use serde_json::Value; +use tokio::sync::mpsc; +use tracing::Event; +use tracing::field::Field; +use tracing::field::Visit; +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); + +pub struct LogDbLayer { + sender: mpsc::Sender, +} + +pub fn start(state_db: std::sync::Arc) -> LogDbLayer { + let (sender, receiver) = mpsc::channel(LOG_QUEUE_CAPACITY); + tokio::spawn(run_inserter(state_db, receiver)); + + LogDbLayer { sender } +} + +impl Layer for LogDbLayer +where + S: tracing::Subscriber + for<'a> LookupSpan<'a>, +{ + fn on_event(&self, event: &Event<'_>, _ctx: tracing_subscriber::layer::Context<'_, S>) { + let metadata = event.metadata(); + let mut visitor = JsonVisitor::default(); + event.record(&mut visitor); + + 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, + fields_json: Value::Object(visitor.fields).to_string(), + 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); + } +} + +async fn run_inserter( + state_db: std::sync::Arc, + mut receiver: mpsc::Receiver, +) { + 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, buffer: &mut Vec) { + if buffer.is_empty() { + return; + } + let entries = buffer.split_off(0); + let _ = state_db.insert_logs(entries.as_slice()).await; +} + +#[derive(Default)] +struct JsonVisitor { + fields: serde_json::Map, + message: Option, +} + +impl JsonVisitor { + fn record_value(&mut self, field: &Field, value: Value) { + if field.name() == "message" && self.message.is_none() { + self.message = Some(match &value { + Value::String(message) => message.clone(), + _ => value.to_string(), + }); + } + self.fields.insert(field.name().to_string(), value); + } +} + +impl Visit for JsonVisitor { + fn record_i64(&mut self, field: &Field, value: i64) { + self.record_value(field, Value::from(value)); + } + + fn record_u64(&mut self, field: &Field, value: u64) { + self.record_value(field, Value::from(value)); + } + + fn record_bool(&mut self, field: &Field, value: bool) { + self.record_value(field, Value::from(value)); + } + + fn record_f64(&mut self, field: &Field, value: f64) { + self.record_value(field, Value::from(value)); + } + + fn record_str(&mut self, field: &Field, value: &str) { + self.record_value(field, Value::from(value)); + } + + fn record_error(&mut self, field: &Field, value: &(dyn std::error::Error + 'static)) { + self.record_value(field, Value::from(value.to_string())); + } + + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.record_value(field, Value::from(format!("{value:?}"))); + } +} diff --git a/codex-rs/state/src/model/log.rs b/codex-rs/state/src/model/log.rs new file mode 100644 index 0000000000..5662a3725c --- /dev/null +++ b/codex-rs/state/src/model/log.rs @@ -0,0 +1,14 @@ +use serde::Serialize; + +#[derive(Clone, Debug, Serialize)] +pub struct LogEntry { + pub ts: i64, + pub ts_nanos: i64, + pub level: String, + pub target: String, + pub message: Option, + pub fields_json: String, + pub module_path: Option, + pub file: Option, + pub line: Option, +} diff --git a/codex-rs/state/src/model/mod.rs b/codex-rs/state/src/model/mod.rs new file mode 100644 index 0000000000..4921e2357a --- /dev/null +++ b/codex-rs/state/src/model/mod.rs @@ -0,0 +1,15 @@ +mod log; +mod thread_metadata; + +pub use log::LogEntry; +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; diff --git a/codex-rs/state/src/model.rs b/codex-rs/state/src/model/thread_metadata.rs similarity index 100% rename from codex-rs/state/src/model.rs rename to codex-rs/state/src/model/thread_metadata.rs diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index f341ee3bb6..34add1347d 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -1,4 +1,5 @@ use crate::DB_ERROR_METRIC; +use crate::LogEntry; use crate::SortKey; use crate::ThreadMetadata; use crate::ThreadMetadataBuilder; @@ -201,6 +202,35 @@ 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::::new( + "INSERT INTO logs (ts, ts_nanos, level, target, message, fields_json, 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.fields_json) + .push_bind(&entry.module_path) + .push_bind(&entry.file) + .push_bind(entry.line); + }); + builder.build().execute(self.pool.as_ref()).await?; + Ok(()) + } + /// List thread ids using the underlying database (no rollout scanning). pub async fn list_thread_ids( &self, diff --git a/codex-rs/tui/Cargo.toml b/codex-rs/tui/Cargo.toml index 820fe20e93..6664e7f9c5 100644 --- a/codex-rs/tui/Cargo.toml +++ b/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"] } diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index bced1e0653..a48b806129 100644 --- a/codex-rs/tui/src/lib.rs +++ b/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; @@ -351,10 +352,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::init_if_enabled(&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(); From 714dc8d8bd1a8d28fe71df4e1c1bc5f3f5752b91 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 29 Jan 2026 10:57:50 +0100 Subject: [PATCH 02/13] feat: async backfill (#10089) --- .../app-server/src/codex_message_processor.rs | 6 +- codex-rs/core/src/rollout/metadata.rs | 22 +++++++- codex-rs/core/src/state_db.rs | 55 ++++++++++++------- codex-rs/tui/src/lib.rs | 2 +- 4 files changed, 58 insertions(+), 27 deletions(-) diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index e86d840b94..e1c982a99d 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/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. diff --git a/codex-rs/core/src/rollout/metadata.rs b/codex-rs/core/src/rollout/metadata.rs index 32d13ebde2..d08e77e241 100644 --- a/codex-rs/core/src/rollout/metadata.rs +++ b/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> { diff --git a/codex-rs/core/src/state_db.rs b/codex-rs/core/src/state_db.rs index 74c9ed7383..7990a017ef 100644 --- a/codex-rs/core/src/state_db.rs +++ b/codex-rs/core/src/state_db.rs @@ -11,7 +11,6 @@ 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; @@ -19,15 +18,18 @@ 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; -/// Initialize the state runtime when the `sqlite` feature flag is enabled. -pub async fn init_if_enabled(config: &Config, otel: Option<&OtelManager>) -> Option { +/// 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 { 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. @@ -59,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 { + 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. diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index a48b806129..c9e69b7125 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -352,7 +352,7 @@ 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::init_if_enabled(&config, None) + let log_db_layer = codex_core::state_db::get_state_db(&config, None) .await .map(|db| log_db::start(db).with_filter(env_filter())); From 6a06726af2469e766528c5f6460e5f047fabd7f3 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 29 Jan 2026 11:11:47 +0100 Subject: [PATCH 03/13] feat: log db client (#10087) ``` just log -h if [ "${1:-}" = "--" ]; then shift; fi; cargo run -p codex-state --bin logs_client -- "$@" Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.21s Running `target/debug/logs_client -h` Tail Codex logs from state.sqlite with simple filters Usage: logs_client [OPTIONS] Options: --codex-home Path to CODEX_HOME. Defaults to $CODEX_HOME or ~/.codex [env: CODEX_HOME=] --db Direct path to the SQLite database. Overrides --codex-home --level Log level to match exactly (case-insensitive) --from Start timestamp (RFC3339 or unix seconds) --to End timestamp (RFC3339 or unix seconds) --module Substring match on module_path --file Substring match on file path --backfill Number of matching rows to show before tailing [default: 200] --poll-ms Poll interval in milliseconds [default: 500] -h, --help Print help ``` --- codex-rs/Cargo.lock | 2 + codex-rs/state/Cargo.toml | 2 + codex-rs/state/src/bin/logs_client.rs | 293 ++++++++++++++++++++++++++ justfile | 4 + 4 files changed, 301 insertions(+) create mode 100644 codex-rs/state/src/bin/logs_client.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 0ae6c60d42..f47e054ba6 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1851,8 +1851,10 @@ version = "0.0.0" dependencies = [ "anyhow", "chrono", + "clap", "codex-otel", "codex-protocol", + "dirs", "pretty_assertions", "serde", "serde_json", diff --git a/codex-rs/state/Cargo.toml b/codex-rs/state/Cargo.toml index b8c7fce0e3..a6e36a3e00 100644 --- a/codex-rs/state/Cargo.toml +++ b/codex-rs/state/Cargo.toml @@ -7,8 +7,10 @@ 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 } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } sqlx = { workspace = true } diff --git a/codex-rs/state/src/bin/logs_client.rs b/codex-rs/state/src/bin/logs_client.rs new file mode 100644 index 0000000000..13bcb1571d --- /dev/null +++ b/codex-rs/state/src/bin/logs_client.rs @@ -0,0 +1,293 @@ +use std::path::Path; +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::STATE_DB_FILENAME; +use dirs::home_dir; +use sqlx::QueryBuilder; +use sqlx::Row; +use sqlx::Sqlite; +use sqlx::SqlitePool; +use sqlx::sqlite::SqliteConnectOptions; +use sqlx::sqlite::SqlitePoolOptions; + +#[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, + + /// Direct path to the SQLite database. Overrides --codex-home. + #[arg(long)] + db: Option, + + /// Log level to match exactly (case-insensitive). + #[arg(long)] + level: Option, + + /// Start timestamp (RFC3339 or unix seconds). + #[arg(long, value_name = "RFC3339|UNIX")] + from: Option, + + /// End timestamp (RFC3339 or unix seconds). + #[arg(long, value_name = "RFC3339|UNIX")] + to: Option, + + /// Substring match on module_path. + #[arg(long)] + module: Option, + + /// Substring match on file path. + #[arg(long)] + file: Option, + + /// 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, sqlx::FromRow)] +struct LogRow { + id: i64, + ts: i64, + ts_nanos: i64, + level: String, + message: Option, + fields_json: String, + module_path: Option, + file: Option, + line: Option, +} + +#[derive(Debug, Clone)] +struct LogFilter { + level_upper: Option, + from_ts: Option, + to_ts: Option, + module_like: Option, + file_like: Option, +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let args = Args::parse(); + let db_path = resolve_db_path(&args)?; + let filter = build_filter(&args)?; + let pool = open_read_only_pool(db_path.as_path()).await?; + + let mut last_id = print_backfill(&pool, &filter, args.backfill).await?; + if last_id == 0 { + last_id = fetch_max_id(&pool, &filter).await?; + } + + let poll_interval = Duration::from_millis(args.poll_ms); + loop { + let rows = fetch_new_rows(&pool, &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 { + 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 { + 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(), + }) +} + +fn parse_timestamp(value: &str) -> anyhow::Result { + if let Ok(secs) = value.parse::() { + 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 open_read_only_pool(path: &Path) -> anyhow::Result { + let options = SqliteConnectOptions::new() + .filename(path) + .create_if_missing(false) + .read_only(true) + .busy_timeout(Duration::from_secs(5)); + + let display = path.display(); + SqlitePoolOptions::new() + .max_connections(1) + .connect_with(options) + .await + .with_context(|| format!("failed to open sqlite db at {display}")) +} + +async fn print_backfill( + pool: &SqlitePool, + filter: &LogFilter, + backfill: usize, +) -> anyhow::Result { + if backfill == 0 { + return Ok(0); + } + + let mut rows = fetch_backfill(pool, 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( + pool: &SqlitePool, + filter: &LogFilter, + backfill: usize, +) -> anyhow::Result> { + let mut builder = base_select_builder(); + push_filters(&mut builder, filter); + builder.push(" ORDER BY id DESC"); + builder.push(" LIMIT ").push_bind(backfill as i64); + + builder + .build_query_as::() + .fetch_all(pool) + .await + .context("failed to fetch backfill logs") +} + +async fn fetch_new_rows( + pool: &SqlitePool, + filter: &LogFilter, + last_id: i64, +) -> anyhow::Result> { + let mut builder = base_select_builder(); + push_filters(&mut builder, filter); + builder.push(" AND id > ").push_bind(last_id); + builder.push(" ORDER BY id ASC"); + + builder + .build_query_as::() + .fetch_all(pool) + .await + .context("failed to fetch new logs") +} + +async fn fetch_max_id(pool: &SqlitePool, filter: &LogFilter) -> anyhow::Result { + let mut builder = QueryBuilder::::new("SELECT MAX(id) AS max_id FROM logs WHERE 1 = 1"); + push_filters(&mut builder, filter); + + let row = builder + .build() + .fetch_one(pool) + .await + .context("failed to fetch max log id")?; + let max_id: Option = row.try_get("max_id")?; + Ok(max_id.unwrap_or(0)) +} + +fn base_select_builder<'a>() -> QueryBuilder<'a, Sqlite> { + QueryBuilder::::new( + "SELECT id, ts, ts_nanos, level, message, fields_json, module_path, file, line FROM logs WHERE 1 = 1", + ) +} + +fn push_filters<'a>(builder: &mut QueryBuilder<'a, Sqlite>, filter: &'a LogFilter) { + if let Some(level_upper) = filter.level_upper.as_ref() { + builder + .push(" AND UPPER(level) = ") + .push_bind(level_upper.as_str()); + } + if let Some(from_ts) = filter.from_ts { + builder.push(" AND ts >= ").push_bind(from_ts); + } + if let Some(to_ts) = filter.to_ts { + builder.push(" AND ts <= ").push_bind(to_ts); + } + if let Some(module_like) = filter.module_like.as_ref() { + builder + .push(" AND module_path LIKE '%' || ") + .push_bind(module_like.as_str()) + .push(" || '%'"); + } + if let Some(file_like) = filter.file_like.as_ref() { + builder + .push(" AND file LIKE '%' || ") + .push_bind(file_like.as_str()) + .push(" || '%'"); + } +} + +fn format_row(row: &LogRow) -> String { + let timestamp = format_timestamp(row.ts, row.ts_nanos); + let location = match (&row.file, row.line) { + (Some(file), Some(line)) => format!("{file}:{line}"), + (Some(file), None) => file.clone(), + _ => "-".to_string(), + }; + let module = row.module_path.as_deref().unwrap_or("-"); + let message = row.message.as_deref().unwrap_or(""); + let fields = row.fields_json.as_str(); + let level = row.level.as_str(); + if fields == "{}" || fields.is_empty() { + return format!("{timestamp} {level:<5} [{module}] {location} - {message}"); + } + format!("{timestamp} {level:<5} [{module}] {location} - {message} {fields}") +} + +fn format_timestamp(ts: i64, ts_nanos: i64) -> String { + let nanos = u32::try_from(ts_nanos).unwrap_or(0); + match DateTime::::from_timestamp(ts, nanos) { + Some(dt) => dt.to_rfc3339_opts(SecondsFormat::Millis, true), + None => format!("{ts}.{ts_nanos:09}Z"), + } +} diff --git a/justfile b/justfile index 0e01ef7e1d..8fb5a7d941 100644 --- a/justfile +++ b/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 -- "$@" From 4ba911d48c1f5c554bd447b35b42238ef3c1b419 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 29 Jan 2026 11:25:22 +0100 Subject: [PATCH 04/13] chore: improve client (#10149) Screenshot 2026-01-29 at 11 13 12 --- codex-rs/Cargo.lock | 1 + codex-rs/state/Cargo.toml | 1 + codex-rs/state/migrations/0002_logs.sql | 1 - codex-rs/state/src/bin/logs_client.rs | 36 ++++++++++++++++++------- codex-rs/state/src/log_db.rs | 33 +++++++++-------------- codex-rs/state/src/model/log.rs | 1 - codex-rs/state/src/runtime.rs | 3 +-- 7 files changed, 43 insertions(+), 33 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index f47e054ba6..4c321b3f62 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1855,6 +1855,7 @@ dependencies = [ "codex-otel", "codex-protocol", "dirs", + "owo-colors", "pretty_assertions", "serde", "serde_json", diff --git a/codex-rs/state/Cargo.toml b/codex-rs/state/Cargo.toml index a6e36a3e00..6019b5888b 100644 --- a/codex-rs/state/Cargo.toml +++ b/codex-rs/state/Cargo.toml @@ -11,6 +11,7 @@ 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 } diff --git a/codex-rs/state/migrations/0002_logs.sql b/codex-rs/state/migrations/0002_logs.sql index 4a193500b9..b9a2c681d4 100644 --- a/codex-rs/state/migrations/0002_logs.sql +++ b/codex-rs/state/migrations/0002_logs.sql @@ -5,7 +5,6 @@ CREATE TABLE logs ( level TEXT NOT NULL, target TEXT NOT NULL, message TEXT, - fields_json TEXT NOT NULL, module_path TEXT, file TEXT, line INTEGER diff --git a/codex-rs/state/src/bin/logs_client.rs b/codex-rs/state/src/bin/logs_client.rs index 13bcb1571d..71acf67e85 100644 --- a/codex-rs/state/src/bin/logs_client.rs +++ b/codex-rs/state/src/bin/logs_client.rs @@ -9,6 +9,7 @@ use chrono::Utc; use clap::Parser; use codex_state::STATE_DB_FILENAME; use dirs::home_dir; +use owo_colors::OwoColorize; use sqlx::QueryBuilder; use sqlx::Row; use sqlx::Sqlite; @@ -64,8 +65,6 @@ struct LogRow { ts_nanos: i64, level: String, message: Option, - fields_json: String, - module_path: Option, file: Option, line: Option, } @@ -237,7 +236,7 @@ async fn fetch_max_id(pool: &SqlitePool, filter: &LogFilter) -> anyhow::Result() -> QueryBuilder<'a, Sqlite> { QueryBuilder::::new( - "SELECT id, ts, ts_nanos, level, message, fields_json, module_path, file, line FROM logs WHERE 1 = 1", + "SELECT id, ts, ts_nanos, level, message, file, line FROM logs WHERE 1 = 1", ) } @@ -269,19 +268,38 @@ fn push_filters<'a>(builder: &mut QueryBuilder<'a, Sqlite>, filter: &'a LogFilte fn format_row(row: &LogRow) -> String { let timestamp = format_timestamp(row.ts, row.ts_nanos); + let level = row.level.as_str(); let location = match (&row.file, row.line) { (Some(file), Some(line)) => format!("{file}:{line}"), (Some(file), None) => file.clone(), _ => "-".to_string(), }; - let module = row.module_path.as_deref().unwrap_or("-"); let message = row.message.as_deref().unwrap_or(""); - let fields = row.fields_json.as_str(); - let level = row.level.as_str(); - if fields == "{}" || fields.is_empty() { - return format!("{timestamp} {level:<5} [{module}] {location} - {message}"); + let level_colored = color_level(level); + let timestamp_colored = timestamp.dimmed().to_string(); + let location_colored = location.dimmed().to_string(); + let message_colored = message.bold().to_string(); + format!("{timestamp_colored} {level_colored} {location_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(); } - format!("{timestamp} {level:<5} [{module}] {location} - {message} {fields}") + 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 { diff --git a/codex-rs/state/src/log_db.rs b/codex-rs/state/src/log_db.rs index c03b7977d3..13f19d0baf 100644 --- a/codex-rs/state/src/log_db.rs +++ b/codex-rs/state/src/log_db.rs @@ -22,7 +22,6 @@ use std::time::Duration; use std::time::SystemTime; use std::time::UNIX_EPOCH; -use serde_json::Value; use tokio::sync::mpsc; use tracing::Event; use tracing::field::Field; @@ -54,7 +53,7 @@ where { fn on_event(&self, event: &Event<'_>, _ctx: tracing_subscriber::layer::Context<'_, S>) { let metadata = event.metadata(); - let mut visitor = JsonVisitor::default(); + let mut visitor = MessageVisitor::default(); event.record(&mut visitor); let now = SystemTime::now() @@ -66,7 +65,6 @@ where level: metadata.level().as_str().to_string(), target: metadata.target().to_string(), message: visitor.message, - fields_json: Value::Object(visitor.fields).to_string(), module_path: metadata.module_path().map(ToString::to_string), file: metadata.file().map(ToString::to_string), line: metadata.line().map(|line| line as i64), @@ -114,49 +112,44 @@ async fn flush(state_db: &std::sync::Arc, buffer: &mut Vec, +struct MessageVisitor { message: Option, } -impl JsonVisitor { - fn record_value(&mut self, field: &Field, value: Value) { +impl MessageVisitor { + fn record_message(&mut self, field: &Field, value: String) { if field.name() == "message" && self.message.is_none() { - self.message = Some(match &value { - Value::String(message) => message.clone(), - _ => value.to_string(), - }); + self.message = Some(value); } - self.fields.insert(field.name().to_string(), value); } } -impl Visit for JsonVisitor { +impl Visit for MessageVisitor { fn record_i64(&mut self, field: &Field, value: i64) { - self.record_value(field, Value::from(value)); + self.record_message(field, value.to_string()); } fn record_u64(&mut self, field: &Field, value: u64) { - self.record_value(field, Value::from(value)); + self.record_message(field, value.to_string()); } fn record_bool(&mut self, field: &Field, value: bool) { - self.record_value(field, Value::from(value)); + self.record_message(field, value.to_string()); } fn record_f64(&mut self, field: &Field, value: f64) { - self.record_value(field, Value::from(value)); + self.record_message(field, value.to_string()); } fn record_str(&mut self, field: &Field, value: &str) { - self.record_value(field, Value::from(value)); + self.record_message(field, value.to_string()); } fn record_error(&mut self, field: &Field, value: &(dyn std::error::Error + 'static)) { - self.record_value(field, Value::from(value.to_string())); + self.record_message(field, value.to_string()); } fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { - self.record_value(field, Value::from(format!("{value:?}"))); + self.record_message(field, format!("{value:?}")); } } diff --git a/codex-rs/state/src/model/log.rs b/codex-rs/state/src/model/log.rs index 5662a3725c..0fa4c0b38e 100644 --- a/codex-rs/state/src/model/log.rs +++ b/codex-rs/state/src/model/log.rs @@ -7,7 +7,6 @@ pub struct LogEntry { pub level: String, pub target: String, pub message: Option, - pub fields_json: String, pub module_path: Option, pub file: Option, pub line: Option, diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 34add1347d..96b2d69122 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -214,7 +214,7 @@ FROM threads } let mut builder = QueryBuilder::::new( - "INSERT INTO logs (ts, ts_nanos, level, target, message, fields_json, module_path, file, line) ", + "INSERT INTO logs (ts, ts_nanos, level, target, message, module_path, file, line) ", ); builder.push_values(entries, |mut row, entry| { row.push_bind(entry.ts) @@ -222,7 +222,6 @@ FROM threads .push_bind(&entry.level) .push_bind(&entry.target) .push_bind(&entry.message) - .push_bind(&entry.fields_json) .push_bind(&entry.module_path) .push_bind(&entry.file) .push_bind(entry.line); From d29129f352d02458fc69bb06066e1339d46e0e4c Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 29 Jan 2026 15:08:22 +0100 Subject: [PATCH 05/13] nit: update npm (#10161) --- .github/workflows/shell-tool-mcp.yml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.github/workflows/shell-tool-mcp.yml b/.github/workflows/shell-tool-mcp.yml index d6c85d9ddc..57bfdd5f5b 100644 --- a/.github/workflows/shell-tool-mcp.yml +++ b/.github/workflows/shell-tool-mcp.yml @@ -451,6 +451,10 @@ jobs: registry-url: https://registry.npmjs.org scope: "@openai" + # Trusted publishing requires npm CLI version 11.5.1 or later. + - name: Update npm + run: npm install -g npm@latest + - name: Download npm tarball uses: actions/download-artifact@v7 with: From 2945667dcc7b655fc3bf044f04b69f6cbdf31b69 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 29 Jan 2026 15:46:15 +0100 Subject: [PATCH 06/13] [experimental] nit: try to speed up apt-install (#10163) --- .github/workflows/shell-tool-mcp.yml | 28 ++++++++++++++++++++++++++-- 1 file changed, 26 insertions(+), 2 deletions(-) diff --git a/.github/workflows/shell-tool-mcp.yml b/.github/workflows/shell-tool-mcp.yml index 57bfdd5f5b..3273bb47bc 100644 --- a/.github/workflows/shell-tool-mcp.yml +++ b/.github/workflows/shell-tool-mcp.yml @@ -248,8 +248,32 @@ jobs: run: | set -euo pipefail if command -v apt-get >/dev/null 2>&1; then - apt-get update - DEBIAN_FRONTEND=noninteractive apt-get install -y git build-essential bison autoconf gettext + # On Ubuntu ARM containers, apt metadata for universe/multiverse is large and + # can make `apt-get update` very slow. We only need packages from `main`, so + # trim sources to `main` for faster, more reliable updates. + if [[ "${{ matrix.target }}" == "aarch64-unknown-linux-musl" ]] && [[ -f /etc/os-release ]]; then + . /etc/os-release + if [[ "${ID-}" == "ubuntu" ]]; then + codename="${VERSION_CODENAME:-}" + if [[ -n "$codename" ]]; then + printf '%s\n' \ + "deb http://ports.ubuntu.com/ubuntu-ports ${codename} main" \ + "deb http://ports.ubuntu.com/ubuntu-ports ${codename}-updates main" \ + "deb http://ports.ubuntu.com/ubuntu-ports ${codename}-backports main" \ + "deb http://ports.ubuntu.com/ubuntu-ports ${codename}-security main" \ + >/etc/apt/sources.list + fi + fi + fi + + apt_opts=( + -o Acquire::Retries=3 + -o Acquire::http::Timeout=30 + -o Acquire::https::Timeout=30 + -o Acquire::Languages=none + ) + apt-get "${apt_opts[@]}" update + DEBIAN_FRONTEND=noninteractive apt-get "${apt_opts[@]}" install -y git build-essential bison autoconf gettext elif command -v dnf >/dev/null 2>&1; then dnf install -y git gcc gcc-c++ make bison autoconf gettext elif command -v yum >/dev/null 2>&1; then From b654b7a9aeb8e222e56bbe2814587ca6fafadac5 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 29 Jan 2026 15:59:56 +0100 Subject: [PATCH 07/13] [experimental] nit: try to speed up apt-install 2 (#10164) --- .github/workflows/shell-tool-mcp.yml | 21 +++++++++++++-------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/.github/workflows/shell-tool-mcp.yml b/.github/workflows/shell-tool-mcp.yml index 3273bb47bc..ec0c16cd5e 100644 --- a/.github/workflows/shell-tool-mcp.yml +++ b/.github/workflows/shell-tool-mcp.yml @@ -254,14 +254,19 @@ jobs: if [[ "${{ matrix.target }}" == "aarch64-unknown-linux-musl" ]] && [[ -f /etc/os-release ]]; then . /etc/os-release if [[ "${ID-}" == "ubuntu" ]]; then - codename="${VERSION_CODENAME:-}" - if [[ -n "$codename" ]]; then - printf '%s\n' \ - "deb http://ports.ubuntu.com/ubuntu-ports ${codename} main" \ - "deb http://ports.ubuntu.com/ubuntu-ports ${codename}-updates main" \ - "deb http://ports.ubuntu.com/ubuntu-ports ${codename}-backports main" \ - "deb http://ports.ubuntu.com/ubuntu-ports ${codename}-security main" \ - >/etc/apt/sources.list + if [[ -f /etc/apt/sources.list.d/ubuntu.sources ]]; then + # Ubuntu 24.04+ uses deb822 sources; trim components to main. + sed -i -E 's/^Components:.*/Components: main/' /etc/apt/sources.list.d/ubuntu.sources + else + codename="${VERSION_CODENAME:-}" + if [[ -n "$codename" ]]; then + printf '%s\n' \ + "deb http://ports.ubuntu.com/ubuntu-ports ${codename} main" \ + "deb http://ports.ubuntu.com/ubuntu-ports ${codename}-updates main" \ + "deb http://ports.ubuntu.com/ubuntu-ports ${codename}-backports main" \ + "deb http://ports.ubuntu.com/ubuntu-ports ${codename}-security main" \ + >/etc/apt/sources.list + fi fi fi fi From 89c5f3c4d43b174054bd5299d0b507a26c70f465 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 29 Jan 2026 16:53:30 +0100 Subject: [PATCH 08/13] feat: adding thread ID to logs + filter in the client (#10150) --- codex-rs/core/src/codex.rs | 8 +- codex-rs/core/src/shell_snapshot.rs | 28 +++-- codex-rs/core/src/tasks/mod.rs | 41 ++++--- codex-rs/state/migrations/0002_logs.sql | 2 + codex-rs/state/src/bin/logs_client.rs | 20 +++- codex-rs/state/src/log_db.rs | 142 ++++++++++++++++++++++-- codex-rs/state/src/model/log.rs | 1 + codex-rs/state/src/runtime.rs | 3 +- 8 files changed, 203 insertions(+), 42 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index dbc2d5c632..8e1ec64370 100644 --- a/codex-rs/core/src/codex.rs +++ b/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, diff --git a/codex-rs/core/src/shell_snapshot.rs b/codex-rs/core/src/shell_snapshot.rs index 78581186c7..d6d738508a 100644 --- a/codex-rs/core/src/shell_snapshot.rs +++ b/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 { diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 08d23f7987..1f1195236c 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/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::info_span; use tracing::trace; use tracing::warn; @@ -130,25 +132,30 @@ 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 thread_id = self.conversation_id; + let session_span = info_span!("session_task", thread_id = %thread_id); + 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 diff --git a/codex-rs/state/migrations/0002_logs.sql b/codex-rs/state/migrations/0002_logs.sql index b9a2c681d4..051bef9136 100644 --- a/codex-rs/state/migrations/0002_logs.sql +++ b/codex-rs/state/migrations/0002_logs.sql @@ -5,9 +5,11 @@ CREATE TABLE logs ( level TEXT NOT NULL, target TEXT NOT NULL, message TEXT, + thread_id TEXT, module_path TEXT, file TEXT, line INTEGER ); CREATE INDEX idx_logs_ts ON logs(ts DESC, ts_nanos DESC, id DESC); +CREATE INDEX idx_logs_thread_id ON logs(thread_id); diff --git a/codex-rs/state/src/bin/logs_client.rs b/codex-rs/state/src/bin/logs_client.rs index 71acf67e85..df57856115 100644 --- a/codex-rs/state/src/bin/logs_client.rs +++ b/codex-rs/state/src/bin/logs_client.rs @@ -49,6 +49,10 @@ struct Args { #[arg(long)] file: Option, + /// Match a specific thread id. + #[arg(long)] + thread_id: Option, + /// Number of matching rows to show before tailing. #[arg(long, default_value_t = 200)] backfill: usize, @@ -65,6 +69,7 @@ struct LogRow { ts_nanos: i64, level: String, message: Option, + thread_id: Option, file: Option, line: Option, } @@ -76,6 +81,7 @@ struct LogFilter { to_ts: Option, module_like: Option, file_like: Option, + thread_id: Option, } #[tokio::main] @@ -139,6 +145,7 @@ fn build_filter(args: &Args) -> anyhow::Result { to_ts, module_like: args.module.clone(), file_like: args.file.clone(), + thread_id: args.thread_id.clone(), }) } @@ -236,7 +243,7 @@ async fn fetch_max_id(pool: &SqlitePool, filter: &LogFilter) -> anyhow::Result() -> QueryBuilder<'a, Sqlite> { QueryBuilder::::new( - "SELECT id, ts, ts_nanos, level, message, file, line FROM logs WHERE 1 = 1", + "SELECT id, ts, ts_nanos, level, message, thread_id, file, line FROM logs WHERE 1 = 1", ) } @@ -264,6 +271,11 @@ fn push_filters<'a>(builder: &mut QueryBuilder<'a, Sqlite>, filter: &'a LogFilte .push_bind(file_like.as_str()) .push(" || '%'"); } + if let Some(thread_id) = filter.thread_id.as_ref() { + builder + .push(" AND thread_id = ") + .push_bind(thread_id.as_str()); + } } fn format_row(row: &LogRow) -> String { @@ -277,9 +289,13 @@ fn format_row(row: &LogRow) -> String { 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.yellow().to_string(); let location_colored = location.dimmed().to_string(); let message_colored = message.bold().to_string(); - format!("{timestamp_colored} {level_colored} {location_colored} - {message_colored}") + format!( + "{timestamp_colored} {level_colored} [{thread_id_colored}] {location_colored} - {message_colored}" + ) } fn color_level(level: &str) -> String { diff --git a/codex-rs/state/src/log_db.rs b/codex-rs/state/src/log_db.rs index 13f19d0baf..a62870974c 100644 --- a/codex-rs/state/src/log_db.rs +++ b/codex-rs/state/src/log_db.rs @@ -26,6 +26,9 @@ 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; @@ -51,10 +54,55 @@ impl Layer for LogDbLayer where S: tracing::Subscriber + for<'a> LookupSpan<'a>, { - fn on_event(&self, event: &Event<'_>, _ctx: tracing_subscriber::layer::Context<'_, S>) { + 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::() { + 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) @@ -65,6 +113,7 @@ where 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), @@ -74,6 +123,75 @@ where } } +#[derive(Clone, Debug, Default)] +struct SpanLogContext { + thread_id: Option, +} + +#[derive(Default)] +struct SpanFieldVisitor { + thread_id: Option, +} + +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( + event: &Event<'_>, + ctx: &tracing_subscriber::layer::Context<'_, S>, +) -> Option +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::() + && log_context.thread_id.is_some() + { + thread_id = log_context.thread_id.clone(); + } + } + } + thread_id +} + async fn run_inserter( state_db: std::sync::Arc, mut receiver: mpsc::Receiver, @@ -114,42 +232,46 @@ async fn flush(state_db: &std::sync::Arc, buffer: &mut Vec, + thread_id: Option, } impl MessageVisitor { - fn record_message(&mut self, field: &Field, value: String) { + fn record_field(&mut self, field: &Field, value: String) { if field.name() == "message" && self.message.is_none() { - self.message = Some(value); + 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_message(field, value.to_string()); + self.record_field(field, value.to_string()); } fn record_u64(&mut self, field: &Field, value: u64) { - self.record_message(field, value.to_string()); + self.record_field(field, value.to_string()); } fn record_bool(&mut self, field: &Field, value: bool) { - self.record_message(field, value.to_string()); + self.record_field(field, value.to_string()); } fn record_f64(&mut self, field: &Field, value: f64) { - self.record_message(field, value.to_string()); + self.record_field(field, value.to_string()); } fn record_str(&mut self, field: &Field, value: &str) { - self.record_message(field, value.to_string()); + self.record_field(field, value.to_string()); } fn record_error(&mut self, field: &Field, value: &(dyn std::error::Error + 'static)) { - self.record_message(field, value.to_string()); + self.record_field(field, value.to_string()); } fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { - self.record_message(field, format!("{value:?}")); + self.record_field(field, format!("{value:?}")); } } diff --git a/codex-rs/state/src/model/log.rs b/codex-rs/state/src/model/log.rs index 0fa4c0b38e..c0c3a2dff2 100644 --- a/codex-rs/state/src/model/log.rs +++ b/codex-rs/state/src/model/log.rs @@ -7,6 +7,7 @@ pub struct LogEntry { pub level: String, pub target: String, pub message: Option, + pub thread_id: Option, pub module_path: Option, pub file: Option, pub line: Option, diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 96b2d69122..8dada32746 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -214,7 +214,7 @@ FROM threads } let mut builder = QueryBuilder::::new( - "INSERT INTO logs (ts, ts_nanos, level, target, message, module_path, file, line) ", + "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) @@ -222,6 +222,7 @@ FROM threads .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); From d6631fb5a99087dd0fb7c37620e1f32977cc5c7e Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 29 Jan 2026 16:55:01 +0100 Subject: [PATCH 09/13] feat: add log retention and delete them after 90 days (#10151) --- codex-rs/state/src/log_db.rs | 14 +++++++++++++- codex-rs/state/src/runtime.rs | 8 ++++++++ 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/codex-rs/state/src/log_db.rs b/codex-rs/state/src/log_db.rs index a62870974c..345e90c6a8 100644 --- a/codex-rs/state/src/log_db.rs +++ b/codex-rs/state/src/log_db.rs @@ -18,6 +18,8 @@ //! # } //! ``` +use chrono::Duration as ChronoDuration; +use chrono::Utc; use std::time::Duration; use std::time::SystemTime; use std::time::UNIX_EPOCH; @@ -38,6 +40,7 @@ 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, @@ -45,7 +48,8 @@ pub struct LogDbLayer { pub fn start(state_db: std::sync::Arc) -> LogDbLayer { let (sender, receiver) = mpsc::channel(LOG_QUEUE_CAPACITY); - tokio::spawn(run_inserter(state_db, receiver)); + tokio::spawn(run_inserter(std::sync::Arc::clone(&state_db), receiver)); + tokio::spawn(run_retention_cleanup(state_db)); LogDbLayer { sender } } @@ -229,6 +233,14 @@ async fn flush(state_db: &std::sync::Arc, buffer: &mut Vec) { + 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, diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 8dada32746..b61ea315c5 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -231,6 +231,14 @@ FROM threads Ok(()) } + pub(crate) async fn delete_logs_before(&self, cutoff_ts: i64) -> anyhow::Result { + let result = sqlx::query("DELETE FROM logs WHERE ts < ?") + .bind(cutoff_ts) + .execute(self.pool.as_ref()) + .await?; + Ok(result.rows_affected()) + } + /// List thread ids using the underlying database (no rollout scanning). pub async fn list_thread_ids( &self, From e6c4f548ab3072b6ae709b3da5aecfb074fe95d7 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 29 Jan 2026 17:28:15 +0100 Subject: [PATCH 10/13] chore: unify log queries (#10152) Unify log queries to only have SQLX code in the runtime and use it for both the log client and for tests --- .github/workflows/shell-tool-mcp.yml | 37 +---- codex-rs/core/tests/suite/sqlite_state.rs | 80 ++++++++++ codex-rs/state/migrations/0002_logs.sql | 2 - .../state/migrations/0003_logs_thread_id.sql | 3 + codex-rs/state/src/bin/logs_client.rs | 144 +++++------------- codex-rs/state/src/lib.rs | 2 + codex-rs/state/src/model/log.rs | 26 ++++ codex-rs/state/src/model/mod.rs | 2 + codex-rs/state/src/runtime.rs | 68 +++++++++ 9 files changed, 225 insertions(+), 139 deletions(-) create mode 100644 codex-rs/state/migrations/0003_logs_thread_id.sql diff --git a/.github/workflows/shell-tool-mcp.yml b/.github/workflows/shell-tool-mcp.yml index ec0c16cd5e..d6c85d9ddc 100644 --- a/.github/workflows/shell-tool-mcp.yml +++ b/.github/workflows/shell-tool-mcp.yml @@ -248,37 +248,8 @@ jobs: run: | set -euo pipefail if command -v apt-get >/dev/null 2>&1; then - # On Ubuntu ARM containers, apt metadata for universe/multiverse is large and - # can make `apt-get update` very slow. We only need packages from `main`, so - # trim sources to `main` for faster, more reliable updates. - if [[ "${{ matrix.target }}" == "aarch64-unknown-linux-musl" ]] && [[ -f /etc/os-release ]]; then - . /etc/os-release - if [[ "${ID-}" == "ubuntu" ]]; then - if [[ -f /etc/apt/sources.list.d/ubuntu.sources ]]; then - # Ubuntu 24.04+ uses deb822 sources; trim components to main. - sed -i -E 's/^Components:.*/Components: main/' /etc/apt/sources.list.d/ubuntu.sources - else - codename="${VERSION_CODENAME:-}" - if [[ -n "$codename" ]]; then - printf '%s\n' \ - "deb http://ports.ubuntu.com/ubuntu-ports ${codename} main" \ - "deb http://ports.ubuntu.com/ubuntu-ports ${codename}-updates main" \ - "deb http://ports.ubuntu.com/ubuntu-ports ${codename}-backports main" \ - "deb http://ports.ubuntu.com/ubuntu-ports ${codename}-security main" \ - >/etc/apt/sources.list - fi - fi - fi - fi - - apt_opts=( - -o Acquire::Retries=3 - -o Acquire::http::Timeout=30 - -o Acquire::https::Timeout=30 - -o Acquire::Languages=none - ) - apt-get "${apt_opts[@]}" update - DEBIAN_FRONTEND=noninteractive apt-get "${apt_opts[@]}" install -y git build-essential bison autoconf gettext + apt-get update + DEBIAN_FRONTEND=noninteractive apt-get install -y git build-essential bison autoconf gettext elif command -v dnf >/dev/null 2>&1; then dnf install -y git gcc gcc-c++ make bison autoconf gettext elif command -v yum >/dev/null 2>&1; then @@ -480,10 +451,6 @@ jobs: registry-url: https://registry.npmjs.org scope: "@openai" - # Trusted publishing requires npm CLI version 11.5.1 or later. - - name: Update npm - run: npm install -g npm@latest - - name: Download npm tarball uses: actions/download-artifact@v7 with: diff --git a/codex-rs/core/tests/suite/sqlite_state.rs b/codex-rs/core/tests/suite/sqlite_state.rs index df6fa2ef72..bf3b5a4f51 100644 --- a/codex-rs/core/tests/suite/sqlite_state.rs +++ b/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(()) +} diff --git a/codex-rs/state/migrations/0002_logs.sql b/codex-rs/state/migrations/0002_logs.sql index 051bef9136..b9a2c681d4 100644 --- a/codex-rs/state/migrations/0002_logs.sql +++ b/codex-rs/state/migrations/0002_logs.sql @@ -5,11 +5,9 @@ CREATE TABLE logs ( level TEXT NOT NULL, target TEXT NOT NULL, message TEXT, - thread_id TEXT, module_path TEXT, file TEXT, line INTEGER ); CREATE INDEX idx_logs_ts ON logs(ts DESC, ts_nanos DESC, id DESC); -CREATE INDEX idx_logs_thread_id ON logs(thread_id); diff --git a/codex-rs/state/migrations/0003_logs_thread_id.sql b/codex-rs/state/migrations/0003_logs_thread_id.sql new file mode 100644 index 0000000000..c4badb6885 --- /dev/null +++ b/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); diff --git a/codex-rs/state/src/bin/logs_client.rs b/codex-rs/state/src/bin/logs_client.rs index df57856115..c02a423253 100644 --- a/codex-rs/state/src/bin/logs_client.rs +++ b/codex-rs/state/src/bin/logs_client.rs @@ -1,4 +1,3 @@ -use std::path::Path; use std::path::PathBuf; use std::time::Duration; @@ -7,15 +6,12 @@ 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; -use sqlx::QueryBuilder; -use sqlx::Row; -use sqlx::Sqlite; -use sqlx::SqlitePool; -use sqlx::sqlite::SqliteConnectOptions; -use sqlx::sqlite::SqlitePoolOptions; #[derive(Debug, Parser)] #[command(name = "codex-state-logs")] @@ -62,18 +58,6 @@ struct Args { poll_ms: u64, } -#[derive(Debug, Clone, sqlx::FromRow)] -struct LogRow { - id: i64, - ts: i64, - ts_nanos: i64, - level: String, - message: Option, - thread_id: Option, - file: Option, - line: Option, -} - #[derive(Debug, Clone)] struct LogFilter { level_upper: Option, @@ -89,16 +73,20 @@ async fn main() -> anyhow::Result<()> { let args = Args::parse(); let db_path = resolve_db_path(&args)?; let filter = build_filter(&args)?; - let pool = open_read_only_pool(db_path.as_path()).await?; + 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(&pool, &filter, args.backfill).await?; + let mut last_id = print_backfill(runtime.as_ref(), &filter, args.backfill).await?; if last_id == 0 { - last_id = fetch_max_id(&pool, &filter).await?; + 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(&pool, &filter, last_id).await?; + 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)); @@ -159,23 +147,8 @@ fn parse_timestamp(value: &str) -> anyhow::Result { Ok(dt.timestamp()) } -async fn open_read_only_pool(path: &Path) -> anyhow::Result { - let options = SqliteConnectOptions::new() - .filename(path) - .create_if_missing(false) - .read_only(true) - .busy_timeout(Duration::from_secs(5)); - - let display = path.display(); - SqlitePoolOptions::new() - .max_connections(1) - .connect_with(options) - .await - .with_context(|| format!("failed to open sqlite db at {display}")) -} - async fn print_backfill( - pool: &SqlitePool, + runtime: &StateRuntime, filter: &LogFilter, backfill: usize, ) -> anyhow::Result { @@ -183,7 +156,7 @@ async fn print_backfill( return Ok(0); } - let mut rows = fetch_backfill(pool, filter, backfill).await?; + let mut rows = fetch_backfill(runtime, filter, backfill).await?; rows.reverse(); let mut last_id = 0; @@ -195,86 +168,53 @@ async fn print_backfill( } async fn fetch_backfill( - pool: &SqlitePool, + runtime: &StateRuntime, filter: &LogFilter, backfill: usize, ) -> anyhow::Result> { - let mut builder = base_select_builder(); - push_filters(&mut builder, filter); - builder.push(" ORDER BY id DESC"); - builder.push(" LIMIT ").push_bind(backfill as i64); - - builder - .build_query_as::() - .fetch_all(pool) + 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( - pool: &SqlitePool, + runtime: &StateRuntime, filter: &LogFilter, last_id: i64, ) -> anyhow::Result> { - let mut builder = base_select_builder(); - push_filters(&mut builder, filter); - builder.push(" AND id > ").push_bind(last_id); - builder.push(" ORDER BY id ASC"); - - builder - .build_query_as::() - .fetch_all(pool) + 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(pool: &SqlitePool, filter: &LogFilter) -> anyhow::Result { - let mut builder = QueryBuilder::::new("SELECT MAX(id) AS max_id FROM logs WHERE 1 = 1"); - push_filters(&mut builder, filter); - - let row = builder - .build() - .fetch_one(pool) +async fn fetch_max_id(runtime: &StateRuntime, filter: &LogFilter) -> anyhow::Result { + let query = to_log_query(filter, None, None, false); + runtime + .max_log_id(&query) .await - .context("failed to fetch max log id")?; - let max_id: Option = row.try_get("max_id")?; - Ok(max_id.unwrap_or(0)) + .context("failed to fetch max log id") } -fn base_select_builder<'a>() -> QueryBuilder<'a, Sqlite> { - QueryBuilder::::new( - "SELECT id, ts, ts_nanos, level, message, thread_id, file, line FROM logs WHERE 1 = 1", - ) -} - -fn push_filters<'a>(builder: &mut QueryBuilder<'a, Sqlite>, filter: &'a LogFilter) { - if let Some(level_upper) = filter.level_upper.as_ref() { - builder - .push(" AND UPPER(level) = ") - .push_bind(level_upper.as_str()); - } - if let Some(from_ts) = filter.from_ts { - builder.push(" AND ts >= ").push_bind(from_ts); - } - if let Some(to_ts) = filter.to_ts { - builder.push(" AND ts <= ").push_bind(to_ts); - } - if let Some(module_like) = filter.module_like.as_ref() { - builder - .push(" AND module_path LIKE '%' || ") - .push_bind(module_like.as_str()) - .push(" || '%'"); - } - if let Some(file_like) = filter.file_like.as_ref() { - builder - .push(" AND file LIKE '%' || ") - .push_bind(file_like.as_str()) - .push(" || '%'"); - } - if let Some(thread_id) = filter.thread_id.as_ref() { - builder - .push(" AND thread_id = ") - .push_bind(thread_id.as_str()); +fn to_log_query( + filter: &LogFilter, + limit: Option, + after_id: Option, + 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, } } diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index 91ebacea71..685552015e 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -12,6 +12,8 @@ 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; diff --git a/codex-rs/state/src/model/log.rs b/codex-rs/state/src/model/log.rs index c0c3a2dff2..d8a9ba3524 100644 --- a/codex-rs/state/src/model/log.rs +++ b/codex-rs/state/src/model/log.rs @@ -1,4 +1,5 @@ use serde::Serialize; +use sqlx::FromRow; #[derive(Clone, Debug, Serialize)] pub struct LogEntry { @@ -12,3 +13,28 @@ pub struct LogEntry { pub file: Option, pub line: Option, } + +#[derive(Clone, Debug, FromRow)] +pub struct LogRow { + pub id: i64, + pub ts: i64, + pub ts_nanos: i64, + pub level: String, + pub message: Option, + pub thread_id: Option, + pub file: Option, + pub line: Option, +} + +#[derive(Clone, Debug, Default)] +pub struct LogQuery { + pub level_upper: Option, + pub from_ts: Option, + pub to_ts: Option, + pub module_like: Option, + pub file_like: Option, + pub thread_id: Option, + pub after_id: Option, + pub limit: Option, + pub descending: bool, +} diff --git a/codex-rs/state/src/model/mod.rs b/codex-rs/state/src/model/mod.rs index 4921e2357a..bd615d7561 100644 --- a/codex-rs/state/src/model/mod.rs +++ b/codex-rs/state/src/model/mod.rs @@ -2,6 +2,8 @@ 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; diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index b61ea315c5..b15e4338bf 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -1,5 +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; @@ -239,6 +241,38 @@ FROM threads Ok(result.rows_affected()) } + /// Query logs with optional filters. + pub async fn query_logs(&self, query: &LogQuery) -> anyhow::Result> { + let mut builder = QueryBuilder::::new( + "SELECT id, ts, ts_nanos, level, 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::() + .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 { + let mut builder = + QueryBuilder::::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 = 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, @@ -411,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 { let options = SqliteConnectOptions::new() .filename(path) From 3e798c5a7dd0556947acc2baea1bb64be63ff3c7 Mon Sep 17 00:00:00 2001 From: Josh McKinney Date: Thu, 29 Jan 2026 09:34:59 -0800 Subject: [PATCH 11/13] Add OpenAI docs MCP tooltip (#10175) --- codex-rs/tui/tooltips.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/codex-rs/tui/tooltips.txt b/codex-rs/tui/tooltips.txt index 31819c6628..dc2d2faa67 100644 --- a/codex-rs/tui/tooltips.txt +++ b/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. From 798c4b3260e857756c2d6e13cc9bcd05ba604c7e Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 29 Jan 2026 19:15:22 +0100 Subject: [PATCH 12/13] feat: reduce span exposition (#10171) This only avoids the creation of duplicates spans --- codex-rs/core/src/stream_events_utils.rs | 7 ++++++- codex-rs/core/src/tasks/mod.rs | 5 ++--- codex-rs/state/src/bin/logs_client.rs | 12 ++++-------- codex-rs/state/src/model/log.rs | 1 + codex-rs/state/src/runtime.rs | 2 +- codex-rs/tui/src/lib.rs | 5 ++++- 6 files changed, 18 insertions(+), 14 deletions(-) diff --git a/codex-rs/core/src/stream_events_utils.rs b/codex-rs/core/src/stream_events_utils.rs index 197ff6b4b6..930ce5afd6 100644 --- a/codex-rs/core/src/stream_events_utils.rs +++ b/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)) diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 1f1195236c..a487041eba 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -14,7 +14,7 @@ use tokio::sync::Notify; use tokio_util::sync::CancellationToken; use tokio_util::task::AbortOnDropHandle; use tracing::Instrument; -use tracing::info_span; +use tracing::Span; use tracing::trace; use tracing::warn; @@ -132,8 +132,7 @@ impl Session { let ctx = Arc::clone(&turn_context); let task_for_run = Arc::clone(&task); let task_cancellation_token = cancellation_token.child_token(); - let thread_id = self.conversation_id; - let session_span = info_span!("session_task", thread_id = %thread_id); + let session_span = Span::current(); tokio::spawn( async move { let ctx_for_finish = Arc::clone(&ctx); diff --git a/codex-rs/state/src/bin/logs_client.rs b/codex-rs/state/src/bin/logs_client.rs index c02a423253..4389c29bc6 100644 --- a/codex-rs/state/src/bin/logs_client.rs +++ b/codex-rs/state/src/bin/logs_client.rs @@ -221,20 +221,16 @@ fn to_log_query( fn format_row(row: &LogRow) -> String { let timestamp = format_timestamp(row.ts, row.ts_nanos); let level = row.level.as_str(); - let location = match (&row.file, row.line) { - (Some(file), Some(line)) => format!("{file}:{line}"), - (Some(file), None) => file.clone(), - _ => "-".to_string(), - }; + 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.yellow().to_string(); - let location_colored = location.dimmed().to_string(); + 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}] {location_colored} - {message_colored}" + "{timestamp_colored} {level_colored} [{thread_id_colored}] {target_colored} - {message_colored}" ) } diff --git a/codex-rs/state/src/model/log.rs b/codex-rs/state/src/model/log.rs index d8a9ba3524..5d774249dd 100644 --- a/codex-rs/state/src/model/log.rs +++ b/codex-rs/state/src/model/log.rs @@ -20,6 +20,7 @@ pub struct LogRow { pub ts: i64, pub ts_nanos: i64, pub level: String, + pub target: String, pub message: Option, pub thread_id: Option, pub file: Option, diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index b15e4338bf..e0805e5b65 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -244,7 +244,7 @@ FROM threads /// Query logs with optional filters. pub async fn query_logs(&self, query: &LogQuery) -> anyhow::Result> { let mut builder = QueryBuilder::::new( - "SELECT id, ts, ts_nanos, level, message, thread_id, file, line FROM logs WHERE 1 = 1", + "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 { diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index c9e69b7125..010f8c83e5 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -306,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(); From feb992b6268c09d4cc33fc13571a2b1ce013fb63 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Thu, 29 Jan 2026 10:15:38 -0800 Subject: [PATCH 13/13] chore: ensure pnpm-workspace.yaml is up-to-date --- codex-cli/package-lock.json | 18 ------------------ pnpm-lock.yaml | 4 ++++ pnpm-workspace.yaml | 3 ++- 3 files changed, 6 insertions(+), 19 deletions(-) delete mode 100644 codex-cli/package-lock.json diff --git a/codex-cli/package-lock.json b/codex-cli/package-lock.json deleted file mode 100644 index 58ee846306..0000000000 --- a/codex-cli/package-lock.json +++ /dev/null @@ -1,18 +0,0 @@ -{ - "name": "@openai/codex", - "version": "0.0.0-dev", - "lockfileVersion": 3, - "packages": { - "": { - "name": "@openai/codex", - "version": "0.0.0-dev", - "license": "Apache-2.0", - "bin": { - "codex": "bin/codex.js" - }, - "engines": { - "node": ">=16" - } - } - } -} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 01112d6d3b..102869a683 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -17,6 +17,10 @@ importers: specifier: ^3.5.3 version: 3.5.3 + codex-cli: {} + + codex-rs/responses-api-proxy/npm: {} + sdk/typescript: devDependencies: '@modelcontextprotocol/sdk': diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 78899b9321..09a61536ab 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -1,5 +1,6 @@ packages: - - docs + - codex-cli + - codex-rs/responses-api-proxy/npm - sdk/typescript - shell-tool-mcp