Preserve inherited previews for paginated thread forks (#60559)

## Why

Reference-backed forks can lack a local opening user message, causing them to disappear from thread listings or lose their inherited preview when a new turn is written. Compaction can also remove the opening message from the model context used to prepare a fork.

## What changed

- Persist inherited previews and preserve them when recording new turns and merging thread-list metadata.
- Include reference-backed forks with inherited history in thread listings before their first local message.
- Recover opening previews from transcript history within the requested fork boundary when compaction omits them from model context.
- Use the shared `user_message_preview` helper when deriving previews from rollout items.

## Testing

Extend thread-fork integration tests to check previews in listings before and after a new turn, after reopening, and when forking at a boundary after compaction.

GitOrigin-RevId: 5962cf71ea7104f07c9013cc77259b105e4c60d7
This commit is contained in:
Eric Traut
2026-11-10 08:52:02 +00:00
committed by copyberry
parent ae5c75dbfc
commit f988a5f0da
11 changed files with 184 additions and 26 deletions
+10 -1
View File
@@ -138,7 +138,16 @@ impl LiveThread {
) -> ThreadStoreResult<Self> {
let thread_id = params.thread_id;
let history_mode = params.history_mode;
let metadata_sync = ThreadMetadataSync::for_create(&params).await;
let inherited_preview = if params.forked_from_id.is_some() {
thread_store
.read_pending_thread_metadata(thread_id)
.await?
.and_then(|patch| patch.preview)
} else {
None
};
let metadata_sync =
ThreadMetadataSync::for_create(&params, inherited_preview.as_deref()).await;
thread_store.create_thread(params).await?;
Ok(Self {
thread_id,
@@ -46,17 +46,19 @@ pub(super) async fn prepare(
operation: "prepare_fork",
});
}
let (history_base, model_context) = lineage
let (history_base, model_context, preview_before_compaction) = lineage
.read_with_fallback(store, |lineage| {
read_fork_history(store, thread_id, boundary, lineage)
})
.await?;
Ok(PreparedFork::with_source_reservation(
let mut fork = PreparedFork::with_source_reservation(
thread_id,
history_base,
Arc::new(model_context),
source_reservation,
))
);
fork.preview_before_compaction = preview_before_compaction;
Ok(fork)
}
async fn read_fork_history(
@@ -64,7 +66,7 @@ async fn read_fork_history(
thread_id: ThreadId,
boundary: &ForkBoundary,
lineage: RolloutLineage,
) -> ThreadStoreResult<(Option<HistoryPosition>, Vec<RolloutItem>)> {
) -> ThreadStoreResult<(Option<HistoryPosition>, Vec<RolloutItem>, Option<String>)> {
let source_segment = lineage
.segments()
.last()
@@ -97,8 +99,19 @@ async fn read_fork_history(
let history_base = history_base_at_boundary(store, thread_id, boundary, &lineage).await?;
drop(source_writer_guard);
let model_context = model_context::load_for_fork(lineage, history_base).await?;
Ok((history_base, model_context))
let model_context = model_context::load_for_fork(lineage.clone(), history_base).await?;
// A bounded replay starts with session metadata followed by the compaction. Its opening
// user message can only be recovered from the transcript within the same fork boundary.
let preview = if let [RolloutItem::SessionMeta(_), RolloutItem::Compacted(_), ..] =
model_context.as_slice()
&& let Some(end) = history_base
{
let prefix = lineage.truncate_at(end).await?;
super::thread_history::first_user_preview(store, &prefix).await?
} else {
None
};
Ok((history_base, model_context, preview))
}
pub(super) async fn history_base_at_boundary(
@@ -22,6 +22,7 @@ pub(super) use realtime::list_timeline;
pub(super) use search::search_thread_occurrences;
pub(super) use turn_lookup::find_source_turn;
pub(super) use turn_lookup::find_visible_turn;
pub(super) use turn_lookup::first_user_preview;
/// A valid complete rollout line with its absolute byte span in durable JSONL.
///
@@ -1,9 +1,16 @@
use codex_app_server_protocol::ThreadItemPayload;
use codex_app_server_protocol::UserInput;
use codex_protocol::ThreadId;
use codex_protocol::items::UserMessageItem;
use codex_protocol::protocol::user_message_preview;
use futures::TryStreamExt;
use sqlx::Row;
use super::super::LocalThreadStore;
use super::super::rollout_lineage::RolloutLineage;
use super::super::rollout_lineage::RolloutLineageSegment;
use super::sqlite_integer;
use super::thread_history_error;
use crate::ThreadStoreError;
use crate::ThreadStoreResult;
@@ -18,6 +25,50 @@ pub(in crate::local) struct TurnRow {
pub final_agent_item_id: Option<String>,
}
pub(in crate::local) async fn first_user_preview(
store: &LocalThreadStore,
prefix: &RolloutLineage,
) -> ThreadStoreResult<Option<String>> {
let pool = store.thread_history_db().await?;
for segment in prefix.for_history_page().await?.segments() {
let mut messages = sqlx::query_scalar::<_, String>(
r#"SELECT item_json FROM thread_items
WHERE thread_id = ?1 AND rollout_ordinal >= ?2 AND rollout_ordinal < ?3
AND updated_at_ordinal < ?3
AND (item_type = 'userMessage'
OR (item_type = '' AND json_extract(item_json, '$.type') = 'userMessage'))
ORDER BY rollout_ordinal"#,
)
.bind(segment.rollout_id().to_string())
.bind(sqlite_integer(segment.start_ordinal(), "rollout ordinal")?)
.bind(
segment
.end_ordinal()
.map(|end| sqlite_integer(end, "rollout ordinal"))
.transpose()?,
)
.fetch(pool);
while let Some(json) = messages.try_next().await.map_err(thread_history_error)? {
if let ThreadItemPayload::UserMessage {
id,
client_id,
content,
} = serde_json::from_str(&json).map_err(thread_history_error)?
{
let user = UserMessageItem {
id,
client_id,
content: content.into_iter().map(UserInput::into_core).collect(),
};
if let Some(preview) = user_message_preview(&user.as_legacy_user_message_event()) {
return Ok(Some(preview));
}
}
}
}
Ok(None)
}
pub(in crate::local) async fn find_source_turn(
pool: &sqlx::SqlitePool,
lineage: &RolloutLineage,
@@ -55,7 +55,10 @@ pub(crate) struct PendingThreadMetadataPatch {
}
impl ThreadMetadataSync {
pub(crate) async fn for_create(params: &CreateThreadParams) -> Self {
pub(crate) async fn for_create(
params: &CreateThreadParams,
inherited_preview: Option<&str>,
) -> Self {
let created_at = Utc::now();
let cwd = params.metadata.cwd.clone().unwrap_or_default();
let git_info = if get_git_repo_root(cwd.as_path()).is_some() {
@@ -93,7 +96,8 @@ impl ThreadMetadataSync {
Self {
thread_id: params.thread_id,
cwd_seen: !cwd.as_os_str().is_empty(),
preview_seen: guardian_review,
preview_seen: guardian_review
|| inherited_preview.is_some_and(|preview| !preview.is_empty()),
first_user_message_seen: guardian_review,
title_seen: guardian_review,
pending_update: Some(update),
@@ -485,7 +489,7 @@ mod tests {
#[tokio::test]
async fn create_metadata_records_originator_without_project() {
let thread_id = ThreadId::new();
let sync = ThreadMetadataSync::for_create(&CreateThreadParams {
let params = CreateThreadParams {
attached_thread: false,
creator_user_id: None,
creator_account_id: None,
@@ -511,8 +515,8 @@ mod tests {
model_provider: "test-provider".to_string(),
memory_mode: ThreadMemoryMode::Enabled,
},
})
.await;
};
let sync = ThreadMetadataSync::for_create(&params, /*inherited_preview*/ None).await;
let update = sync.take_pending_update().expect("pending metadata update");
assert_eq!(
+4
View File
@@ -244,6 +244,8 @@ pub struct PreparedFork {
pub history_base: Option<HistoryPosition>,
/// Bounded model context selected by the requested fork boundary.
pub model_context: Arc<Vec<RolloutItem>>,
/// Opening preview from the frozen transcript when compaction omitted it from model context.
pub preview_before_compaction: Option<String>,
/// Prevents the source thread from being deleted until the child's history reference is
/// finalized.
///
@@ -264,6 +266,7 @@ impl PreparedFork {
source_thread_id,
history_base,
model_context,
preview_before_compaction: None,
_source_reservation: Some(Box::new(source_reservation)),
}
}
@@ -279,6 +282,7 @@ impl PreparedFork {
source_thread_id,
history_base,
model_context,
preview_before_compaction: None,
_source_reservation: None,
}
}