diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 08cc850524..4c321b3f62 100644 --- a/codex-rs/Cargo.lock +++ b/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", 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/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/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/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/state_db.rs b/codex-rs/core/src/state_db.rs index d859328e1c..7990a017ef 100644 --- a/codex-rs/core/src/state_db.rs +++ b/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; -/// 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. @@ -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 { + 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/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 08d23f7987..a487041eba 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::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 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/Cargo.toml b/codex-rs/state/Cargo.toml index 810e250ca7..6019b5888b 100644 --- a/codex-rs/state/Cargo.toml +++ b/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] diff --git a/codex-rs/state/migrations/0002_logs.sql b/codex-rs/state/migrations/0002_logs.sql new file mode 100644 index 0000000000..b9a2c681d4 --- /dev/null +++ b/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); 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 new file mode 100644 index 0000000000..4389c29bc6 --- /dev/null +++ b/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, + + /// 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, + + /// 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, + + /// Poll interval in milliseconds. + #[arg(long, default_value_t = 500)] + poll_ms: u64, +} + +#[derive(Debug, Clone)] +struct LogFilter { + level_upper: Option, + from_ts: Option, + to_ts: Option, + module_like: Option, + file_like: Option, + thread_id: 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 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 { + 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(), + thread_id: args.thread_id.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 print_backfill( + runtime: &StateRuntime, + filter: &LogFilter, + backfill: usize, +) -> anyhow::Result { + 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> { + 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> { + 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 { + 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, + 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, + } +} + +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::::from_timestamp(ts, nanos) { + Some(dt) => dt.to_rfc3339_opts(SecondsFormat::Millis, true), + None => format!("{ts}.{ts_nanos:09}Z"), + } +} diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index 67533a3791..685552015e 100644 --- a/codex-rs/state/src/lib.rs +++ b/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; diff --git a/codex-rs/state/src/log_db.rs b/codex-rs/state/src/log_db.rs new file mode 100644 index 0000000000..345e90c6a8 --- /dev/null +++ b/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) { +//! 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, +} + +pub fn start(state_db: std::sync::Arc) -> 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 Layer 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::() { + 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, +} + +#[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, +) { + 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; +} + +async fn run_retention_cleanup(state_db: std::sync::Arc) { + 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, + thread_id: Option, +} + +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:?}")); + } +} diff --git a/codex-rs/state/src/model/log.rs b/codex-rs/state/src/model/log.rs new file mode 100644 index 0000000000..5d774249dd --- /dev/null +++ b/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, + pub thread_id: Option, + pub module_path: Option, + 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 target: 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 new file mode 100644 index 0000000000..bd615d7561 --- /dev/null +++ b/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; 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..e0805e5b65 100644 --- a/codex-rs/state/src/runtime.rs +++ b/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::::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 { + 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> { + let mut builder = QueryBuilder::::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::() + .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, @@ -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 { let options = SqliteConnectOptions::new() .filename(path) 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..010f8c83e5 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; @@ -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(); 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. 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 -- "$@"