feat(tui/limits): auto refresh secondary windows

This commit is contained in:
James Peter
2026-01-02 14:39:57 +10:00
parent b67c9bde4b
commit 903bfc8eba
6 changed files with 485 additions and 26 deletions
+50
View File
@@ -16,6 +16,7 @@ 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);
const RATE_LIMIT_REFRESH_STALE_INTERVAL_SECS: i64 = 30 * 60;
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub enum RateLimitWarningScope {
@@ -101,6 +102,8 @@ struct RateLimitInfo {
#[serde(default)]
secondary_next_reset_at: Option<DateTime<Utc>>,
#[serde(default)]
last_refresh_attempt_at: Option<DateTime<Utc>>,
#[serde(default)]
last_usage_limit_hit_at: Option<DateTime<Utc>>,
#[serde(default)]
primary_threshold_logs: Vec<RateLimitWarningRecord>,
@@ -535,6 +538,53 @@ pub fn record_usage_limit_hint(
})
}
pub fn rate_limit_refresh_stale_interval() -> Duration {
Duration::seconds(RATE_LIMIT_REFRESH_STALE_INTERVAL_SECS)
}
pub fn mark_rate_limit_refresh_attempt_if_due(
code_home: &Path,
account_id: &str,
plan: Option<&str>,
reset_at: Option<DateTime<Utc>>,
now: DateTime<Utc>,
stale_interval: Duration,
) -> std::io::Result<bool> {
let mut should_refresh = false;
with_usage_file(code_home, account_id, plan, |data| {
let had_info = data.rate_limit.is_some();
let mut info = data.rate_limit.take().unwrap_or_default();
let last_attempt = info.last_refresh_attempt_at;
let refresh_due = if let Some(reset_at) = reset_at {
if now + RESET_PASSED_TOLERANCE < reset_at {
false
} else if last_attempt
.is_some_and(|attempt| attempt + RESET_PASSED_TOLERANCE >= reset_at)
{
false
} else {
true
}
} else {
match last_attempt {
Some(attempt) => now.signed_duration_since(attempt) >= stale_interval,
None => true,
}
};
if refresh_due {
info.last_refresh_attempt_at = Some(now);
should_refresh = true;
}
if refresh_due || had_info {
data.rate_limit = Some(info);
}
})?;
Ok(should_refresh)
}
fn record_threshold_log(
logs: &mut Vec<RateLimitWarningRecord>,
threshold: f64,
+115
View File
@@ -290,6 +290,26 @@ impl CodexAuth {
crate::default_client::create_client(crate::default_client::DEFAULT_ORIGINATOR),
)
}
pub fn from_tokens_with_originator(
tokens: TokenData,
last_refresh: Option<DateTime<Utc>>,
originator: &str,
) -> Self {
let auth_dot_json = AuthDotJson {
openai_api_key: None,
tokens: Some(tokens),
last_refresh,
};
Self {
api_key: None,
mode: AuthMode::ChatGPT,
auth_file: PathBuf::new(),
auth_dot_json: Arc::new(Mutex::new(Some(auth_dot_json))),
client: crate::default_client::create_client(originator),
}
}
}
pub const OPENAI_API_KEY_ENV_VAR: &str = "OPENAI_API_KEY";
@@ -343,6 +363,88 @@ pub fn login_with_api_key(code_home: &Path, api_key: &str) -> std::io::Result<()
Ok(())
}
pub async fn auth_for_stored_account(
code_home: &Path,
account: &crate::auth_accounts::StoredAccount,
originator: &str,
) -> std::io::Result<CodexAuth> {
match account.mode {
AuthMode::ApiKey => {
let api_key = account.openai_api_key.clone().ok_or_else(|| {
std::io::Error::other("stored API key account is missing the key value")
})?;
Ok(CodexAuth::from_api_key_with_client(
&api_key,
crate::default_client::create_client(originator),
))
}
AuthMode::ChatGPT => {
let mut tokens = account.tokens.clone().ok_or_else(|| {
std::io::Error::other("stored ChatGPT account is missing token data")
})?;
let mut last_refresh = account.last_refresh;
let now = Utc::now();
let refresh_needed = last_refresh
.map(|last| last < now - chrono::Duration::days(28))
.unwrap_or(true);
if refresh_needed {
let client = crate::default_client::create_client(originator);
let refresh_response = tokio::time::timeout(
Duration::from_secs(60),
try_refresh_token(tokens.refresh_token.clone(), &client),
)
.await
.map_err(|_| {
std::io::Error::other("timed out while refreshing OpenAI API key")
})?;
let refresh_response = match refresh_response {
Ok(response) => response,
Err(err) => {
if err.is_refresh_token_reused() {
if let Ok(Some(updated)) =
crate::auth_accounts::find_account(code_home, &account.id)
{
if let Some(updated_tokens) = updated.tokens {
return Ok(CodexAuth::from_tokens_with_originator(
updated_tokens,
updated.last_refresh,
originator,
));
}
}
}
return Err(std::io::Error::other(err));
}
};
tokens.id_token =
parse_id_token(&refresh_response.id_token).map_err(std::io::Error::other)?;
if let Some(access_token) = refresh_response.access_token {
tokens.access_token = access_token;
}
if let Some(refresh_token) = refresh_response.refresh_token {
tokens.refresh_token = refresh_token;
}
last_refresh = Some(now);
let _ = crate::auth_accounts::upsert_chatgpt_account(
code_home,
tokens.clone(),
now,
account.label.clone(),
false,
)?;
}
Ok(CodexAuth::from_tokens_with_originator(
tokens,
last_refresh,
originator,
))
}
}
}
/// Activate a stored account by writing its credentials to auth.json and
/// marking it active in the account store.
pub fn activate_account(code_home: &Path, account_id: &str) -> std::io::Result<()> {
@@ -1167,6 +1269,19 @@ impl AuthManager {
})
}
pub fn from_auth(auth: CodexAuth, code_home: PathBuf, originator: String) -> Arc<Self> {
let preferred_auth_mode = auth.mode;
Arc::new(Self {
code_home,
originator,
inner: RwLock::new(CachedAuth {
preferred_auth_mode,
auth: Some(auth),
}),
enable_code_api_key_env: false,
})
}
/// Current cached auth (clone). May be `None` if not logged in or load failed.
pub fn auth(&self) -> Option<CodexAuth> {
self.inner.read().ok().and_then(|c| c.auth.clone())
+6
View File
@@ -1304,6 +1304,12 @@ impl App<'_> {
AppState::Chat { widget } => widget.on_rate_limit_refresh_failed(message),
AppState::Onboarding { .. } => {}
},
AppEvent::RateLimitSnapshotStored { account_id } => match &mut self.app_state {
AppState::Chat { widget } => {
widget.on_rate_limit_snapshot_stored(account_id)
}
AppState::Onboarding { .. } => {}
},
AppEvent::RequestRedraw => {
self.schedule_redraw();
}
+3
View File
@@ -516,6 +516,9 @@ pub(crate) enum AppEvent {
/// Background rate limit refresh failed (threaded request).
RateLimitFetchFailed { message: String },
/// Background rate limit refresh persisted an account snapshot.
RateLimitSnapshotStored { account_id: String },
#[allow(dead_code)]
StartCommitAnimation,
#[allow(dead_code)]
+164 -1
View File
@@ -208,7 +208,10 @@ use code_auto_drive_core::{
use self::limits_overlay::{LimitsOverlayContent, LimitsTab};
use crate::chrome_launch::ChromeLaunchOption;
use crate::insert_history::word_wrap_lines;
use self::rate_limit_refresh::start_rate_limit_refresh;
use self::rate_limit_refresh::{
start_rate_limit_refresh,
start_rate_limit_refresh_for_account,
};
use self::history_render::{
CachedLayout, HistoryRenderState, RenderRequest, RenderRequestKind, RenderSettings, VisibleCell,
};
@@ -1635,6 +1638,8 @@ pub(crate) struct ChatWidget<'a> {
rate_limit_last_fetch_at: Option<DateTime<Utc>>,
rate_limit_primary_next_reset_at: Option<DateTime<Utc>>,
rate_limit_secondary_next_reset_at: Option<DateTime<Utc>>,
rate_limit_refresh_scheduled_for: Option<DateTime<Utc>>,
rate_limit_refresh_schedule_id: Arc<AtomicU64>,
content_buffer: String,
// Buffer for streaming assistant answer text; we do not surface partial
// We wait for the final AgentMessage event and then emit the full text
@@ -6470,6 +6475,8 @@ impl ChatWidget<'_> {
rate_limit_last_fetch_at: None,
rate_limit_primary_next_reset_at: None,
rate_limit_secondary_next_reset_at: None,
rate_limit_refresh_scheduled_for: None,
rate_limit_refresh_schedule_id: Arc::new(AtomicU64::new(0)),
content_buffer: String::new(),
last_assistant_message: None,
last_answer_stream_id_in_turn: None,
@@ -6678,6 +6685,7 @@ impl ChatWidget<'_> {
if let Some(record) = records.into_iter().find(|r| r.account_id == active_id) {
new_widget.rate_limit_primary_next_reset_at = record.primary_next_reset_at;
new_widget.rate_limit_secondary_next_reset_at = record.secondary_next_reset_at;
new_widget.maybe_schedule_rate_limit_refresh();
}
}
}
@@ -6819,6 +6827,8 @@ impl ChatWidget<'_> {
rate_limit_last_fetch_at: None,
rate_limit_primary_next_reset_at: None,
rate_limit_secondary_next_reset_at: None,
rate_limit_refresh_scheduled_for: None,
rate_limit_refresh_schedule_id: Arc::new(AtomicU64::new(0)),
content_buffer: String::new(),
last_assistant_message: None,
last_answer_stream_id_in_turn: None,
@@ -7043,6 +7053,7 @@ impl ChatWidget<'_> {
if let Some(record) = records.into_iter().find(|r| r.account_id == active_id) {
w.rate_limit_primary_next_reset_at = record.primary_next_reset_at;
w.rate_limit_secondary_next_reset_at = record.secondary_next_reset_at;
w.maybe_schedule_rate_limit_refresh();
}
}
}
@@ -15469,6 +15480,62 @@ impl ChatWidget<'_> {
if needs_refresh {
self.request_latest_rate_limits(snapshot.is_none());
}
self.refresh_limits_for_other_accounts_if_due();
}
fn refresh_limits_for_other_accounts_if_due(&mut self) {
let code_home = self.config.code_home.clone();
let active_id = auth_accounts::get_active_account_id(&code_home)
.ok()
.flatten();
let accounts = auth_accounts::list_accounts(&code_home).unwrap_or_default();
if accounts.is_empty() {
return;
}
let usage_records = account_usage::list_rate_limit_snapshots(&code_home).unwrap_or_default();
let snapshot_map: HashMap<String, StoredRateLimitSnapshot> = usage_records
.into_iter()
.map(|record| (record.account_id.clone(), record))
.collect();
let now = Utc::now();
let stale_interval = account_usage::rate_limit_refresh_stale_interval();
for account in accounts {
if active_id.as_deref() == Some(account.id.as_str()) {
continue;
}
let reset_at = snapshot_map
.get(&account.id)
.and_then(|record| record.secondary_next_reset_at);
let plan = account
.tokens
.as_ref()
.and_then(|tokens| tokens.id_token.get_chatgpt_plan_type());
let should_refresh = account_usage::mark_rate_limit_refresh_attempt_if_due(
&code_home,
&account.id,
plan.as_deref(),
reset_at,
now,
stale_interval,
)
.unwrap_or(false);
if should_refresh {
start_rate_limit_refresh_for_account(
self.app_event_tx.clone(),
self.config.clone(),
self.config.debug,
account,
false,
false,
);
}
}
}
fn request_latest_rate_limits(&mut self, show_loading: bool) {
@@ -15524,6 +15591,23 @@ impl ChatWidget<'_> {
}
}
pub(crate) fn on_rate_limit_snapshot_stored(&mut self, _account_id: String) {
self.refresh_settings_overview_rows();
let refresh_limits_settings = self
.settings
.overlay
.as_ref()
.map(|overlay| {
overlay.active_section() == SettingsSection::Limits && !overlay.is_menu_active()
})
.unwrap_or(false);
if refresh_limits_settings {
self.show_limits_settings_ui();
} else {
self.request_redraw();
}
}
fn rate_limit_reset_info(&self) -> RateLimitResetInfo {
let auto_compact_limit = self
.config
@@ -15575,6 +15659,85 @@ impl ChatWidget<'_> {
} else {
self.rate_limit_secondary_next_reset_at = None;
}
self.maybe_schedule_rate_limit_refresh();
}
fn maybe_schedule_rate_limit_refresh(&mut self) {
let Some(reset_at) = self.rate_limit_secondary_next_reset_at else {
self.rate_limit_refresh_scheduled_for = None;
self.rate_limit_refresh_schedule_id.fetch_add(1, Ordering::SeqCst);
return;
};
if self.rate_limit_refresh_scheduled_for == Some(reset_at) {
return;
}
self.rate_limit_refresh_scheduled_for = Some(reset_at);
let schedule_id = self
.rate_limit_refresh_schedule_id
.fetch_add(1, Ordering::SeqCst)
.saturating_add(1);
let schedule_token = self.rate_limit_refresh_schedule_id.clone();
let app_event_tx = self.app_event_tx.clone();
let config = self.config.clone();
let debug_enabled = self.config.debug;
let account = auth_accounts::get_active_account_id(&config.code_home)
.ok()
.flatten()
.and_then(|id| auth_accounts::find_account(&config.code_home, &id).ok())
.flatten();
if account.is_none() {
return;
}
if thread_spawner::spawn_lightweight("rate-reset-refresh", move || {
let now = Utc::now();
let delay = reset_at.signed_duration_since(now) + ChronoDuration::seconds(1);
if let Ok(delay) = delay.to_std() {
if !delay.is_zero() {
std::thread::sleep(delay);
}
}
if schedule_token.load(Ordering::SeqCst) != schedule_id {
return;
}
let Some(account) = account else {
return;
};
let plan = account
.tokens
.as_ref()
.and_then(|tokens| tokens.id_token.get_chatgpt_plan_type());
let should_refresh = account_usage::mark_rate_limit_refresh_attempt_if_due(
&config.code_home,
&account.id,
plan.as_deref(),
Some(reset_at),
Utc::now(),
account_usage::rate_limit_refresh_stale_interval(),
)
.unwrap_or(false);
if should_refresh {
start_rate_limit_refresh_for_account(
app_event_tx,
config,
debug_enabled,
account,
true,
false,
);
}
})
.is_none()
{
tracing::warn!("rate reset refresh scheduling failed: worker unavailable");
}
}
pub(crate) fn handle_update_command(&mut self, command_args: &str) {
+147 -25
View File
@@ -1,12 +1,16 @@
use std::sync::{Arc, Mutex};
use anyhow::{Context, Result};
use code_core::auth::auth_for_stored_account;
use code_core::auth_accounts::{self, StoredAccount};
use code_core::{AuthManager, ModelClient, Prompt, ResponseEvent};
use code_core::account_usage;
use code_core::config::Config;
use code_core::config_types::ReasoningEffort;
use code_core::debug_logger::DebugLogger;
use code_core::protocol::{Event, EventMsg, RateLimitSnapshotEvent, TokenCountEvent};
use code_protocol::models::{ContentItem, ResponseItem};
use chrono::Utc;
use futures::StreamExt;
use tokio::runtime::Runtime;
use uuid::Uuid;
@@ -25,18 +29,69 @@ pub(super) fn start_rate_limit_refresh(
app_event_tx: AppEventSender,
config: Config,
debug_enabled: bool,
) {
start_rate_limit_refresh_with_options(
app_event_tx,
config,
debug_enabled,
None,
true,
true,
);
}
pub(super) fn start_rate_limit_refresh_for_account(
app_event_tx: AppEventSender,
config: Config,
debug_enabled: bool,
account: StoredAccount,
emit_ui: bool,
notify_on_failure: bool,
) {
start_rate_limit_refresh_with_options(
app_event_tx,
config,
debug_enabled,
Some(account),
emit_ui,
notify_on_failure,
);
}
fn start_rate_limit_refresh_with_options(
app_event_tx: AppEventSender,
config: Config,
debug_enabled: bool,
account: Option<StoredAccount>,
emit_ui: bool,
notify_on_failure: bool,
) {
let fallback_tx = app_event_tx.clone();
if thread_spawner::spawn_lightweight("rate-refresh", move || {
if let Err(err) = run_refresh(app_event_tx.clone(), config, debug_enabled) {
let message = format!("Failed to refresh rate limits: {err}");
app_event_tx.send(AppEvent::RateLimitFetchFailed { message });
if let Err(err) = run_refresh(
app_event_tx.clone(),
config,
debug_enabled,
account,
emit_ui,
) {
if notify_on_failure {
let message = format!("Failed to refresh rate limits: {err}");
app_event_tx.send(AppEvent::RateLimitFetchFailed { message });
} else {
tracing::warn!("Failed to refresh rate limits: {err}");
}
}
})
.is_none()
{
let message = "Failed to refresh rate limits: background worker unavailable".to_string();
fallback_tx.send(AppEvent::RateLimitFetchFailed { message });
if notify_on_failure {
let message =
"Failed to refresh rate limits: background worker unavailable".to_string();
fallback_tx.send(AppEvent::RateLimitFetchFailed { message });
} else {
tracing::warn!("Failed to refresh rate limits: background worker unavailable");
}
}
}
@@ -44,25 +99,52 @@ fn run_refresh(
app_event_tx: AppEventSender,
config: Config,
debug_enabled: bool,
account: Option<StoredAccount>,
emit_ui: bool,
) -> Result<()> {
let runtime = build_runtime()?;
runtime.block_on(async move {
let auth_mode = if config.using_chatgpt_auth {
code_protocol::mcp_protocol::AuthMode::ChatGPT
} else {
code_protocol::mcp_protocol::AuthMode::ApiKey
let (auth_mgr, stored_account) = match account {
Some(account) => {
let auth = auth_for_stored_account(
&config.code_home,
&account,
&config.responses_originator_header,
)
.await
.context("building auth for stored account")?;
(
AuthManager::from_auth(
auth,
config.code_home.clone(),
config.responses_originator_header.clone(),
),
Some(account),
)
}
None => {
let auth_mode = if config.using_chatgpt_auth {
code_protocol::mcp_protocol::AuthMode::ChatGPT
} else {
code_protocol::mcp_protocol::AuthMode::ApiKey
};
(
AuthManager::shared_with_mode_and_originator(
config.code_home.clone(),
auth_mode,
config.responses_originator_header.clone(),
),
None,
)
}
};
let auth_mgr = AuthManager::shared_with_mode_and_originator(
config.code_home.clone(),
auth_mode,
config.responses_originator_header.clone(),
);
let client = build_model_client(&config, auth_mgr, debug_enabled)?;
let mut prompt = Prompt::default();
prompt.store = false;
prompt.user_instructions = config.user_instructions.clone();
prompt.base_instructions_override = config.base_instructions.clone();
prompt.input.push(ResponseItem::Message {
id: None,
role: "user".to_string(),
@@ -93,20 +175,60 @@ fn run_refresh(
let snapshot: RateLimitSnapshotEvent = proto_snapshot.clone();
let (record_account_id, record_plan) = if let Some(account) = &stored_account {
(
Some(account.id.clone()),
account
.tokens
.as_ref()
.and_then(|tokens| tokens.id_token.get_chatgpt_plan_type()),
)
} else {
let active_id =
auth_accounts::get_active_account_id(&config.code_home).ok().flatten();
let account = active_id
.as_deref()
.and_then(|id| auth_accounts::find_account(&config.code_home, id).ok())
.flatten();
(
active_id,
account
.as_ref()
.and_then(|acc| acc.tokens.as_ref())
.and_then(|tokens| tokens.id_token.get_chatgpt_plan_type()),
)
};
if let Some(account_id) = record_account_id.as_deref() {
if let Err(err) = account_usage::record_rate_limit_snapshot(
&config.code_home,
account_id,
record_plan.as_deref(),
&snapshot,
Utc::now(),
) {
tracing::warn!("Failed to persist rate limit snapshot: {err}");
}
}
#[cfg(feature = "code-fork")]
handle_rate_limit(&snapshot, &app_event_tx);
let event = Event {
id: "rate-limit-refresh".to_string(),
event_seq: 0,
msg: EventMsg::TokenCount(TokenCountEvent {
info: None,
rate_limits: Some(snapshot),
}),
order: None,
};
if emit_ui {
let event = Event {
id: "rate-limit-refresh".to_string(),
event_seq: 0,
msg: EventMsg::TokenCount(TokenCountEvent {
info: None,
rate_limits: Some(snapshot),
}),
order: None,
};
app_event_tx.send(AppEvent::CodexEvent(event));
app_event_tx.send(AppEvent::CodexEvent(event));
} else if let Some(account_id) = record_account_id {
app_event_tx.send(AppEvent::RateLimitSnapshotStored { account_id });
}
Ok(())
})
}