Keep paginated thread Git metadata in SQLite (#34382)

## Why

Paginated rollouts contain only the initial Git metadata tuple. Reusing that
tuple during later reads, reconciliation, or resume can overwrite newer SQLite
values and undo explicit field clears.

## What changed

- Treat SQLite as authoritative for Git metadata once a thread is paginated.
- Apply paginated Git metadata patches directly to SQLite without appending
  compatibility metadata to the rollout.
- Preserve explicit null values when reading by rollout path or reconciling the
  initial session metadata, and ignore the initial Git tuple during resume.

## Testing

Added coverage for updating and clearing paginated Git metadata, preserving the
result through reconciliation and rollout-path reads, and avoiding stale Git
metadata updates on resume.

GitOrigin-RevId: 4d3e68cb24965b93d337d923092a98bac9b352b5
This commit is contained in:
Owen Lin authored and copyberry committed 2026-07-20 17:23:18 +00:00
1 parent 6b9a5592a6
commit 19b2273d8a
4 files changed
+225 -70

No files matched your search

+28 -1
View File
@@ -262,8 +262,20 @@ impl ThreadMetadataBuilder {
} }
impl ThreadMetadata { impl ThreadMetadata {
/// Preserve existing non-null Git fields when rollout-derived metadata is reconciled. /// Preserve SQLite-owned Git fields when rollout-derived metadata is reconciled.
pub fn prefer_existing_git_info(&mut self, existing: &Self) { pub fn prefer_existing_git_info(&mut self, existing: &Self) {
if matches!(self.history_mode, ThreadHistoryMode::Paginated)
&& matches!(existing.history_mode, ThreadHistoryMode::Paginated)
{
// `self` was rebuilt from the rollout's initial SessionMeta. `existing` is the
// current SQLite row. Once that row says paginated, metadata updates are SQLite-only,
// so a NULL is an explicit clear, not missing data. Copy the whole tuple or the stale
// rollout value would be written back during reconciliation.
self.git_sha = existing.git_sha.clone();
self.git_branch = existing.git_branch.clone();
self.git_origin_url = existing.git_origin_url.clone();
return;
}
if existing.git_sha.is_some() { if existing.git_sha.is_some() {
self.git_sha = existing.git_sha.clone(); self.git_sha = existing.git_sha.clone();
} }
@@ -668,4 +680,19 @@ mod tests {
assert!(ThreadMetadata::try_from(row).is_err()); assert!(ThreadMetadata::try_from(row).is_err());
} }
#[test]
fn paginated_rollout_git_info_keeps_rollout_values_until_sqlite_mode_is_paginated() {
let mut reconciled = expected_thread_metadata(/*reasoning_effort*/ None);
reconciled.history_mode = ThreadHistoryMode::Paginated;
reconciled.git_sha = Some("rollout-sha".to_string());
reconciled.git_branch = Some("rollout-branch".to_string());
reconciled.git_origin_url = Some("rollout-origin".to_string());
let existing = expected_thread_metadata(/*reasoning_effort*/ None);
let expected = reconciled.clone();
reconciled.prefer_existing_git_info(&existing);
assert_eq!(reconciled, expected);
}
} }
+23 -13
View File
@@ -125,20 +125,30 @@ pub(super) async fn read_thread_by_rollout_path(
thread.name = sqlite_thread_name(&metadata); thread.name = sqlite_thread_name(&metadata);
} }
thread.recency_at = metadata.recency_at; thread.recency_at = metadata.recency_at;
let existing_git_info = thread.git_info.take(); thread.git_info = if thread.history_mode == ThreadHistoryMode::Paginated {
let (fallback_sha, fallback_branch, fallback_origin_url) = match existing_git_info { // A paginated rollout only has the initial Git tuple. Do not turn an explicit SQLite
Some(info) => ( // clear back into the stale rollout value while reading by path.
info.commit_hash.map(|sha| sha.0), git_info_from_parts(
info.branch, metadata.git_sha,
info.repository_url, metadata.git_branch,
), metadata.git_origin_url,
None => (None, None, None), )
} else {
let (fallback_sha, fallback_branch, fallback_origin_url) = match thread.git_info.take()
{
Some(info) => (
info.commit_hash.map(|sha| sha.0),
info.branch,
info.repository_url,
),
None => (None, None, None),
};
git_info_from_parts(
metadata.git_sha.or(fallback_sha),
metadata.git_branch.or(fallback_branch),
metadata.git_origin_url.or(fallback_origin_url),
)
}; };
thread.git_info = git_info_from_parts(
metadata.git_sha.or(fallback_sha),
metadata.git_branch.or(fallback_branch),
metadata.git_origin_url.or(fallback_origin_url),
);
} }
reject_paginated_history(&thread, include_history)?; reject_paginated_history(&thread, include_history)?;
attach_history_if_requested(&mut thread, include_history).await?; attach_history_if_requested(&mut thread, include_history).await?;
@@ -28,7 +28,6 @@ use crate::ThreadMetadataPatch;
use crate::ThreadStoreError; use crate::ThreadStoreError;
use crate::ThreadStoreResult; use crate::ThreadStoreResult;
use crate::UpdateThreadMetadataParams; use crate::UpdateThreadMetadataParams;
use crate::error::reject_paginated_history_mode;
use crate::local::read_thread; use crate::local::read_thread;
struct ResolvedRolloutPath { struct ResolvedRolloutPath {
@@ -55,7 +54,8 @@ pub(super) async fn update_thread_metadata(
} }
let requires_rollout_compat = requires_rollout_compatibility_update(&patch); let requires_rollout_compat = requires_rollout_compatibility_update(&patch);
let history_mode = if patch.name.is_some() || requires_rollout_compat { let has_explicit_metadata = patch.name.is_some() || requires_rollout_compat;
let history_mode = if has_explicit_metadata {
Some( Some(
read_thread::read_thread( read_thread::read_thread(
store, store,
@@ -71,18 +71,9 @@ pub(super) async fn update_thread_metadata(
} else { } else {
None None
}; };
if requires_rollout_compat { let paginated = matches!(history_mode, Some(ThreadHistoryMode::Paginated));
// Explicit patches that require legacy rollout state must fail before either let require_sqlite_write = sqlite_write_failure_should_block(&patch) || paginated;
// persistence layer is mutated. let mut updated = apply_metadata_update(
if let Some(history_mode) = history_mode {
reject_paginated_history_mode(history_mode)?;
}
}
let paginated_name =
patch.name.is_some() && matches!(history_mode, Some(ThreadHistoryMode::Paginated));
let needs_rollout_compat = requires_rollout_compat || patch.name.is_some() && !paginated_name;
let require_sqlite_write = sqlite_write_failure_should_block(&patch) || paginated_name;
let updated = apply_metadata_update(
store, store,
thread_id, thread_id,
patch.clone(), patch.clone(),
@@ -91,18 +82,44 @@ pub(super) async fn update_thread_metadata(
history_mode, history_mode,
) )
.await?; .await?;
if paginated_name && let Some(name) = patch.name.as_ref() { if paginated
if let Err(err) = append_thread_name( && requires_rollout_compat
store.config.codex_home.as_path(), && let Some(git_info) = patch.git_info.as_ref()
thread_id, {
name.as_deref().unwrap_or_default(), // The generic upsert preserves non-null Git fields for rollout reconciliation. Apply the
// explicit patch afterward so clears are written to SQLite too.
let Some(state_db) = store.state_db().await else {
return Err(ThreadStoreError::Internal {
message: format!("sqlite state db unavailable for thread {thread_id}"),
});
};
apply_thread_git_info_patch(state_db.as_ref(), thread_id, git_info).await?;
updated = read_thread::read_thread(
store,
ReadThreadParams {
thread_id,
include_archived: params.include_archived,
include_history: false,
},
) )
.await .await?;
}
if paginated {
// Paginated metadata lives in SQLite. Keep the name index update, then stop before the
// legacy SessionMeta compatibility path below.
if let Some(name) = patch.name.as_ref()
&& let Err(err) = append_thread_name(
store.config.codex_home.as_path(),
thread_id,
name.as_deref().unwrap_or_default(),
)
.await
{ {
warn!("failed to index paginated thread name for {thread_id}: {err}"); warn!("failed to index paginated thread name for {thread_id}: {err}");
} }
return Ok(updated); return Ok(updated);
} }
let needs_rollout_compat = requires_rollout_compat || patch.name.is_some();
if !needs_rollout_compat { if !needs_rollout_compat {
return Ok(updated); return Ok(updated);
} }
@@ -284,6 +301,11 @@ async fn apply_metadata_update(
if let Some(rollout_path) = rollout_path { if let Some(rollout_path) = rollout_path {
metadata.rollout_path = rollout_path; metadata.rollout_path = rollout_path;
} }
if let Some(history_mode) = history_mode {
// The read above gets the canonical mode from the rollout. Persist it before an
// explicit paginated patch makes SQLite metadata authoritative.
metadata.history_mode = history_mode;
}
if let Some(preview) = patch.preview { if let Some(preview) = patch.preview {
metadata.preview = Some(preview); metadata.preview = Some(preview);
} }
@@ -565,6 +587,34 @@ fn normalize_cwd(cwd: PathBuf) -> PathBuf {
codex_utils_path::normalize_for_path_comparison(cwd.as_path()).unwrap_or(cwd) codex_utils_path::normalize_for_path_comparison(cwd.as_path()).unwrap_or(cwd)
} }
async fn apply_thread_git_info_patch(
state_db: &codex_state::StateRuntime,
thread_id: ThreadId,
git_info: &GitInfoPatch,
) -> ThreadStoreResult<()> {
let updated = state_db
.update_thread_git_info(
thread_id,
git_info.sha.as_ref().map(|sha| sha.as_deref()),
git_info.branch.as_ref().map(|branch| branch.as_deref()),
git_info
.origin_url
.as_ref()
.map(|origin_url| origin_url.as_deref()),
)
.await
.map_err(|err| ThreadStoreError::Internal {
message: format!("failed to update git metadata for thread {thread_id}: {err}"),
})?;
if updated {
Ok(())
} else {
Err(ThreadStoreError::Internal {
message: format!("thread metadata unavailable before git update: {thread_id}"),
})
}
}
async fn apply_thread_git_info( async fn apply_thread_git_info(
store: &LocalThreadStore, store: &LocalThreadStore,
thread_id: ThreadId, thread_id: ThreadId,
@@ -934,7 +984,7 @@ mod tests {
} }
#[tokio::test] #[tokio::test]
async fn update_thread_metadata_rejects_paginated_rollout_compatibility_writes() { async fn update_thread_metadata_updates_paginated_git_info_in_sqlite_only() {
let home = TempDir::new().expect("temp dir"); let home = TempDir::new().expect("temp dir");
let config = test_config(home.path()); let config = test_config(home.path());
let uuid = Uuid::from_u128(303); let uuid = Uuid::from_u128(303);
@@ -953,52 +1003,86 @@ mod tests {
) )
.await .await
.expect("state db should initialize"); .expect("state db should initialize");
let store = LocalThreadStore::new(config, Some(runtime.clone())); codex_rollout::state_db::reconcile_rollout(
Some(runtime.as_ref()),
path.as_path(),
config.default_model_provider_id.as_str(),
/*builder*/ None,
&[],
/*archived_only*/ None,
/*new_thread_memory_mode*/ None,
)
.await;
let mut stale_metadata = runtime
.get_thread(thread_id)
.await
.expect("read metadata")
.expect("thread metadata");
stale_metadata.history_mode = ThreadHistoryMode::Legacy;
runtime
.upsert_thread(&stale_metadata)
.await
.expect("seed stale history mode");
let store = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
assert!(matches!( let thread = store
store .update_thread_metadata(UpdateThreadMetadataParams {
.update_thread_metadata(UpdateThreadMetadataParams { thread_id,
thread_id, patch: ThreadMetadataPatch {
patch: ThreadMetadataPatch { git_info: Some(GitInfoPatch {
name: Some(Some("Must not persist".to_string())), sha: Some(None),
memory_mode: Some(ThreadMemoryMode::Disabled), branch: Some(Some("feature".to_string())),
..Default::default() ..Default::default()
}, }),
include_archived: false, ..Default::default()
}) },
.await include_archived: false,
.expect_err("paginated rollout compatibility write should fail"), })
ThreadStoreError::Unsupported { .await
operation: "paginated_threads" .expect("paginated metadata update");
}
));
let git_info = thread.git_info.expect("git info");
assert_eq!(git_info.commit_hash, None);
assert_eq!(git_info.branch.as_deref(), Some("feature"));
assert_eq!( assert_eq!(
std::fs::read_to_string(&path).expect("read rollout"), git_info.repository_url.as_deref(),
original_rollout Some("https://example.com/repo.git")
);
assert_eq!(
runtime
.get_thread_memory_mode(thread_id)
.await
.expect("thread memory mode should be readable")
.as_deref(),
Some("enabled")
); );
assert_eq!( assert_eq!(
runtime runtime
.get_thread(thread_id) .get_thread(thread_id)
.await .await
.expect("thread metadata should be readable") .expect("read metadata")
.expect("thread metadata") .expect("thread metadata")
.name, .history_mode,
None ThreadHistoryMode::Paginated
); );
assert_eq!( assert_eq!(
codex_rollout::find_thread_name_by_id(home.path(), &thread_id) std::fs::read_to_string(&path).expect("read rollout"),
.await original_rollout
.expect("find thread name"), );
None codex_rollout::state_db::reconcile_rollout(
Some(runtime.as_ref()),
path.as_path(),
config.default_model_provider_id.as_str(),
/*builder*/ None,
&[],
/*archived_only*/ None,
/*new_thread_memory_mode*/ None,
)
.await;
let thread = store
.read_thread_by_rollout_path(
path, /*include_archived*/ false, /*include_history*/ false,
)
.await
.expect("read paginated thread by rollout path");
let git_info = thread.git_info.expect("git info");
assert_eq!(git_info.commit_hash, None);
assert_eq!(git_info.branch.as_deref(), Some("feature"));
assert_eq!(
git_info.repository_url.as_deref(),
Some("https://example.com/repo.git")
); );
} }
@@ -11,6 +11,7 @@ use codex_protocol::items::TurnItem;
use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::GitInfo; use codex_protocol::protocol::GitInfo;
use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadMemoryMode; use codex_protocol::protocol::ThreadMemoryMode;
use codex_protocol::protocol::UserMessageEvent; use codex_protocol::protocol::UserMessageEvent;
use codex_protocol::protocol::strip_user_message_prefix; use codex_protocol::protocol::strip_user_message_prefix;
@@ -20,6 +21,7 @@ use crate::CreateThreadParams;
use crate::GitInfoPatch; use crate::GitInfoPatch;
use crate::ResumeThreadParams; use crate::ResumeThreadParams;
use crate::ThreadMetadataPatch; use crate::ThreadMetadataPatch;
use crate::types::canonical_history_mode_from_rollout_items;
const THREAD_UPDATED_AT_TOUCH_INTERVAL: Duration = Duration::from_secs(5); const THREAD_UPDATED_AT_TOUCH_INTERVAL: Duration = Duration::from_secs(5);
@@ -193,7 +195,16 @@ impl ThreadMetadataSync {
} }
fn observe_resume_history(&mut self, items: &[RolloutItem]) -> Option<ThreadMetadataPatch> { fn observe_resume_history(&mut self, items: &[RolloutItem]) -> Option<ThreadMetadataPatch> {
self.observe_items_with_update(items, ThreadMetadataPatch::default()) let mut update = self.observe_items_with_update(items, ThreadMetadataPatch::default())?;
if matches!(
canonical_history_mode_from_rollout_items(items),
ThreadHistoryMode::Paginated
) {
// Paginated rollouts never append metadata-only SessionMeta updates. Do not reapply
// the initial Git tuple when resume history is flushed after the first append.
update.git_info = None;
}
Some(update)
} }
fn observe_items_with_update( fn observe_items_with_update(
@@ -663,6 +674,29 @@ mod tests {
); );
} }
#[test]
fn paginated_resume_history_does_not_reapply_initial_git_info() {
let thread_id = ThreadId::new();
let mut meta = session_meta(thread_id);
meta.meta.history_mode = ThreadHistoryMode::Paginated;
meta.git = Some(GitInfo {
commit_hash: None,
branch: Some("stale-branch".to_string()),
repository_url: None,
});
let sync = ThreadMetadataSync::for_resume(&resume_params(
thread_id,
vec![
RolloutItem::SessionMeta(meta),
RolloutItem::EventMsg(EventMsg::UserMessage(user_message("hello metadata"))),
],
));
let update = sync.take_pending_update().expect("pending metadata update");
assert_eq!(update.patch.git_info, None);
assert_eq!(update.patch.preview.as_deref(), Some("hello metadata"));
}
fn resume_params(thread_id: ThreadId, history: Vec<RolloutItem>) -> ResumeThreadParams { fn resume_params(thread_id: ThreadId, history: Vec<RolloutItem>) -> ResumeThreadParams {
ResumeThreadParams { ResumeThreadParams {
thread_id, thread_id,