mirror of
https://github.com/openai/codex.git
synced 2026-10-22 17:58:46 +00:00
Honor the configured SQLite home in the logs client (#35695)
## Why `just log` derived the logs database path from `CODEX_HOME`, so it could read the wrong database when `sqlite_home` or `CODEX_SQLITE_HOME` selected a different location. ## What changed - Move `logs_client` into `codex-cli` so it can resolve the shared `SqliteConfig` through the standard configuration loader. - Keep `--db` as a direct override that skips config loading and preserves native path bytes. - Update the `just log` recipes to run the client from its new crate. ## Testing - Add coverage for bypassing invalid Codex config with `--db`. - Add Unix coverage for non-UTF-8 database paths. GitOrigin-RevId: fabd64a66543be26a6f5d3b5e509016c3270350e
This commit is contained in:
Adam Perry @ OpenAI
committed by
copyberry
copyberry
parent
f029bb795c
commit
3418498f01
@@ -0,0 +1,467 @@
|
||||
use std::path::PathBuf;
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::Context;
|
||||
use chrono::DateTime;
|
||||
use clap::Parser;
|
||||
use clap::ValueEnum;
|
||||
use codex_core::config::ConfigBuilder;
|
||||
use codex_state::LogQuery;
|
||||
use codex_state::LogRow;
|
||||
use codex_state::SqliteConfig;
|
||||
use codex_state::StateRuntime;
|
||||
use codex_utils_absolute_path::AbsolutePathBuf;
|
||||
use owo_colors::OwoColorize;
|
||||
|
||||
#[derive(Debug, Parser)]
|
||||
#[command(name = "codex-state-logs")]
|
||||
#[command(about = "Tail Codex logs from the dedicated logs SQLite DB with simple filters")]
|
||||
struct Args {
|
||||
/// Path to CODEX_HOME. Defaults to $CODEX_HOME or ~/.codex.
|
||||
#[arg(long, env = "CODEX_HOME")]
|
||||
codex_home: Option<PathBuf>,
|
||||
|
||||
/// Direct path to the logs SQLite database. Overrides --codex-home.
|
||||
#[arg(long)]
|
||||
db: Option<PathBuf>,
|
||||
|
||||
/// Minimum log level to include.
|
||||
#[arg(long, value_enum, ignore_case = true)]
|
||||
level: Option<LogLevelThreshold>,
|
||||
|
||||
/// Start timestamp (RFC3339 or unix seconds).
|
||||
#[arg(long, value_name = "RFC3339|UNIX")]
|
||||
from: Option<String>,
|
||||
|
||||
/// End timestamp (RFC3339 or unix seconds).
|
||||
#[arg(long, value_name = "RFC3339|UNIX")]
|
||||
to: Option<String>,
|
||||
|
||||
/// Substring match on module_path. Repeat to include multiple substrings.
|
||||
#[arg(long = "module")]
|
||||
module: Vec<String>,
|
||||
|
||||
/// Substring match on file path. Repeat to include multiple substrings.
|
||||
#[arg(long = "file")]
|
||||
file: Vec<String>,
|
||||
|
||||
/// Match one or more thread ids. Repeat to include multiple threads.
|
||||
#[arg(long = "thread-id")]
|
||||
thread_id: Vec<String>,
|
||||
|
||||
/// Substring match against the rendered log body.
|
||||
#[arg(long)]
|
||||
search: Option<String>,
|
||||
|
||||
/// Include logs that do not have a thread id.
|
||||
#[arg(long)]
|
||||
threadless: bool,
|
||||
|
||||
/// 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,
|
||||
|
||||
/// Show compact output with only time, level, and rendered log body.
|
||||
#[arg(long)]
|
||||
compact: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
struct LogFilter {
|
||||
levels_upper: Vec<String>,
|
||||
from_ts: Option<i64>,
|
||||
to_ts: Option<i64>,
|
||||
module_like: Vec<String>,
|
||||
file_like: Vec<String>,
|
||||
thread_ids: Vec<String>,
|
||||
search: Option<String>,
|
||||
include_threadless: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
|
||||
enum LogLevelThreshold {
|
||||
Trace,
|
||||
Debug,
|
||||
Info,
|
||||
Warn,
|
||||
Error,
|
||||
}
|
||||
|
||||
impl LogLevelThreshold {
|
||||
fn levels_upper(self) -> Vec<String> {
|
||||
let levels = match self {
|
||||
LogLevelThreshold::Trace => &["TRACE", "DEBUG", "INFO", "WARN", "ERROR"][..],
|
||||
LogLevelThreshold::Debug => &["DEBUG", "INFO", "WARN", "ERROR"],
|
||||
LogLevelThreshold::Info => &["INFO", "WARN", "ERROR"],
|
||||
LogLevelThreshold::Warn => &["WARN", "ERROR"],
|
||||
LogLevelThreshold::Error => &["ERROR"],
|
||||
};
|
||||
levels.iter().map(ToString::to_string).collect()
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
let args = Args::parse();
|
||||
let sqlite = resolve_sqlite_config(&args).await?;
|
||||
let filter = build_filter(&args)?;
|
||||
let runtime = StateRuntime::init(sqlite, "logs-client".to_string()).await?;
|
||||
|
||||
let mut last_id =
|
||||
print_backfill(runtime.as_ref(), &filter, args.backfill, args.compact).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, args.compact));
|
||||
}
|
||||
tokio::time::sleep(poll_interval).await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn resolve_sqlite_config(args: &Args) -> anyhow::Result<SqliteConfig> {
|
||||
if let Some(db_path) = args.db.as_ref() {
|
||||
let sqlite_home = db_path
|
||||
.parent()
|
||||
.map(ToOwned::to_owned)
|
||||
.unwrap_or_else(|| PathBuf::from("."));
|
||||
return Ok(SqliteConfig::from_sqlite_home(
|
||||
AbsolutePathBuf::relative_to_current_dir(sqlite_home)?,
|
||||
));
|
||||
}
|
||||
|
||||
let mut config_builder = ConfigBuilder::default();
|
||||
if let Some(codex_home) = args.codex_home.as_ref() {
|
||||
config_builder = config_builder.codex_home(codex_home.clone());
|
||||
}
|
||||
let config = config_builder.build().await?;
|
||||
Ok(config.sqlite_config().clone())
|
||||
}
|
||||
|
||||
fn build_filter(args: &Args) -> anyhow::Result<LogFilter> {
|
||||
let from_ts = args
|
||||
.from
|
||||
.as_deref()
|
||||
.map(parse_timestamp)
|
||||
.transpose()
|
||||
.context("failed to parse --from")?;
|
||||
let to_ts = args
|
||||
.to
|
||||
.as_deref()
|
||||
.map(parse_timestamp)
|
||||
.transpose()
|
||||
.context("failed to parse --to")?;
|
||||
|
||||
let levels_upper = args
|
||||
.level
|
||||
.map_or_else(Vec::new, LogLevelThreshold::levels_upper);
|
||||
let module_like = args
|
||||
.module
|
||||
.iter()
|
||||
.filter(|module| !module.is_empty())
|
||||
.cloned()
|
||||
.collect::<Vec<_>>();
|
||||
let file_like = args
|
||||
.file
|
||||
.iter()
|
||||
.filter(|file| !file.is_empty())
|
||||
.cloned()
|
||||
.collect::<Vec<_>>();
|
||||
let thread_ids = args
|
||||
.thread_id
|
||||
.iter()
|
||||
.filter(|thread_id| !thread_id.is_empty())
|
||||
.cloned()
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
Ok(LogFilter {
|
||||
levels_upper,
|
||||
from_ts,
|
||||
to_ts,
|
||||
module_like,
|
||||
file_like,
|
||||
thread_ids,
|
||||
search: args.search.clone(),
|
||||
include_threadless: args.threadless,
|
||||
})
|
||||
}
|
||||
|
||||
fn parse_timestamp(value: &str) -> anyhow::Result<i64> {
|
||||
if let Ok(secs) = value.parse::<i64>() {
|
||||
return Ok(secs);
|
||||
}
|
||||
|
||||
let dt = DateTime::parse_from_rfc3339(value)
|
||||
.with_context(|| format!("expected RFC3339 or unix seconds, got {value}"))?;
|
||||
Ok(dt.timestamp())
|
||||
}
|
||||
|
||||
async fn print_backfill(
|
||||
runtime: &StateRuntime,
|
||||
filter: &LogFilter,
|
||||
backfill: usize,
|
||||
compact: bool,
|
||||
) -> anyhow::Result<i64> {
|
||||
if backfill == 0 {
|
||||
return Ok(0);
|
||||
}
|
||||
|
||||
let mut rows = fetch_backfill(runtime, filter, backfill).await?;
|
||||
rows.reverse();
|
||||
|
||||
let mut last_id = 0;
|
||||
for row in rows {
|
||||
last_id = last_id.max(row.id);
|
||||
println!("{}", format_row(&row, compact));
|
||||
}
|
||||
Ok(last_id)
|
||||
}
|
||||
|
||||
async fn fetch_backfill(
|
||||
runtime: &StateRuntime,
|
||||
filter: &LogFilter,
|
||||
backfill: usize,
|
||||
) -> anyhow::Result<Vec<LogRow>> {
|
||||
let query = to_log_query(
|
||||
filter,
|
||||
Some(backfill),
|
||||
/*after_id*/ None,
|
||||
/*descending*/ true,
|
||||
);
|
||||
runtime
|
||||
.query_logs(&query)
|
||||
.await
|
||||
.context("failed to fetch backfill logs")
|
||||
}
|
||||
|
||||
async fn fetch_new_rows(
|
||||
runtime: &StateRuntime,
|
||||
filter: &LogFilter,
|
||||
last_id: i64,
|
||||
) -> anyhow::Result<Vec<LogRow>> {
|
||||
let query = to_log_query(
|
||||
filter,
|
||||
/*limit*/ None,
|
||||
Some(last_id),
|
||||
/*descending*/ false,
|
||||
);
|
||||
runtime
|
||||
.query_logs(&query)
|
||||
.await
|
||||
.context("failed to fetch new logs")
|
||||
}
|
||||
|
||||
async fn fetch_max_id(runtime: &StateRuntime, filter: &LogFilter) -> anyhow::Result<i64> {
|
||||
let query = to_log_query(
|
||||
filter, /*limit*/ None, /*after_id*/ None, /*descending*/ false,
|
||||
);
|
||||
runtime
|
||||
.max_log_id(&query)
|
||||
.await
|
||||
.context("failed to fetch max log id")
|
||||
}
|
||||
|
||||
fn to_log_query(
|
||||
filter: &LogFilter,
|
||||
limit: Option<usize>,
|
||||
after_id: Option<i64>,
|
||||
descending: bool,
|
||||
) -> LogQuery {
|
||||
LogQuery {
|
||||
levels_upper: filter.levels_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_ids: filter.thread_ids.clone(),
|
||||
search: filter.search.clone(),
|
||||
include_threadless: filter.include_threadless,
|
||||
after_id,
|
||||
limit,
|
||||
descending,
|
||||
}
|
||||
}
|
||||
|
||||
fn format_row(row: &LogRow, compact: bool) -> String {
|
||||
let timestamp = formatter::ts(row.ts, row.ts_nanos, compact);
|
||||
let level = row.level.as_str();
|
||||
let target = row.target.as_str();
|
||||
let message = row.message.as_deref().unwrap_or("");
|
||||
let level_colored = formatter::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 = heuristic_formatting(message);
|
||||
if compact {
|
||||
format!("{timestamp_colored} {level_colored} {message_colored}")
|
||||
} else {
|
||||
format!(
|
||||
"{timestamp_colored} {level_colored} [{thread_id_colored}] {target_colored} - {message_colored}"
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
fn heuristic_formatting(message: &str) -> String {
|
||||
if matcher::apply_patch(message) {
|
||||
formatter::apply_patch(message)
|
||||
} else {
|
||||
message.bold().to_string()
|
||||
}
|
||||
}
|
||||
|
||||
mod matcher {
|
||||
pub(super) fn apply_patch(message: &str) -> bool {
|
||||
message.contains("ToolCall: apply_patch")
|
||||
}
|
||||
}
|
||||
|
||||
mod formatter {
|
||||
use chrono::DateTime;
|
||||
use chrono::SecondsFormat;
|
||||
use chrono::Utc;
|
||||
use owo_colors::OwoColorize;
|
||||
|
||||
pub(super) fn apply_patch(message: &str) -> String {
|
||||
message
|
||||
.lines()
|
||||
.map(|line| {
|
||||
if line.starts_with('+') {
|
||||
line.green().bold().to_string()
|
||||
} else if line.starts_with('-') {
|
||||
line.red().bold().to_string()
|
||||
} else {
|
||||
line.bold().to_string()
|
||||
}
|
||||
})
|
||||
.collect::<Vec<_>>()
|
||||
.join("\n")
|
||||
}
|
||||
|
||||
pub(super) fn ts(ts: i64, ts_nanos: i64, compact: bool) -> String {
|
||||
let nanos = u32::try_from(ts_nanos).unwrap_or(0);
|
||||
match DateTime::<Utc>::from_timestamp(ts, nanos) {
|
||||
Some(dt) if compact => dt.format("%H:%M:%S").to_string(),
|
||||
Some(dt) => dt.to_rfc3339_opts(SecondsFormat::Millis, true),
|
||||
None => format!("{ts}.{ts_nanos:09}Z"),
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn 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()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use pretty_assertions::assert_eq;
|
||||
use std::ffi::OsString;
|
||||
|
||||
#[test]
|
||||
fn log_level_threshold_includes_more_severe_levels() {
|
||||
assert_eq!(
|
||||
LogLevelThreshold::Warn.levels_upper(),
|
||||
vec!["WARN".to_string(), "ERROR".to_string()]
|
||||
);
|
||||
assert_eq!(
|
||||
LogLevelThreshold::Trace.levels_upper(),
|
||||
vec![
|
||||
"TRACE".to_string(),
|
||||
"DEBUG".to_string(),
|
||||
"INFO".to_string(),
|
||||
"WARN".to_string(),
|
||||
"ERROR".to_string(),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn log_level_rejects_aliases_and_unknown_values() {
|
||||
assert!(Args::try_parse_from(["codex-state-logs", "--level", "warning"]).is_err());
|
||||
assert!(Args::try_parse_from(["codex-state-logs", "--level", "err"]).is_err());
|
||||
assert!(Args::try_parse_from(["codex-state-logs", "--level", "warn,error"]).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn log_level_accepts_canonical_values_case_insensitively() {
|
||||
let args = Args::try_parse_from(["codex-state-logs", "--level", "WARN"])
|
||||
.expect("parse uppercase log level");
|
||||
|
||||
assert_eq!(args.level, Some(LogLevelThreshold::Warn));
|
||||
}
|
||||
|
||||
/// Explicit database selection must not parse an overridden Codex home.
|
||||
#[tokio::test]
|
||||
async fn direct_db_skips_codex_home_config() {
|
||||
let codex_home = tempfile::tempdir().expect("create Codex home");
|
||||
std::fs::write(codex_home.path().join("config.toml"), "model = [")
|
||||
.expect("write invalid config");
|
||||
let sqlite_home = tempfile::tempdir().expect("create SQLite home");
|
||||
let db_path = sqlite_home.path().join("logs_2.sqlite");
|
||||
let args = Args::try_parse_from([
|
||||
OsString::from("codex-state-logs"),
|
||||
OsString::from("--codex-home"),
|
||||
codex_home.path().as_os_str().to_owned(),
|
||||
OsString::from("--db"),
|
||||
db_path.as_os_str().to_owned(),
|
||||
])
|
||||
.expect("parse arguments");
|
||||
|
||||
let sqlite = resolve_sqlite_config(&args)
|
||||
.await
|
||||
.expect("resolve SQLite config");
|
||||
|
||||
assert_eq!(sqlite.logs_db_path(), db_path);
|
||||
}
|
||||
|
||||
/// Direct database selection must preserve native path bytes.
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn direct_db_preserves_non_utf8_path() {
|
||||
use std::os::unix::ffi::OsStrExt;
|
||||
use std::os::unix::ffi::OsStringExt;
|
||||
|
||||
let temp_dir = tempfile::tempdir().expect("create temp dir");
|
||||
let mut db_path = temp_dir.path().as_os_str().as_bytes().to_vec();
|
||||
db_path.extend_from_slice(b"/non-utf8-\xff/logs_2.sqlite");
|
||||
let db_path = PathBuf::from(OsString::from_vec(db_path));
|
||||
let args = Args::try_parse_from([
|
||||
OsString::from("codex-state-logs"),
|
||||
OsString::from("--db"),
|
||||
db_path.as_os_str().to_owned(),
|
||||
])
|
||||
.expect("parse arguments");
|
||||
|
||||
let sqlite = resolve_sqlite_config(&args)
|
||||
.await
|
||||
.expect("resolve SQLite config");
|
||||
|
||||
assert_eq!(sqlite.logs_db_path(), db_path);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user