Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions codex-rs/app-server/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ codex-protocol = { workspace = true }
codex-app-server-protocol = { workspace = true }
codex-feedback = { workspace = true }
codex-rmcp-client = { workspace = true }
codex-state = { workspace = true }
codex-utils-absolute-path = { workspace = true }
codex-utils-json-to-toml = { workspace = true }
chrono = { workspace = true }
Expand Down
30 changes: 30 additions & 0 deletions codex-rs/app-server/src/codex_message_processor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -272,6 +272,7 @@ use codex_protocol::protocol::USER_MESSAGE_BEGIN;
use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS;
use codex_protocol::user_input::UserInput as CoreInputItem;
use codex_rmcp_client::perform_oauth_login_return_url;
use codex_state::log_db::LogDbLayer;
use codex_utils_json_to_toml::json_to_toml;
use std::collections::HashMap;
use std::collections::HashSet;
Expand Down Expand Up @@ -378,6 +379,7 @@ pub(crate) struct CodexMessageProcessor {
pending_fuzzy_searches: Arc<Mutex<HashMap<String, Arc<AtomicBool>>>>,
fuzzy_search_sessions: Arc<Mutex<HashMap<String, FuzzyFileSearchSession>>>,
feedback: CodexFeedback,
log_db: Option<LogDbLayer>,
}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
Expand All @@ -397,6 +399,7 @@ pub(crate) struct CodexMessageProcessorArgs {
pub(crate) cloud_requirements: Arc<RwLock<CloudRequirementsLoader>>,
pub(crate) single_client_mode: bool,
pub(crate) feedback: CodexFeedback,
pub(crate) log_db: Option<LogDbLayer>,
}

impl CodexMessageProcessor {
Expand Down Expand Up @@ -434,6 +437,7 @@ impl CodexMessageProcessor {
cloud_requirements,
single_client_mode,
feedback,
log_db,
} = args;
Self {
auth_manager,
Expand All @@ -451,6 +455,7 @@ impl CodexMessageProcessor {
pending_fuzzy_searches: Arc::new(Mutex::new(HashMap::new())),
fuzzy_search_sessions: Arc::new(Mutex::new(HashMap::new())),
feedback,
log_db,
}
}

Expand Down Expand Up @@ -6717,6 +6722,30 @@ impl CodexMessageProcessor {

let snapshot = self.feedback.snapshot(conversation_id);
let thread_id = snapshot.thread_id.clone();
let sqlite_feedback_logs = if include_logs {
if let Some(log_db) = self.log_db.as_ref() {
log_db.flush().await;
}
let state_db_ctx = get_state_db(&self.config, None).await;
match (state_db_ctx.as_ref(), conversation_id) {
(Some(state_db_ctx), Some(conversation_id)) => {
let thread_id_text = conversation_id.to_string();
match state_db_ctx.query_feedback_logs(&thread_id_text).await {
Ok(logs) if logs.is_empty() => None,
Ok(logs) => Some(logs),
Comment thread
charley-oai marked this conversation as resolved.
Err(err) => {
warn!(
"failed to query feedback logs from sqlite for thread_id={thread_id_text}: {err}"
);
None
}
}
}
_ => None,
}
} else {
None
};

let validated_rollout_path = if include_logs {
match conversation_id {
Expand All @@ -6740,6 +6769,7 @@ impl CodexMessageProcessor {
include_logs,
&attachment_paths,
Some(session_source),
sqlite_feedback_logs,
)
})
.await;
Expand Down
21 changes: 21 additions & 0 deletions codex-rs/app-server/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -38,16 +38,20 @@ use codex_core::ExecPolicyError;
use codex_core::check_execpolicy_for_warnings;
use codex_core::config_loader::ConfigLoadError;
use codex_core::config_loader::TextRange as CoreTextRange;
use codex_core::features::Feature;
use codex_feedback::CodexFeedback;
use codex_state::log_db;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use toml::Value as TomlValue;
use tracing::Level;
use tracing::error;
use tracing::info;
use tracing::warn;
use tracing_subscriber::EnvFilter;
use tracing_subscriber::Layer;
use tracing_subscriber::filter::Targets;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::registry::Registry;
use tracing_subscriber::util::SubscriberInitExt;
Expand Down Expand Up @@ -475,13 +479,29 @@ pub async fn run_main_with_transport(

let feedback_layer = feedback.logger_layer();
let feedback_metadata_layer = feedback.metadata_layer();
let log_db = if config.features.enabled(Feature::Sqlite) {
codex_state::StateRuntime::init(
config.sqlite_home.clone(),
config.model_provider_id.clone(),
None,
)
.await
.ok()
.map(log_db::start)
} else {
None
};
let log_db_layer = log_db
.clone()
.map(|layer| layer.with_filter(Targets::new().with_default(Level::TRACE)));
let otel_logger_layer = otel.as_ref().and_then(|o| o.logger_layer());
let otel_tracing_layer = otel.as_ref().and_then(|o| o.tracing_layer());

let _ = tracing_subscriber::registry()
.with(stderr_fmt)
.with(feedback_layer)
.with(feedback_metadata_layer)
.with(log_db_layer)
.with(otel_logger_layer)
.with(otel_tracing_layer)
.try_init();
Expand Down Expand Up @@ -561,6 +581,7 @@ pub async fn run_main_with_transport(
loader_overrides,
cloud_requirements: cloud_requirements.clone(),
feedback: feedback.clone(),
log_db,
config_warnings,
});
let mut thread_created_rx = processor.thread_created_receiver();
Expand Down
4 changes: 4 additions & 0 deletions codex-rs/app-server/src/message_processor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ use codex_core::models_manager::collaboration_mode_presets::CollaborationModesCo
use codex_feedback::CodexFeedback;
use codex_protocol::ThreadId;
use codex_protocol::protocol::SessionSource;
use codex_state::log_db::LogDbLayer;
use futures::FutureExt;
use tokio::sync::broadcast;
use tokio::sync::watch;
Expand Down Expand Up @@ -152,6 +153,7 @@ pub(crate) struct MessageProcessorArgs {
pub(crate) loader_overrides: LoaderOverrides,
pub(crate) cloud_requirements: CloudRequirementsLoader,
pub(crate) feedback: CodexFeedback,
pub(crate) log_db: Option<LogDbLayer>,
pub(crate) config_warnings: Vec<ConfigWarningNotification>,
}

Expand All @@ -168,6 +170,7 @@ impl MessageProcessor {
loader_overrides,
cloud_requirements,
feedback,
log_db,
config_warnings,
} = args;
let auth_manager = AuthManager::shared(
Expand Down Expand Up @@ -201,6 +204,7 @@ impl MessageProcessor {
cloud_requirements: cloud_requirements.clone(),
single_client_mode,
feedback,
log_db,
});
let config_api = ConfigApi::new(
config.codex_home.clone(),
Expand Down
3 changes: 2 additions & 1 deletion codex-rs/feedback/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -226,6 +226,7 @@ impl CodexLogSnapshot {
include_logs: bool,
extra_log_files: &[PathBuf],
session_source: Option<SessionSource>,
logs_override: Option<Vec<u8>>,
) -> Result<()> {
use std::collections::BTreeMap;
use std::fs;
Expand Down Expand Up @@ -310,7 +311,7 @@ impl CodexLogSnapshot {

if include_logs {
envelope.add_item(EnvelopeItem::Attachment(Attachment {
buffer: self.bytes.clone(),
buffer: logs_override.unwrap_or_else(|| self.bytes.clone()),
filename: String::from("codex-logs.log"),
content_type: Some("text/plain".to_string()),
ty: None,
Expand Down
Loading
Loading