mirror of
https://github.com/openai/codex.git
synced 2026-11-01 19:33:35 +00:00
preserve live threads during list validation
This commit is contained in:
@@ -379,6 +379,7 @@ impl RolloutRecorder {
|
||||
/*parent_thread_id*/ None,
|
||||
archived,
|
||||
search_term,
|
||||
/*missing_rollout_thread_ids_to_preserve*/ None,
|
||||
)
|
||||
.await
|
||||
.map(Into::into)
|
||||
@@ -492,6 +493,7 @@ impl RolloutRecorder {
|
||||
/*parent_thread_id*/ None,
|
||||
archived,
|
||||
search_term,
|
||||
/*missing_rollout_thread_ids_to_preserve*/ None,
|
||||
)
|
||||
.await;
|
||||
if let Some(db_page) = db_page {
|
||||
@@ -521,6 +523,7 @@ impl RolloutRecorder {
|
||||
/*parent_thread_id*/ None,
|
||||
archived,
|
||||
search_term,
|
||||
/*missing_rollout_thread_ids_to_preserve*/ None,
|
||||
)
|
||||
.await
|
||||
{
|
||||
@@ -561,6 +564,7 @@ impl RolloutRecorder {
|
||||
/*parent_thread_id*/ None,
|
||||
archived,
|
||||
search_term,
|
||||
/*missing_rollout_thread_ids_to_preserve*/ None,
|
||||
)
|
||||
.await
|
||||
{
|
||||
@@ -639,6 +643,7 @@ impl RolloutRecorder {
|
||||
/*parent_thread_id*/ None,
|
||||
/*archived*/ false,
|
||||
/*search_term*/ None,
|
||||
/*missing_rollout_thread_ids_to_preserve*/ None,
|
||||
)
|
||||
.await
|
||||
else {
|
||||
|
||||
@@ -16,6 +16,7 @@ use codex_state::ThreadMetadata;
|
||||
use codex_state::ThreadMetadataBuilder;
|
||||
use codex_utils_path::normalize_for_path_comparison;
|
||||
use serde_json::Value;
|
||||
use std::collections::HashSet;
|
||||
use std::path::Path;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
@@ -371,6 +372,7 @@ pub async fn list_threads_db(
|
||||
parent_thread_id: Option<ThreadId>,
|
||||
archived: bool,
|
||||
search_term: Option<&str>,
|
||||
missing_rollout_thread_ids_to_preserve: Option<&HashSet<ThreadId>>,
|
||||
) -> Option<codex_state::ThreadListPage> {
|
||||
let ctx = context?;
|
||||
if ctx.codex_home() != codex_home {
|
||||
@@ -431,6 +433,11 @@ pub async fn list_threads_db(
|
||||
let mut item = item;
|
||||
item.rollout_path = existing_path;
|
||||
valid_items.push(item);
|
||||
// Deferred live writers can own state rows before their rollout file materializes.
|
||||
} else if missing_rollout_thread_ids_to_preserve
|
||||
.is_some_and(|thread_ids| thread_ids.contains(&item.id))
|
||||
{
|
||||
valid_items.push(item);
|
||||
} else {
|
||||
warn!(
|
||||
"state db list_threads returned stale rollout path for thread {}: {}",
|
||||
|
||||
@@ -176,7 +176,7 @@ pub(super) fn stored_thread_from_state_list_item(
|
||||
model_provider,
|
||||
cwd,
|
||||
cli_version,
|
||||
title: _,
|
||||
title,
|
||||
name,
|
||||
preview,
|
||||
first_user_message,
|
||||
@@ -188,6 +188,11 @@ pub(super) fn stored_thread_from_state_list_item(
|
||||
let preview = preview
|
||||
.or_else(|| first_user_message.clone())
|
||||
.unwrap_or_default();
|
||||
let name = name.filter(|name| !name.trim().is_empty()).or_else(|| {
|
||||
let title = title.trim();
|
||||
(!title.is_empty() && first_user_message.as_deref().map(str::trim) != Some(title))
|
||||
.then(|| title.to_string())
|
||||
});
|
||||
let git_info = git_info_from_parts(git_sha, git_branch, git_origin_url);
|
||||
let rollout_path = codex_rollout::plain_rollout_path(rollout_path.as_path());
|
||||
let model_provider = if model_provider.is_empty() {
|
||||
|
||||
@@ -100,6 +100,7 @@ async fn list_threads_from_state_db(
|
||||
sort_key: codex_rollout::ThreadSortKey,
|
||||
sort_direction: codex_rollout::SortDirection,
|
||||
) -> ThreadStoreResult<ThreadPage> {
|
||||
let live_thread_ids = store.live_thread_ids().await;
|
||||
let page = codex_rollout::state_db::list_threads_db(
|
||||
state_db,
|
||||
store.config.codex_home.as_path(),
|
||||
@@ -113,6 +114,7 @@ async fn list_threads_from_state_db(
|
||||
params.parent_thread_id,
|
||||
params.archived,
|
||||
params.search_term.as_deref(),
|
||||
Some(&live_thread_ids),
|
||||
)
|
||||
.await;
|
||||
|
||||
@@ -195,6 +197,7 @@ pub(super) async fn list_rollout_threads(
|
||||
Some(parent_thread_id),
|
||||
params.archived,
|
||||
params.search_term.as_deref(),
|
||||
/*missing_rollout_thread_ids_to_preserve*/ None,
|
||||
)
|
||||
.await
|
||||
.ok_or_else(|| ThreadStoreError::Internal {
|
||||
|
||||
@@ -16,6 +16,7 @@ use codex_protocol::ThreadId;
|
||||
use codex_rollout::RolloutRecorder;
|
||||
use codex_rollout::StateDbHandle;
|
||||
use std::collections::HashMap;
|
||||
use std::collections::HashSet;
|
||||
use std::collections::hash_map::Entry;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
@@ -139,6 +140,10 @@ impl LocalThreadStore {
|
||||
.ok_or(ThreadStoreError::ThreadNotFound { thread_id })
|
||||
}
|
||||
|
||||
pub(super) async fn live_thread_ids(&self) -> HashSet<ThreadId> {
|
||||
self.live_recorders.lock().await.keys().copied().collect()
|
||||
}
|
||||
|
||||
pub(super) async fn ensure_live_recorder_absent(
|
||||
&self,
|
||||
thread_id: ThreadId,
|
||||
|
||||
Reference in New Issue
Block a user