mirror of
https://github.com/openai/codex.git
synced 2026-11-07 20:30:33 +00:00
Keep automatic thread title saves off the TUI event loop (#58970)
## Why Saving a generated title awaited thread metadata reads and name writes in the TUI event handler, delaying its return to keyboard polling. ## What changed Move automatic title persistence into the background title-generation task. Make metadata reads cancellable and serialize automatic writes with manual renames so a manual name takes precedence, including when an automatic write is already in flight. ## Testing Extend title tests to check that completion dispatch returns immediately, canceled metadata reads finish, and manual renames wait for in-flight automatic writes. Retain coverage for saving titles to the originating thread after switching threads and preserving existing names. GitOrigin-RevId: b284a06ca3f0fdcf22fd2fc87b07447f8ab754c0
This commit is contained in:
@@ -114,7 +114,7 @@ impl BackendSessions {
|
||||
self.entries
|
||||
.iter()
|
||||
.filter(|entry| entry.listening || entry.session.backend_id() == displayed_owner)
|
||||
.map(|entry| entry.session.picker_view())
|
||||
.map(|entry| entry.session.request_view())
|
||||
.collect()
|
||||
}
|
||||
|
||||
|
||||
@@ -3119,36 +3119,12 @@ impl App {
|
||||
return Ok(AppRunControl::Continue);
|
||||
}
|
||||
self.finish_thread_title_generation(thread_id, destination);
|
||||
match destination {
|
||||
ThreadTitleDestination::Automatic => {
|
||||
if let Ok(response) = result
|
||||
&& let Some(title) = super::thread_title::parse_thread_title(&response)
|
||||
&& let Ok(thread) = app_server
|
||||
.thread_read(thread_id, /*include_turns*/ false)
|
||||
.await
|
||||
&& thread.name.is_none()
|
||||
{
|
||||
match app_server.thread_set_name(thread_id, title.clone()).await {
|
||||
Ok(()) => self
|
||||
.chat_widget
|
||||
.on_thread_name_updated(thread_id, Some(title)),
|
||||
Err(error) => {
|
||||
tracing::debug!(%error, "failed to apply generated thread title");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
ThreadTitleDestination::RenameSuggestion { request_id } => {
|
||||
let suggestion = result
|
||||
.ok()
|
||||
.and_then(|response| super::thread_title::parse_thread_title(&response));
|
||||
|
||||
self.chat_widget.apply_thread_name_suggestion(
|
||||
thread_id,
|
||||
request_id,
|
||||
suggestion.as_deref(),
|
||||
);
|
||||
}
|
||||
if let ThreadTitleDestination::RenameSuggestion { request_id } = destination {
|
||||
let suggestion = result.ok()
|
||||
.and_then(|response| super::thread_title::parse_thread_title(&response));
|
||||
self.chat_widget.apply_thread_name_suggestion(
|
||||
thread_id, request_id, suggestion.as_deref(),
|
||||
);
|
||||
}
|
||||
}
|
||||
AppEvent::HideAgentsOverviewThread { thread_id } => {
|
||||
|
||||
@@ -130,7 +130,7 @@ impl App {
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub(super) fn on_thread_title_started(
|
||||
&mut self,
|
||||
app_server: &AppServerSession,
|
||||
app_server: &mut AppServerSession,
|
||||
thread_id: ThreadId,
|
||||
destination: ThreadTitleDestination,
|
||||
prompt: String,
|
||||
@@ -174,6 +174,8 @@ impl App {
|
||||
self.temporary_structured_requests
|
||||
.insert(temporary_thread_id, sender);
|
||||
|
||||
let save_name = (destination == ThreadTitleDestination::Automatic)
|
||||
.then(|| app_server.automatic_thread_name_writer(thread_id));
|
||||
let request_handle = app_server.request_handle();
|
||||
let event_sender = self.app_event_tx.clone();
|
||||
tokio::spawn(async move {
|
||||
@@ -189,6 +191,12 @@ impl App {
|
||||
.await
|
||||
.map_err(|error| error.to_string());
|
||||
|
||||
if let Some(save_name) = save_name
|
||||
&& let Ok(response) = &result
|
||||
&& let Some(title) = parse_thread_title(response)
|
||||
{
|
||||
save_name.save(title, &cancellation).await;
|
||||
}
|
||||
event_sender.send(AppEvent::GeneratedThreadTitle {
|
||||
cancellation,
|
||||
thread_id,
|
||||
|
||||
@@ -32,6 +32,7 @@ use core_test_support::responses;
|
||||
use crossterm::event::KeyCode;
|
||||
use crossterm::event::KeyEvent;
|
||||
use crossterm::event::KeyModifiers;
|
||||
use futures::FutureExt;
|
||||
use pretty_assertions::assert_eq;
|
||||
use tempfile::tempdir;
|
||||
use tokio::sync::mpsc::unbounded_channel;
|
||||
@@ -140,6 +141,7 @@ async fn automatic_thread_title_respects_origin_metadata_after_switching() -> co
|
||||
);
|
||||
|
||||
let mut tui = crate::tui::test_support::make_test_tui()?;
|
||||
let automatic = app_server.automatic_thread_name_writer(thread_id);
|
||||
if switch_threads {
|
||||
let second = app_server.start_thread(&app.config).await?;
|
||||
let second_id = second.session.thread_id;
|
||||
@@ -158,6 +160,9 @@ async fn automatic_thread_title_respects_origin_metadata_after_switching() -> co
|
||||
.await?;
|
||||
}
|
||||
let displayed_thread = (app.chat_widget.thread_id(), app.chat_widget.thread_name());
|
||||
automatic
|
||||
.save("Generated title".into(), &CancellationToken::new())
|
||||
.await;
|
||||
app.handle_event(
|
||||
&mut tui,
|
||||
&mut app_server,
|
||||
@@ -169,7 +174,8 @@ async fn automatic_thread_title_respects_origin_metadata_after_switching() -> co
|
||||
result: Ok(r#"{"title":"Generated title"}"#.to_string()),
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
.now_or_never()
|
||||
.expect("title dispatch must return to keyboard polling")?;
|
||||
let expected_name = Some(manual_name.unwrap_or("Generated title").to_string());
|
||||
assert_eq!(
|
||||
app_server
|
||||
@@ -430,7 +436,13 @@ async fn check_thread_title_generation(scenario: TitleScenario) -> color_eyre::R
|
||||
renamed = true;
|
||||
}
|
||||
}
|
||||
if generated && (!cancel || interrupted) {
|
||||
if generated && automatic && !cancel {
|
||||
app.drain_active_thread_events(&mut tui).await?;
|
||||
}
|
||||
if generated
|
||||
&& (!cancel || interrupted)
|
||||
&& (!automatic || app.chat_widget.thread_name().is_some())
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
@@ -495,12 +507,20 @@ async fn canceled_thread_title_ignores_late_start_and_completion() -> color_eyre
|
||||
app.pending_thread_titles
|
||||
.insert((thread_id, destination), cancellation.clone());
|
||||
|
||||
let writer = app_server.automatic_thread_name_writer(thread_id);
|
||||
let token = cancellation.clone();
|
||||
let mut automatic = std::pin::pin!(writer.save("Stale title".into(), &token));
|
||||
assert!(automatic.as_mut().now_or_never().is_none());
|
||||
|
||||
app.try_submit_active_thread_op_via_app_server(
|
||||
&mut app_server,
|
||||
thread_id,
|
||||
&AppCommand::set_thread_name("Keep this title".to_string()),
|
||||
)
|
||||
.await?;
|
||||
automatic
|
||||
.now_or_never()
|
||||
.expect("canceled metadata read must finish");
|
||||
assert!(cancellation.is_cancelled());
|
||||
assert!(app.pending_thread_titles.is_empty());
|
||||
|
||||
@@ -509,7 +529,7 @@ async fn canceled_thread_title_ignores_late_start_and_completion() -> color_eyre
|
||||
app.pending_thread_titles
|
||||
.insert((thread_id, destination), replacement.clone());
|
||||
app.on_thread_title_started(
|
||||
&app_server,
|
||||
&mut app_server,
|
||||
thread_id,
|
||||
destination,
|
||||
"prompt".to_string(),
|
||||
@@ -547,6 +567,57 @@ async fn canceled_thread_title_ignores_late_start_and_completion() -> color_eyre
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn manual_rename_waits_for_an_in_flight_automatic_write() -> color_eyre::Result<()> {
|
||||
let thread_id = ThreadId::new();
|
||||
let (client, mut requests, _events) =
|
||||
codex_app_server_client::CloudClient::channel(/*capacity*/ 1);
|
||||
let mut session = crate::app_server_session::AppServerSession::new(
|
||||
crate::backend_id::BackendId::new(),
|
||||
codex_app_server_client::AppServerClient::Cloud(client),
|
||||
crate::app_server_session::ThreadParamsMode::Remote,
|
||||
);
|
||||
let writer = session.automatic_thread_name_writer(thread_id);
|
||||
let token = CancellationToken::new();
|
||||
let mut automatic = std::pin::pin!(writer.save("Generated".into(), &token));
|
||||
assert!(automatic.as_mut().now_or_never().is_none());
|
||||
let mut thread = crate::app::agents_overview::tests::overview_thread(
|
||||
thread_id,
|
||||
/*parent_thread_id*/ None,
|
||||
"",
|
||||
codex_app_server_protocol::ThreadStatus::Idle,
|
||||
);
|
||||
thread.name = None;
|
||||
requests
|
||||
.try_recv()?
|
||||
.response
|
||||
.send(Ok(Ok(serde_json::json!({"thread": thread}))))
|
||||
.unwrap();
|
||||
assert!(automatic.as_mut().now_or_never().is_none());
|
||||
let write = requests.try_recv()?;
|
||||
|
||||
let mut manual = std::pin::pin!(session.thread_set_name(thread_id, "Manual".into()));
|
||||
assert!(manual.as_mut().now_or_never().is_none());
|
||||
assert!(
|
||||
requests.try_recv().is_err(),
|
||||
"manual rename raced the automatic write"
|
||||
);
|
||||
let acknowledge = |request: codex_app_server_client::CloudCommand| {
|
||||
request
|
||||
.response
|
||||
.send(Ok(Ok(serde_json::json!({}))))
|
||||
.unwrap();
|
||||
};
|
||||
acknowledge(write);
|
||||
automatic
|
||||
.now_or_never()
|
||||
.expect("automatic write acknowledged");
|
||||
assert!(manual.as_mut().now_or_never().is_none());
|
||||
acknowledge(requests.try_recv()?);
|
||||
manual.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recent_conversation_messages_escape_markup_and_ignore_commentary() {
|
||||
let items = vec![
|
||||
@@ -813,7 +884,7 @@ async fn failed_thread_title_requests_are_removed() -> color_eyre::Result<()> {
|
||||
]);
|
||||
|
||||
app.on_thread_title_started(
|
||||
&app_server,
|
||||
&mut app_server,
|
||||
thread_id,
|
||||
ThreadTitleDestination::Automatic,
|
||||
"prompt".to_string(),
|
||||
@@ -822,7 +893,7 @@ async fn failed_thread_title_requests_are_removed() -> color_eyre::Result<()> {
|
||||
CancellationToken::new(),
|
||||
);
|
||||
app.on_thread_title_started(
|
||||
&app_server,
|
||||
&mut app_server,
|
||||
thread_id,
|
||||
suggestion,
|
||||
"prompt".to_string(),
|
||||
|
||||
@@ -164,6 +164,8 @@ use std::sync::Mutex;
|
||||
use std::sync::atomic::AtomicU64;
|
||||
use std::time::Duration;
|
||||
use std::time::Instant;
|
||||
use tokio::sync::Semaphore;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use uuid::Uuid;
|
||||
|
||||
const JSONRPC_INVALID_REQUEST: i64 = -32600;
|
||||
@@ -327,12 +329,13 @@ pub(crate) struct AppServerBootstrap {
|
||||
|
||||
pub(crate) struct AppServerSession {
|
||||
backend_id: crate::backend_id::BackendId,
|
||||
/// Owns events and the connection lifetime for live sessions. Picker views borrow only the
|
||||
/// request handle below and deliberately have no connection owner.
|
||||
/// Owns events and the connection lifetime for live sessions. Request-only views share the
|
||||
/// request handle below and have no connection owner.
|
||||
connection_owner: Option<AppServerClient>,
|
||||
request_handle: AppServerRequestHandle,
|
||||
request_id_namespace: Option<Uuid>,
|
||||
next_request_id: i64,
|
||||
automatic_name_writes: HashMap<ThreadId, Arc<Semaphore>>,
|
||||
history_pagination: HashMap<ThreadId, history::ThreadHistoryPagination>,
|
||||
pub(crate) focused_conversation: Option<ThreadId>,
|
||||
pub(crate) attempted_revision: Option<String>,
|
||||
@@ -357,6 +360,35 @@ pub(crate) struct AppServerSession {
|
||||
Option<Box<crate::legacy_core::config::ConfigBuilder>>,
|
||||
}
|
||||
|
||||
/// An automatic save holds the gate only while writing, so a manual rename can skip a slow read.
|
||||
pub(crate) struct AutomaticThreadName {
|
||||
session: AppServerSession,
|
||||
thread_id: ThreadId,
|
||||
gate: Arc<Semaphore>,
|
||||
}
|
||||
|
||||
impl AutomaticThreadName {
|
||||
pub(crate) async fn save(mut self, title: String, cancellation: &CancellationToken) {
|
||||
let read = self
|
||||
.session
|
||||
.thread_read(self.thread_id, /*include_turns*/ false);
|
||||
let Some(Ok(thread)) = cancellation.run_until_cancelled(read).await else {
|
||||
return;
|
||||
};
|
||||
if thread.name.is_some() {
|
||||
return;
|
||||
}
|
||||
let Ok(_permit) = self.gate.acquire().await else {
|
||||
return;
|
||||
};
|
||||
if !cancellation.is_cancelled()
|
||||
&& let Err(error) = self.session.thread_set_name(self.thread_id, title).await
|
||||
{
|
||||
tracing::debug!(%error, "failed to apply generated thread title");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub(crate) enum ThreadParamsMode {
|
||||
Embedded,
|
||||
@@ -509,6 +541,7 @@ impl AppServerSession {
|
||||
request_handle,
|
||||
request_id_namespace: None,
|
||||
next_request_id: 1,
|
||||
automatic_name_writes: HashMap::new(),
|
||||
history_pagination: HashMap::new(),
|
||||
focused_conversation: None,
|
||||
attempted_revision: None,
|
||||
@@ -536,8 +569,8 @@ impl AppServerSession {
|
||||
self.backend_id
|
||||
}
|
||||
|
||||
/// Creates a request-only picker view without transferring the live connection's events.
|
||||
pub(crate) fn picker_view(&self) -> Self {
|
||||
/// Creates a request-only view without transferring the live connection's events.
|
||||
pub(crate) fn request_view(&self) -> Self {
|
||||
let mut view = Self::from_parts(
|
||||
self.backend_id,
|
||||
/*connection_owner*/ None,
|
||||
@@ -1688,11 +1721,34 @@ impl AppServerSession {
|
||||
.await
|
||||
}
|
||||
|
||||
pub(crate) fn automatic_thread_name_writer(
|
||||
&mut self,
|
||||
thread_id: ThreadId,
|
||||
) -> AutomaticThreadName {
|
||||
self.automatic_name_writes
|
||||
.retain(|_, gate| Arc::strong_count(gate) > 1);
|
||||
let session = self.request_view();
|
||||
let gate = self
|
||||
.automatic_name_writes
|
||||
.entry(thread_id)
|
||||
.or_insert_with(|| Arc::new(Semaphore::new(/*permits*/ 1)));
|
||||
AutomaticThreadName {
|
||||
session,
|
||||
thread_id,
|
||||
gate: Arc::clone(gate),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn thread_set_name(
|
||||
&mut self,
|
||||
thread_id: ThreadId,
|
||||
name: String,
|
||||
) -> Result<()> {
|
||||
let gate = self.automatic_name_writes.get(&thread_id).cloned();
|
||||
let _permit = match &gate {
|
||||
Some(gate) => Some(gate.acquire().await?),
|
||||
None => None,
|
||||
};
|
||||
let request_id = self.next_request_id();
|
||||
let _: ThreadSetNameResponse = self
|
||||
.request_handle
|
||||
@@ -1705,6 +1761,10 @@ impl AppServerSession {
|
||||
})
|
||||
.await
|
||||
.wrap_err("thread/name/set failed in TUI")?;
|
||||
if let Some(gate) = gate.as_ref() {
|
||||
gate.close();
|
||||
self.automatic_name_writes.remove(&thread_id);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -2883,8 +2943,8 @@ mod tests {
|
||||
AppServerClient::Cloud(client),
|
||||
ThreadParamsMode::Cloud,
|
||||
);
|
||||
let mut first = session.picker_view();
|
||||
let mut second = session.picker_view();
|
||||
let mut first = session.request_view();
|
||||
let mut second = session.request_view();
|
||||
|
||||
let first_id = first.next_request_id();
|
||||
assert_ne!(session.next_request_id(), first_id);
|
||||
@@ -2902,7 +2962,7 @@ mod tests {
|
||||
ThreadParamsMode::Cloud,
|
||||
);
|
||||
|
||||
session.picker_view().shutdown().await.unwrap();
|
||||
session.request_view().shutdown().await.unwrap();
|
||||
events
|
||||
.send(codex_app_server_client::AppServerEvent::Disconnected {
|
||||
message: "still live".to_string(),
|
||||
|
||||
Reference in New Issue
Block a user