merge local main into code-branch-20250926-200929 before merge

This commit is contained in:
James Peter
2025-09-27 08:49:59 +10:00
9 changed files with 725 additions and 81 deletions
+35 -4
View File
@@ -22,6 +22,8 @@ use codex_core::find_conversation_path_by_id_str;
use codex_core::RolloutRecorder;
use codex_exec::Cli as ExecCli;
use codex_tui::Cli as TuiCli;
use codex_tui::ExitSummary;
use codex_tui::RESUME_COMMAND_NAME;
use std::path::Path;
use std::path::PathBuf;
use std::process;
@@ -257,9 +259,22 @@ async fn cli_main(codex_linux_sandbox_exe: Option<PathBuf>) -> anyhow::Result<()
&mut interactive.config_overrides,
root_config_overrides.clone(),
);
let usage = codex_tui::run_main(interactive, codex_linux_sandbox_exe).await?;
if !usage.is_zero() {
println!("{}", codex_core::protocol::FinalOutput::from(usage));
let ExitSummary {
token_usage,
session_id,
} = codex_tui::run_main(interactive, codex_linux_sandbox_exe).await?;
if !token_usage.is_zero() {
println!(
"{}",
codex_core::protocol::FinalOutput::from(token_usage.clone())
);
}
if let Some(session_id) = session_id {
println!(
"To continue this session, run {} resume {}.",
RESUME_COMMAND_NAME,
session_id
);
}
}
Some(Subcommand::Exec(mut exec_cli)) => {
@@ -287,7 +302,23 @@ async fn cli_main(codex_linux_sandbox_exe: Option<PathBuf>) -> anyhow::Result<()
last,
config_overrides,
);
codex_tui::run_main(interactive, codex_linux_sandbox_exe).await?;
let ExitSummary {
token_usage,
session_id,
} = codex_tui::run_main(interactive, codex_linux_sandbox_exe).await?;
if !token_usage.is_zero() {
println!(
"{}",
codex_core::protocol::FinalOutput::from(token_usage.clone())
);
}
if let Some(session_id) = session_id {
println!(
"To continue this session, run {} resume {}.",
RESUME_COMMAND_NAME,
session_id
);
}
}
Some(Subcommand::Login(mut login_cli)) => {
prepend_config_flags(
+284 -20
View File
@@ -14,6 +14,7 @@ const USAGE_VERSION: u32 = 1;
const USAGE_SUBDIR: &str = "usage";
const HOURLY_HISTORY_DAYS: i64 = 183; // retain ~6 months of hourly usage for history views
const UNKNOWN_RESET_RELOG_INTERVAL: Duration = Duration::hours(24);
const RESET_PASSED_TOLERANCE: Duration = Duration::seconds(5);
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub enum RateLimitWarningScope {
@@ -375,18 +376,6 @@ pub fn record_usage_limit_hint(
})
}
fn reset_changed(prev: Option<DateTime<Utc>>, next: Option<DateTime<Utc>>) -> bool {
match (prev, next) {
(Some(a), Some(b)) => a
.signed_duration_since(b)
.num_seconds()
.abs()
> 5,
(None, None) => false,
_ => true,
}
}
fn record_threshold_log(
logs: &mut Vec<RateLimitWarningRecord>,
threshold: f64,
@@ -396,20 +385,49 @@ fn record_threshold_log(
if let Some(existing) = logs.iter_mut().find(|entry| {
(entry.threshold - threshold).abs() < f64::EPSILON
}) {
if reset_changed(existing.reset_at, reset_at) {
existing.logged_at = None;
} else if reset_at.is_none()
&& existing
.logged_at
.is_some_and(|logged| observed_at.signed_duration_since(logged) >= UNKNOWN_RESET_RELOG_INTERVAL)
{
let previous_reset = existing.reset_at;
let previous_logged = existing.logged_at;
let new_reset = reset_at;
let reset_moved_earlier = match (previous_reset, new_reset) {
(Some(prev), Some(next)) => next + RESET_PASSED_TOLERANCE < prev,
_ => false,
};
let logged_after_prev_reset = match (previous_logged, previous_reset) {
(Some(logged), Some(prev)) => logged >= prev,
_ => false,
};
let prev_reset_elapsed = previous_reset
.map(|prev| observed_at + RESET_PASSED_TOLERANCE >= prev)
.unwrap_or(false);
let unknown_reset_elapsed = (previous_reset.is_none() || new_reset.is_none())
&& previous_logged
.is_some_and(|logged| observed_at.signed_duration_since(logged) >= UNKNOWN_RESET_RELOG_INTERVAL);
let mut should_clear = false;
if reset_moved_earlier {
should_clear = true;
} else if prev_reset_elapsed && !logged_after_prev_reset {
should_clear = true;
} else if unknown_reset_elapsed {
should_clear = true;
}
existing.reset_at = new_reset;
if should_clear {
existing.logged_at = None;
}
existing.reset_at = reset_at;
if existing.logged_at.is_none() {
existing.logged_at = Some(observed_at);
return true;
}
return false;
}
@@ -537,6 +555,25 @@ pub fn load_account_usage(
#[cfg(test)]
mod tests {
//! Regression coverage for rate-limit warning relogging.
//!
//! These cases enforce the desired behaviour:
//! - **No duplicate within a window**: once a threshold logs, subsequent polls before
//! the stored reset timestamp must remain silent even if the backend repeats or
//! extends the reset time.
//! - **Relog after reset passes**: the first poll at or after the recorded reset may
//! emit again, regardless of whether the backend has already advanced the window.
//! - **Relog on earlier reset**: if the backend moves the reset earlier (window
//! shrinks), we allow an immediate relog even before the previously stored reset.
//! - **Unknown reset fallback**: when reset metadata disappears, we rely on the
//! 24-hour `UNKNOWN_RESET_RELOG_INTERVAL` to unblock further warnings. When the
//! backend begins reporting timestamps again, we should also allow a relog provided
//! the fallback window has elapsed.
//! - **Missing metadata alone is not enough**: before the fallback timer elapses we
//! must keep warnings muted even if new snapshots omit reset times.
//!
//! The helper tests below construct scenarios targeting each rule so the state
//! machine in `record_threshold_log` can be refactored confidently.
use super::*;
use std::fs::File;
use crate::protocol::TokenUsage;
@@ -552,6 +589,233 @@ mod tests {
}
}
#[test]
fn rate_limit_warning_only_logs_once_per_reset() {
let home = TempDir::new().expect("tempdir");
let now = Utc::now();
let reset_at = now + Duration::days(7);
let first = record_rate_limit_warning(
home.path(),
"acct-1",
Some("Team"),
RateLimitWarningScope::Secondary,
75.0,
Some(reset_at),
now,
"Secondary usage exceeded 75% of the limit. Run /limits for detailed usage.",
)
.expect("first record succeeds");
assert!(first, "first logging should emit");
let second = record_rate_limit_warning(
home.path(),
"acct-1",
Some("Team"),
RateLimitWarningScope::Secondary,
75.0,
Some(now + Duration::days(7) + Duration::hours(6)),
now + Duration::hours(6),
"Secondary usage exceeded 75% of the limit. Run /limits for detailed usage.",
)
.expect("second record succeeds");
assert!(!second, "duplicate logging before reset should be suppressed");
let third = record_rate_limit_warning(
home.path(),
"acct-1",
Some("Team"),
RateLimitWarningScope::Secondary,
75.0,
Some(now + Duration::days(15)),
now + Duration::days(8),
"Secondary usage exceeded 75% of the limit. Run /limits for detailed usage.",
)
.expect("third record succeeds");
assert!(third, "after reset passes we should emit again");
}
#[test]
fn rate_limit_warning_relogs_after_reset_with_new_timestamp() {
let home = TempDir::new().expect("tempdir");
let now = Utc::now();
let msg = "Secondary usage exceeded 75% of the limit. Run /limits for detailed usage.";
let first = record_rate_limit_warning(
home.path(),
"acct-1",
Some("Team"),
RateLimitWarningScope::Secondary,
75.0,
Some(now + Duration::hours(1)),
now,
msg,
)
.expect("first record succeeds");
assert!(first);
// Backend extends reset window beyond the old reset time; we should re-emit now that
// the prior window has expired and a new one started.
let second = record_rate_limit_warning(
home.path(),
"acct-1",
Some("Team"),
RateLimitWarningScope::Secondary,
75.0,
Some(now + Duration::hours(2)),
now + Duration::minutes(65),
msg,
)
.expect("second record succeeds");
assert!(second, "after reset we should log again even if next window is later");
// Subsequent updates inside the new window should remain suppressed until that reset passes.
let third = record_rate_limit_warning(
home.path(),
"acct-1",
Some("Team"),
RateLimitWarningScope::Secondary,
75.0,
Some(now + Duration::hours(2)),
now + Duration::minutes(70),
msg,
)
.expect("third record succeeds");
assert!(!third, "duplicate logging inside the same window should stay muted");
}
#[test]
fn rate_limit_warning_relogs_after_reset_even_if_logged_just_before() {
let home = TempDir::new().expect("tempdir");
let now = Utc::now();
let reset_at = now + Duration::minutes(1);
let msg = "Secondary usage exceeded 75% of the limit. Run /limits for detailed usage.";
let first = record_rate_limit_warning(
home.path(),
"acct-1",
Some("Team"),
RateLimitWarningScope::Secondary,
75.0,
Some(reset_at),
reset_at - Duration::seconds(3),
msg,
)
.expect("first record succeeds");
assert!(first);
// After the reset passes, with a new window scheduled further out, we should relog.
let second = record_rate_limit_warning(
home.path(),
"acct-1",
Some("Team"),
RateLimitWarningScope::Secondary,
75.0,
Some(reset_at + Duration::hours(1)),
reset_at + Duration::seconds(45),
msg,
)
.expect("second record succeeds");
assert!(second, "post-reset poll should emit again even if prior log was moments before reset");
}
#[test]
fn rate_limit_warning_relogs_after_unknown_reset_interval() {
let home = TempDir::new().expect("tempdir");
let now = Utc::now();
let msg = "Secondary usage exceeded 75% of the limit. Run /limits for detailed usage.";
let first = record_rate_limit_warning(
home.path(),
"acct-1",
Some("Team"),
RateLimitWarningScope::Secondary,
75.0,
Some(now + Duration::hours(1)),
now,
msg,
)
.expect("first record succeeds");
assert!(first);
// Backend stops providing reset info — still within backoff window.
let second = record_rate_limit_warning(
home.path(),
"acct-1",
Some("Team"),
RateLimitWarningScope::Secondary,
75.0,
None,
now + Duration::minutes(20),
msg,
)
.expect("second record succeeds");
assert!(!second, "dropping reset info should keep warning muted initially");
// After the unknown interval we should allow another log.
let third = record_rate_limit_warning(
home.path(),
"acct-1",
Some("Team"),
RateLimitWarningScope::Secondary,
75.0,
None,
now + Duration::hours(25),
msg,
)
.expect("third record succeeds");
assert!(third, "after backoff expires we should re-emit");
}
#[test]
fn rate_limit_warning_relogs_when_reset_info_returns() {
let home = TempDir::new().expect("tempdir");
let now = Utc::now();
let msg = "Secondary usage exceeded 75% of the limit. Run /limits for detailed usage.";
let first = record_rate_limit_warning(
home.path(),
"acct-1",
Some("Team"),
RateLimitWarningScope::Secondary,
75.0,
Some(now + Duration::hours(1)),
now,
msg,
)
.expect("first record succeeds");
assert!(first);
let second = record_rate_limit_warning(
home.path(),
"acct-1",
Some("Team"),
RateLimitWarningScope::Secondary,
75.0,
None,
now + Duration::minutes(20),
msg,
)
.expect("second record succeeds");
assert!(!second);
let third = record_rate_limit_warning(
home.path(),
"acct-1",
Some("Team"),
RateLimitWarningScope::Secondary,
75.0,
Some(now + Duration::hours(30)),
now + Duration::hours(25),
msg,
)
.expect("third record succeeds");
assert!(third, "restored reset metadata after fallback window should re-log");
}
#[test]
fn creates_usage_file_and_accumulates_tokens() {
let home = TempDir::new().expect("tempdir");
+12
View File
@@ -2456,6 +2456,11 @@ impl App<'_> {
// Schedule the next redraw with the requested duration
self.schedule_redraw_in(duration);
}
AppEvent::GhostSnapshotFinished { job_id, result, elapsed } => {
if let AppState::Chat { widget } = &mut self.app_state {
widget.handle_ghost_snapshot_finished(job_id, result, elapsed);
}
}
}
}
if self.alt_screen_active {
@@ -2556,6 +2561,13 @@ impl App<'_> {
usage
}
pub(crate) fn session_id(&self) -> Option<uuid::Uuid> {
match &self.app_state {
AppState::Chat { widget } => widget.session_id(),
AppState::Onboarding { .. } => None,
}
}
/// Return a human-readable performance summary if timing was enabled.
pub(crate) fn perf_summary(&self) -> Option<String> {
if !self.timing_enabled {
+9
View File
@@ -13,6 +13,8 @@ use ratatui::text::Line;
use crate::streaming::StreamKind;
use std::time::Duration;
use codex_git_tooling::{GhostCommit, GitToolingError};
use crate::app::ChatWidgetArgs;
use crate::bottom_pane::chrome_selection_view::ChromeLaunchOption;
use crate::slash_command::SlashCommand;
@@ -92,6 +94,13 @@ pub(crate) enum AppEvent {
/// Multiple requests are coalesced by the central frame scheduler.
ScheduleFrameIn(Duration),
/// Background ghost snapshot job finished (success or failure).
GhostSnapshotFinished {
job_id: u64,
result: Result<GhostCommit, GitToolingError>,
elapsed: Duration,
},
/// Internal: flush any pending out-of-order ExecEnd events that did not
/// receive a matching ExecBegin within a short pairing window. This lets
/// the TUI render a fallback "Ran call_<id>" cell so output is not lost.
+22
View File
@@ -242,6 +242,10 @@ impl TextArea {
KeyEvent { code: KeyCode::Char('\u{0006}'), modifiers: KeyModifiers::NONE, .. } /* ^F */ => {
self.move_cursor_right();
}
KeyEvent {
code: KeyCode::Char(c),
..
} if matches!(c, '\n' | '\r') => self.insert_str("\n"),
KeyEvent {
code: KeyCode::Char(c),
// Insert plain characters (and Shift-modified). Do NOT insert when ALT is held,
@@ -321,6 +325,13 @@ impl TextArea {
} => {
self.undo();
}
KeyEvent {
code: KeyCode::Char('z'),
modifiers: KeyModifiers::SUPER,
..
} => {
self.undo();
}
// macOS-like shortcuts (when terminals report the Command key as SUPER):
// Cmd+Left -> move to beginning of line
// Cmd+Right -> move to end of line
@@ -1275,6 +1286,17 @@ mod tests {
assert_eq!(t.cursor(), 0);
}
#[test]
fn alt_enter_inserts_newline() {
let mut t = TextArea::new();
t.input(KeyEvent::new(KeyCode::Enter, KeyModifiers::ALT));
assert_eq!(t.text(), "\n");
let mut t = TextArea::new();
t.input(KeyEvent::new(KeyCode::Char('\n'), KeyModifiers::ALT));
assert_eq!(t.text(), "\n");
}
#[test]
fn control_h_backspace() {
// Test Ctrl+H as backspace
+328 -48
View File
@@ -2,6 +2,7 @@ use once_cell::sync::Lazy;
use std::collections::HashMap;
use std::collections::HashSet;
use std::collections::VecDeque;
use std::io;
use std::path::PathBuf;
use std::rc::{Rc, Weak};
use std::sync::Arc;
@@ -333,10 +334,14 @@ impl RateLimitWarningState {
) -> Vec<RateLimitWarning> {
let mut warnings = Vec::new();
while self.weekly_index < RATE_LIMIT_WARNING_THRESHOLDS.len()
&& secondary_used_percent >= RATE_LIMIT_WARNING_THRESHOLDS[self.weekly_index]
let mut next_weekly_index = self.weekly_index;
while next_weekly_index < RATE_LIMIT_WARNING_THRESHOLDS.len()
&& secondary_used_percent >= RATE_LIMIT_WARNING_THRESHOLDS[next_weekly_index]
{
let threshold = RATE_LIMIT_WARNING_THRESHOLDS[self.weekly_index];
next_weekly_index += 1;
}
if next_weekly_index > self.weekly_index {
let threshold = RATE_LIMIT_WARNING_THRESHOLDS[next_weekly_index - 1];
warnings.push(RateLimitWarning {
scope: RateLimitWarningScope::Secondary,
threshold,
@@ -344,13 +349,17 @@ impl RateLimitWarningState {
"Secondary usage exceeded {threshold:.0}% of the limit. Run /limits for detailed usage."
),
});
self.weekly_index += 1;
self.weekly_index = next_weekly_index;
}
while self.hourly_index < RATE_LIMIT_WARNING_THRESHOLDS.len()
&& primary_used_percent >= RATE_LIMIT_WARNING_THRESHOLDS[self.hourly_index]
let mut next_hourly_index = self.hourly_index;
while next_hourly_index < RATE_LIMIT_WARNING_THRESHOLDS.len()
&& primary_used_percent >= RATE_LIMIT_WARNING_THRESHOLDS[next_hourly_index]
{
let threshold = RATE_LIMIT_WARNING_THRESHOLDS[self.hourly_index];
next_hourly_index += 1;
}
if next_hourly_index > self.hourly_index {
let threshold = RATE_LIMIT_WARNING_THRESHOLDS[next_hourly_index - 1];
warnings.push(RateLimitWarning {
scope: RateLimitWarningScope::Primary,
threshold,
@@ -358,7 +367,7 @@ impl RateLimitWarningState {
"Hourly usage exceeded {threshold:.0}% of the limit. Run /limits for detailed usage."
),
});
self.hourly_index += 1;
self.hourly_index = next_hourly_index;
}
warnings
@@ -556,6 +565,10 @@ pub(crate) struct ChatWidget<'a> {
ghost_snapshots: Vec<GhostSnapshot>,
ghost_snapshots_disabled: bool,
ghost_snapshots_disabled_reason: Option<GhostSnapshotsDisabledReason>,
ghost_snapshot_queue: VecDeque<(u64, GhostSnapshotRequest)>,
active_ghost_snapshot: Option<(u64, GhostSnapshotRequest)>,
next_ghost_snapshot_id: u64,
pending_snapshot_dispatches: VecDeque<PendingSnapshotDispatch>,
// Event sequencing to preserve original order across streaming/tool events
// and stream-related flags moved into stream_state
@@ -670,6 +683,43 @@ impl GhostSnapshot {
}
}
#[derive(Clone)]
struct GhostSnapshotRequest {
summary: Option<String>,
conversation: ConversationSnapshot,
started_at: Instant,
}
impl GhostSnapshotRequest {
fn new(summary: Option<String>, conversation: ConversationSnapshot) -> Self {
Self {
summary,
conversation,
started_at: Instant::now(),
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum GhostSnapshotJobHandle {
Scheduled(u64),
Skipped,
}
enum PendingSnapshotDispatch {
Direct { job_id: u64, batch: Vec<UserMessage> },
Queued { job_id: u64, message: UserMessage },
}
impl PendingSnapshotDispatch {
fn job_id(&self) -> u64 {
match self {
PendingSnapshotDispatch::Direct { job_id, .. } |
PendingSnapshotDispatch::Queued { job_id, .. } => *job_id,
}
}
}
#[derive(Default)]
struct GitBranchCache {
value: Option<String>,
@@ -2723,6 +2773,10 @@ impl ChatWidget<'_> {
ghost_snapshots: Vec::new(),
ghost_snapshots_disabled: false,
ghost_snapshots_disabled_reason: None,
ghost_snapshot_queue: VecDeque::new(),
active_ghost_snapshot: None,
next_ghost_snapshot_id: 0,
pending_snapshot_dispatches: VecDeque::new(),
browser_is_external: false,
// Stable ordering & routing init
cell_order_seq: vec![OrderKey {
@@ -2955,6 +3009,10 @@ impl ChatWidget<'_> {
ghost_snapshots: Vec::new(),
ghost_snapshots_disabled: false,
ghost_snapshots_disabled_reason: None,
ghost_snapshot_queue: VecDeque::new(),
active_ghost_snapshot: None,
next_ghost_snapshot_id: 0,
pending_snapshot_dispatches: VecDeque::new(),
browser_is_external: false,
// Strict ordering init for forked widget
cell_order_seq: vec![OrderKey {
@@ -4887,13 +4945,6 @@ impl ChatWidget<'_> {
return;
}
let prompt_summary = if message.display_text.trim().is_empty() {
None
} else {
Some(message.display_text.clone())
};
self.capture_ghost_snapshot(prompt_summary);
let turn_active = self.is_task_running()
|| !self.active_task_ids.is_empty()
|| self.stream.is_write_cycle_active()
@@ -4904,54 +4955,216 @@ impl ChatWidget<'_> {
"Queuing user input while turn is active (queued: {})",
self.queued_user_messages.len() + 1
);
self.queued_user_messages.push_back(message);
let queued_clone = message.clone();
self.queued_user_messages.push_back(queued_clone);
self.refresh_queued_user_messages();
let queue_items = self
.queued_user_messages
.back()
.map(|msg| msg.ordered_items.clone())
.unwrap_or_default();
let prompt_summary = if message.display_text.trim().is_empty() {
None
} else {
Some(message.display_text.clone())
};
match self.codex_op_tx.send(Op::QueueUserInput { items: queue_items }) {
Ok(()) => {
if let Some(sent_message) = self.queued_user_messages.pop_back() {
self.refresh_queued_user_messages();
self.finalize_sent_user_message(sent_message);
}
match self.capture_ghost_snapshot(prompt_summary) {
GhostSnapshotJobHandle::Scheduled(job_id) => {
self.pending_snapshot_dispatches
.push_back(PendingSnapshotDispatch::Queued { job_id, message });
}
Err(e) => {
tracing::error!("failed to send QueueUserInput op: {e}");
GhostSnapshotJobHandle::Skipped => {
self.dispatch_queued_user_message_now(message);
}
}
return;
}
let mut batch: Vec<UserMessage> = self.queued_user_messages.drain(..).collect();
batch.push(message);
self.refresh_queued_user_messages();
self.send_user_messages_to_agent(batch);
let summary = batch
.last()
.and_then(|msg| {
let trimmed = msg.display_text.trim();
if trimmed.is_empty() {
None
} else {
Some(msg.display_text.clone())
}
});
match self.capture_ghost_snapshot(summary) {
GhostSnapshotJobHandle::Scheduled(job_id) => {
self.pending_snapshot_dispatches
.push_back(PendingSnapshotDispatch::Direct { job_id, batch });
}
GhostSnapshotJobHandle::Skipped => {
self.send_user_messages_to_agent(batch);
}
}
// (debug watchdog removed)
}
fn capture_ghost_snapshot(&mut self, summary: Option<String>) {
fn capture_ghost_snapshot(&mut self, summary: Option<String>) -> GhostSnapshotJobHandle {
if self.ghost_snapshots_disabled {
return;
return GhostSnapshotJobHandle::Skipped;
}
let conversation = self.current_conversation_snapshot();
let options = CreateGhostCommitOptions::new(&self.config.cwd);
match create_ghost_commit(&options) {
let request = GhostSnapshotRequest::new(summary, self.current_conversation_snapshot());
self.enqueue_ghost_snapshot(request)
}
fn capture_ghost_snapshot_blocking(&mut self, summary: Option<String>) -> Option<GhostSnapshot> {
if self.ghost_snapshots_disabled {
return None;
}
let request = GhostSnapshotRequest::new(summary, self.current_conversation_snapshot());
let repo_path = self.config.cwd.clone();
let started_at = request.started_at;
let result = create_ghost_commit(&CreateGhostCommitOptions::new(repo_path.as_path()));
let elapsed = started_at.elapsed();
let snapshot = self.finalize_ghost_snapshot(request, result, elapsed);
snapshot
}
fn dispatch_pending_snapshot(&mut self, job_id: u64) {
let Some(position) = self
.pending_snapshot_dispatches
.iter()
.position(|pending| pending.job_id() == job_id)
else {
return;
};
let Some(pending) = self.pending_snapshot_dispatches.remove(position) else {
return;
};
self.process_snapshot_dispatch(pending);
}
fn flush_pending_snapshot_dispatches(&mut self) {
while let Some(pending) = self.pending_snapshot_dispatches.pop_front() {
self.process_snapshot_dispatch(pending);
}
}
fn process_snapshot_dispatch(&mut self, pending: PendingSnapshotDispatch) {
match pending {
PendingSnapshotDispatch::Direct { batch, .. } => {
self.send_user_messages_to_agent(batch);
}
PendingSnapshotDispatch::Queued { message, .. } => {
self.dispatch_queued_user_message_now(message);
}
}
}
fn dispatch_queued_user_message_now(&mut self, message: UserMessage) {
let message = self.take_queued_user_message(&message).unwrap_or(message);
let items = message.ordered_items.clone();
match self.codex_op_tx.send(Op::QueueUserInput { items }) {
Ok(()) => {
self.finalize_sent_user_message(message);
}
Err(err) => {
tracing::error!("failed to send QueueUserInput op: {err}");
self.queued_user_messages.push_front(message);
self.refresh_queued_user_messages();
}
}
}
fn take_queued_user_message(&mut self, target: &UserMessage) -> Option<UserMessage> {
let position = self
.queued_user_messages
.iter()
.position(|message| message == target)?;
let removed = self.queued_user_messages.remove(position)?;
self.refresh_queued_user_messages();
Some(removed)
}
fn enqueue_ghost_snapshot(&mut self, request: GhostSnapshotRequest) -> GhostSnapshotJobHandle {
let job_id = self.next_ghost_snapshot_id;
self.next_ghost_snapshot_id = self.next_ghost_snapshot_id.wrapping_add(1);
self.ghost_snapshot_queue.push_back((job_id, request));
self.spawn_next_ghost_snapshot();
GhostSnapshotJobHandle::Scheduled(job_id)
}
fn spawn_next_ghost_snapshot(&mut self) {
if self.ghost_snapshots_disabled {
self.ghost_snapshot_queue.clear();
return;
}
if self.active_ghost_snapshot.is_some() {
return;
}
let Some((job_id, request)) = self.ghost_snapshot_queue.pop_front() else {
return;
};
let repo_path = self.config.cwd.clone();
let app_event_tx = self.app_event_tx.clone();
let started_at = request.started_at;
self.active_ghost_snapshot = Some((job_id, request));
tokio::spawn(async move {
let result = tokio::task::spawn_blocking(move || {
let options = CreateGhostCommitOptions::new(repo_path.as_path());
create_ghost_commit(&options)
})
.await;
let elapsed = started_at.elapsed();
let event = match result {
Ok(Ok(commit)) => AppEvent::GhostSnapshotFinished {
job_id,
result: Ok(commit),
elapsed,
},
Ok(Err(err)) => AppEvent::GhostSnapshotFinished {
job_id,
result: Err(err),
elapsed,
},
Err(join_err) => {
let err = GitToolingError::Io(io::Error::new(
io::ErrorKind::Other,
format!("ghost snapshot task failed: {join_err}"),
));
AppEvent::GhostSnapshotFinished {
job_id,
result: Err(err),
elapsed,
}
}
};
app_event_tx.send(event);
});
}
fn finalize_ghost_snapshot(
&mut self,
request: GhostSnapshotRequest,
result: Result<GhostCommit, GitToolingError>,
elapsed: Duration,
) -> Option<GhostSnapshot> {
match result {
Ok(commit) => {
self.ghost_snapshots_disabled = false;
self.ghost_snapshots_disabled_reason = None;
self.ghost_snapshots
.push(GhostSnapshot::new(commit, summary, conversation));
let snapshot = GhostSnapshot::new(commit, request.summary, request.conversation);
self.ghost_snapshots.push(snapshot.clone());
if self.ghost_snapshots.len() > MAX_TRACKED_GHOST_COMMITS {
self.ghost_snapshots.remove(0);
}
if elapsed >= Duration::from_secs(2) {
self.push_background_tail(format!(
"Workspace snapshot captured in {}.",
format_duration(elapsed)
));
}
Some(snapshot)
}
Err(err) => {
self.ghost_snapshots_disabled = true;
@@ -4976,10 +5189,40 @@ impl ChatWidget<'_> {
self.push_background_tail(hint);
}
tracing::warn!("failed to create ghost snapshot: {err}");
self.ghost_snapshot_queue.clear();
None
}
}
}
pub(crate) fn handle_ghost_snapshot_finished(
&mut self,
job_id: u64,
result: Result<GhostCommit, GitToolingError>,
elapsed: Duration,
) {
let Some((active_id, request)) = self.active_ghost_snapshot.take() else {
tracing::warn!("ghost snapshot finished without active job (id={job_id})");
return;
};
if active_id != job_id {
tracing::warn!(
"ghost snapshot job id mismatch: expected {active_id}, got {job_id}"
);
self.active_ghost_snapshot = Some((active_id, request));
return;
}
let snapshot = self.finalize_ghost_snapshot(request, result, elapsed);
self.request_redraw();
self.dispatch_pending_snapshot(job_id);
if snapshot.is_none() {
self.flush_pending_snapshot_dispatches();
}
self.spawn_next_ghost_snapshot();
}
fn current_conversation_snapshot(&self) -> ConversationSnapshot {
use crate::history_cell::HistoryCellType;
let mut user_turns = 0usize;
@@ -5285,9 +5528,10 @@ impl ChatWidget<'_> {
if restore_files {
let previous_len = self.ghost_snapshots.len();
let pre_summary = Some("Pre-undo checkpoint".to_string());
self.capture_ghost_snapshot(pre_summary);
if self.ghost_snapshots.len() > previous_len {
pre_restore_snapshot = self.ghost_snapshots.last().cloned();
let captured_snapshot = self.capture_ghost_snapshot_blocking(pre_summary);
let added_snapshot = self.ghost_snapshots.len() > previous_len;
if let Some(snapshot) = captured_snapshot {
pre_restore_snapshot = Some(snapshot);
}
match restore_ghost_commit(&self.config.cwd, snapshot.commit()) {
@@ -5302,7 +5546,7 @@ impl ChatWidget<'_> {
}
}
Err(err) => {
if self.ghost_snapshots.len() > previous_len {
if added_snapshot && !self.ghost_snapshots.is_empty() {
self.ghost_snapshots.pop();
}
errors.push(format!("Failed to restore workspace files: {err}"));
@@ -7419,12 +7663,18 @@ impl ChatWidget<'_> {
fn update_rate_limit_resets(&mut self, current: &RateLimitSnapshotEvent) {
let now = Utc::now();
self.rate_limit_primary_next_reset_at = current
.primary_reset_after_seconds
.map(|secs| now + ChronoDuration::seconds(secs as i64));
self.rate_limit_secondary_next_reset_at = current
.secondary_reset_after_seconds
.map(|secs| now + ChronoDuration::seconds(secs as i64));
if let Some(secs) = current.primary_reset_after_seconds {
self.rate_limit_primary_next_reset_at =
Some(now + ChronoDuration::seconds(secs as i64));
} else {
self.rate_limit_primary_next_reset_at = None;
}
if let Some(secs) = current.secondary_reset_after_seconds {
self.rate_limit_secondary_next_reset_at =
Some(now + ChronoDuration::seconds(secs as i64));
} else {
self.rate_limit_secondary_next_reset_at = None;
}
}
pub(crate) fn handle_update_command(&mut self) {
@@ -13627,6 +13877,10 @@ impl ChatWidget<'_> {
&self.total_token_usage
}
pub(crate) fn session_id(&self) -> Option<uuid::Uuid> {
self.session_id
}
pub(crate) fn clear_token_usage(&mut self) {
self.total_token_usage = TokenUsage::default();
self.rate_limit_snapshot = None;
@@ -15711,6 +15965,32 @@ mod tests {
);
}
#[test]
fn rate_limit_warnings_emit_only_highest_threshold() {
let mut state = RateLimitWarningState::default();
// Jump straight past multiple weekly thresholds; emit just the highest crossed one.
let weekly = state.take_warnings(80.0, 10.0);
assert_eq!(weekly.len(), 1);
assert_eq!(weekly[0].threshold, 75.0);
assert!(matches!(weekly[0].scope, RateLimitWarningScope::Secondary));
// Calling again without crossing new thresholds should produce nothing.
assert!(state.take_warnings(80.0, 10.0).is_empty());
// Crossing the final threshold should produce it exactly once.
let weekly_final = state.take_warnings(95.0, 10.0);
assert_eq!(weekly_final.len(), 1);
assert_eq!(weekly_final[0].threshold, 90.0);
assert!(matches!(weekly_final[0].scope, RateLimitWarningScope::Secondary));
// Primary scope follows the same rule.
let primary = state.take_warnings(95.0, 80.0);
assert_eq!(primary.len(), 1);
assert_eq!(primary[0].threshold, 75.0);
assert!(matches!(primary[0].scope, RateLimitWarningScope::Primary));
}
#[tokio::test(flavor = "current_thread")]
async fn spinner_stays_while_any_agent_running() {
let mut chat = make_widget();
+1 -2
View File
@@ -4,7 +4,7 @@ use codex_core::protocol::InputItem;
use serde::{Deserialize, Serialize};
use std::path::PathBuf;
#[derive(Debug, Serialize, Deserialize)]
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
pub struct UserMessage {
/// What to show in the chat history (keeps placeholders like "[image: name.png]")
pub display_text: String,
@@ -39,4 +39,3 @@ pub fn create_initial_user_message(text: String, image_paths: Vec<PathBuf>) -> O
Some(UserMessage { display_text: text, ordered_items: ordered })
}
}
+16 -3
View File
@@ -27,6 +27,7 @@ use std::path::PathBuf;
use tracing_appender::non_blocking;
use tracing_subscriber::EnvFilter;
use tracing_subscriber::prelude::*;
use uuid::Uuid;
mod app;
mod app_event;
@@ -112,10 +113,18 @@ fn theme_configured_in_config_file(codex_home: &std::path::Path) -> bool {
// (tests access modules directly within the crate)
#[derive(Debug)]
pub struct ExitSummary {
pub token_usage: codex_core::protocol::TokenUsage,
pub session_id: Option<Uuid>,
}
pub const RESUME_COMMAND_NAME: &str = "code";
pub async fn run_main(
mut cli: Cli,
codex_linux_sandbox_exe: Option<PathBuf>,
) -> std::io::Result<codex_core::protocol::TokenUsage> {
) -> std::io::Result<ExitSummary> {
cli.finalize_defaults();
let (sandbox_mode, approval_policy) = if cli.full_auto {
@@ -307,7 +316,7 @@ fn run_ratatui_app(
startup_footer_notice: Option<String>,
latest_upgrade_version: Option<String>,
theme_configured_explicitly: bool,
) -> color_eyre::Result<codex_core::protocol::TokenUsage> {
) -> color_eyre::Result<ExitSummary> {
color_eyre::install()?;
// Forward panic reports through tracing so they appear in the UI status
@@ -365,6 +374,7 @@ fn run_ratatui_app(
);
let app_result = app.run(&mut terminal);
let session_id = app.session_id();
let usage = app.token_usage();
// Optionally print timing summary to stderr after restoring the terminal.
@@ -381,7 +391,10 @@ fn run_ratatui_app(
}
// ignore error when collecting usage – report underlying error instead
app_result.map(|_| usage)
app_result.map(|_| ExitSummary {
token_usage: usage,
session_id,
})
}
#[expect(
+18 -4
View File
@@ -2,7 +2,9 @@ use clap::Parser;
use codex_arg0::arg0_dispatch_or_else;
use codex_common::CliConfigOverrides;
use codex_tui::Cli;
use codex_tui::ExitSummary;
use codex_tui::run_main;
use codex_tui::RESUME_COMMAND_NAME;
#[derive(Parser, Debug)]
struct TopCli {
@@ -22,10 +24,22 @@ fn main() -> anyhow::Result<()> {
.config_overrides
.raw_overrides
.splice(0..0, top_cli.config_overrides.raw_overrides);
let usage = run_main(inner, codex_linux_sandbox_exe).await?;
if !usage.is_zero() {
println!("{}", codex_core::protocol::FinalOutput::from(usage));
// Conversation id hint not available in fork run_main() return type.
let ExitSummary {
token_usage,
session_id,
} = run_main(inner, codex_linux_sandbox_exe).await?;
if !token_usage.is_zero() {
println!(
"{}",
codex_core::protocol::FinalOutput::from(token_usage.clone())
);
}
if let Some(session_id) = session_id {
println!(
"To continue this session, run {} resume {}.",
RESUME_COMMAND_NAME,
session_id
);
}
Ok(())
})