agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(app-server): add history_mode to thread (#29927)

Browse Source 
## Description

This PR adds a new `historyMode = "legacy" | "paginated"` to `Thread`.
This will be stored in `SessionMeta` in the JSONL rollout file and as a
new column in the SQLite thread_metadata table, and exposed on
`thread/start` and on the `Thread` object in app-server.

## What changed

- Added canonical `ThreadHistoryMode` with `legacy` and `paginated`,
defaulting old and new SessionMeta to `legacy`.
- Carried `history_mode` through core session config, ThreadStore stored
metadata, local/in-memory stores, rollout metadata extraction, and the
existing SQLite `threads` table.
- Added experimental `historyMode` to app-server v2 `Thread` and
`thread/start`.
- Made paginated stored threads metadata-discoverable but unsupported
for legacy full-history reads, `load_history`, live resume, and create
paths.
- Regenerated app-server schema fixtures and added
protocol/state/thread-store/app-server coverage for persistence and
fail-closed behavior.

## Compatibility floor
Because users may be running various versions of Codex binaries on the
same machine (TUI, Codex App, etc.), we will need to establish a
compatibility floor for upcoming paginated threads, which will change
how thread storage reads and writes work.

The overall plan here:
```
Release N:
- Add historyMode to SessionMeta / Thread / SQLite metadata.
- Teach binaries to understand paginated threads.
- If a binary sees `historyMode="paginated"` but does not support the paginated contract, it refuses to resume/mutate the thread.
- Default remains `"legacy"`.

Release N+1:
- First-party clients start opting into paginated threads where appropriate.
- Internal dogfood / staged rollout.
- Measure old-client usage and paginated-thread unsupported errors.

Release N+2:
- Only after Release N+ is overwhelmingly deployed, make paginated the default.
- Accept that a small tail of N-1-or-older binaries may not understand paginated threads.
```

The important behavior change is fail-closed handling for a binary that
encounters a persisted `paginated` thread before it knows how to fully
support paginated history. In app-server, if a thread is `paginated`, we
will:

- allow metadata-only discovery paths like `thread/list` and
`thread/read(includeTurns=false)`, so clients can still see the thread
and inspect its `historyMode`
- reject legacy full-history/live-thread paths like
`thread/read(includeTurns=true)` and `thread/resume` with an unsupported
JSON-RPC error
- avoid silently treating an unknown or future `historyMode` as `legacy`

Under the hood, the ThreadStore layer also rejects legacy operations
that would need to load or replay the full thread history for a
paginated thread. That gives us the behavior we want for Release N:
future paginated threads are visible, but this binary fails closed
instead of trying to operate on them as if they were legacy threads.
This commit is contained in:
Owen Lin
2026-06-26 09:12:42 -07:00
committed by GitHub
parent 2c5bc5e284
commit 5267e805fb
91 changed files with 1385 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/thread-store/src/error.rs
+12
View File
@@ -1,8 +1,20 @@
use codex_protocol::ThreadId;
use codex_protocol::protocol::ThreadHistoryMode;
/// Result type returned by thread-store operations.
pub type ThreadStoreResult<T> = Result<T, ThreadStoreError>;
pub(crate) fn reject_paginated_history_mode(
history_mode: ThreadHistoryMode,
) -> ThreadStoreResult<()> {
if matches!(history_mode, ThreadHistoryMode::Paginated) {
return Err(ThreadStoreError::Unsupported {
operation: "paginated_threads",
});
}
Ok(())
}
/// Error type shared by thread-store implementations.
#[derive(Debug, thiserror::Error)]
pub enum ThreadStoreError {
codex-rs/thread-store/src/in_memory.rs
+197 -8
View File
@@ -14,6 +14,7 @@ use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionContextWindow;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadMemoryMode;
use codex_rollout::persisted_rollout_items;
@@ -36,6 +37,8 @@ use crate::ThreadStoreError;
use crate::ThreadStoreFuture;
use crate::ThreadStoreResult;
use crate::UpdateThreadMetadataParams;
use crate::error::reject_paginated_history_mode;
use crate::types::canonical_history_mode_from_rollout_items;
static IN_MEMORY_THREAD_STORES: OnceLock<Mutex<HashMap<String, Arc<InMemoryThreadStore>>>> =
OnceLock::new();
@@ -128,6 +131,7 @@ mod tests {
dynamic_tools: Vec::new(),
selected_capability_roots: Vec::new(),
multi_agent_version: None,
history_mode: ThreadHistoryMode::Legacy,
initial_window_id: uuid::Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
cwd: None,
@@ -193,6 +197,152 @@ mod tests {
HashSet::from([child_thread_id, grandchild_thread_id])
);
}
#[tokio::test]
async fn paginated_threads_allow_metadata_reads_and_reject_legacy_history_paths() {
let store = InMemoryThreadStore::default();
let thread_id = ThreadId::default();
let rollout_path = PathBuf::from("/tmp/paginated-thread.jsonl");
store
.create_thread(create_thread_params(thread_id, ThreadHistoryMode::Legacy))
.await
.expect("create legacy thread");
store
.resume_thread(ResumeThreadParams {
thread_id,
rollout_path: Some(rollout_path.clone()),
history: None,
include_archived: false,
metadata: thread_metadata(),
})
.await
.expect("register rollout path");
store
.update_thread_metadata(UpdateThreadMetadataParams {
thread_id,
patch: ThreadMetadataPatch {
history_mode: Some(ThreadHistoryMode::Paginated),
..Default::default()
},
include_archived: false,
})
.await
.expect("seed paginated metadata");
let thread = store
.read_thread(ReadThreadParams {
thread_id,
include_archived: false,
include_history: false,
})
.await
.expect("metadata read");
assert_eq!(thread.history_mode, ThreadHistoryMode::Paginated);
assert!(thread.history.is_none());
let thread = store
.read_thread_by_rollout_path(ReadThreadByRolloutPathParams {
rollout_path,
include_archived: false,
include_history: false,
})
.await
.expect("metadata path read");
assert_eq!(thread.history_mode, ThreadHistoryMode::Paginated);
assert!(thread.history.is_none());
assert_paginated_threads_unsupported(
store
.read_thread(ReadThreadParams {
thread_id,
include_archived: false,
include_history: true,
})
.await
.expect_err("full history read should fail"),
);
assert_paginated_threads_unsupported(
store
.read_thread_by_rollout_path(ReadThreadByRolloutPathParams {
rollout_path: PathBuf::from("/tmp/paginated-thread.jsonl"),
include_archived: false,
include_history: true,
})
.await
.expect_err("full history path read should fail"),
);
assert_paginated_threads_unsupported(
store
.load_history(LoadThreadHistoryParams {
thread_id,
include_archived: false,
})
.await
.expect_err("history load should fail"),
);
assert_paginated_threads_unsupported(
store
.resume_thread(ResumeThreadParams {
thread_id,
rollout_path: None,
history: None,
include_archived: false,
metadata: thread_metadata(),
})
.await
.expect_err("resume should fail"),
);
assert_paginated_threads_unsupported(
store
.create_thread(create_thread_params(
ThreadId::default(),
ThreadHistoryMode::Paginated,
))
.await
.expect_err("paginated create should fail"),
);
}
fn create_thread_params(
thread_id: ThreadId,
history_mode: ThreadHistoryMode,
) -> CreateThreadParams {
CreateThreadParams {
session_id: thread_id.into(),
thread_id,
extra_config: None,
forked_from_id: None,
parent_thread_id: None,
source: SessionSource::Exec,
thread_source: None,
originator: "test_originator".to_string(),
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
selected_capability_roots: Vec::new(),
multi_agent_version: None,
history_mode,
initial_window_id: uuid::Uuid::now_v7().to_string(),
metadata: thread_metadata(),
}
}
fn thread_metadata() -> ThreadPersistenceMetadata {
ThreadPersistenceMetadata {
cwd: None,
model_provider: "test-provider".to_string(),
memory_mode: ThreadMemoryMode::Enabled,
}
}
fn assert_paginated_threads_unsupported(err: ThreadStoreError) {
assert!(matches!(
err,
ThreadStoreError::Unsupported {
operation: "paginated_threads"
}
));
}
}
fn stores_guard() -> MutexGuard<'static, HashMap<String, Arc<InMemoryThreadStore>>> {
@@ -265,6 +415,7 @@ impl InMemoryThreadStore {
}
async fn create_thread(&self, params: CreateThreadParams) -> ThreadStoreResult<()> {
reject_paginated_history_mode(params.history_mode)?;
let mut state = self.state.lock().await;
state.calls.create_thread += 1;
let session_meta = SessionMeta {
@@ -285,6 +436,7 @@ impl InMemoryThreadStore {
selected_capability_roots: params.selected_capability_roots.clone(),
memory_mode: matches!(params.metadata.memory_mode, ThreadMemoryMode::Disabled)
.then_some("disabled".to_string()),
history_mode: params.history_mode,
multi_agent_version: params.multi_agent_version,
context_window: Some(SessionContextWindow::new(params.initial_window_id.clone())),
..SessionMeta::default()
@@ -304,6 +456,13 @@ impl InMemoryThreadStore {
async fn resume_thread(&self, params: ResumeThreadParams) -> ThreadStoreResult<()> {
let mut state = self.state.lock().await;
state.calls.resume_thread += 1;
let history_mode = params
.history
.as_deref()
.map(Vec::as_slice)
.map(canonical_history_mode_from_rollout_items)
.unwrap_or_else(|| history_mode_from_state(&state, params.thread_id));
reject_paginated_history_mode(history_mode)?;
if let Some(history) = params.history {
state
.histories
@@ -338,14 +497,18 @@ impl InMemoryThreadStore {
) -> ThreadStoreResult<StoredThreadHistory> {
let mut state = self.state.lock().await;
state.calls.load_history += 1;
let items = state.histories.get(&params.thread_id).cloned().ok_or(
ThreadStoreError::ThreadNotFound {
thread_id: params.thread_id,
},
)?;
let items =
state
.histories
.get(&params.thread_id)
.ok_or(ThreadStoreError::ThreadNotFound {
thread_id: params.thread_id,
})?;
let history_mode = history_mode_from_state(&state, params.thread_id);
reject_paginated_history_mode(history_mode)?;
Ok(StoredThreadHistory {
thread_id: params.thread_id,
items,
items: items.clone(),
})
}
@@ -354,8 +517,10 @@ impl InMemoryThreadStore {
state.calls.read_thread += 1;
if params.include_history {
state.calls.read_thread_with_history += 1;
reject_paginated_history_mode(history_mode_from_state(&state, params.thread_id))?;
}
stored_thread_from_state(&state, params.thread_id, params.include_history)
let thread = stored_thread_from_state(&state, params.thread_id, params.include_history)?;
Ok(thread)
}
async fn read_thread_by_rollout_path(
@@ -372,7 +537,11 @@ impl InMemoryThreadStore {
),
});
};
stored_thread_from_state(&state, thread_id, params.include_history)
if params.include_history {
reject_paginated_history_mode(history_mode_from_state(&state, thread_id))?;
}
let thread = stored_thread_from_state(&state, thread_id, params.include_history)?;
Ok(thread)
}
async fn list_threads(&self) -> ThreadStoreResult<ThreadPage> {
@@ -615,6 +784,9 @@ fn stored_thread_from_state(
source: metadata
.and_then(|metadata| metadata.source.clone())
.unwrap_or_else(|| created.source.clone()),
history_mode: metadata
.and_then(|metadata| metadata.history_mode)
.unwrap_or(created.history_mode),
thread_source: metadata
.and_then(|metadata| metadata.thread_source.clone())
.unwrap_or_else(|| created.thread_source.clone()),
@@ -634,6 +806,23 @@ fn stored_thread_from_state(
})
}
fn history_mode_from_state(
state: &InMemoryThreadStoreState,
thread_id: ThreadId,
) -> ThreadHistoryMode {
state
.metadata_updates
.get(&thread_id)
.and_then(|metadata| metadata.history_mode)
.or_else(|| {
state
.created_threads
.get(&thread_id)
.map(|thread| thread.history_mode)
})
.unwrap_or_default()
}
fn git_info_from_patch(patch: &ThreadMetadataPatch) -> Option<codex_protocol::protocol::GitInfo> {
let git_info = patch.git_info.as_ref()?;
let sha = git_info.sha.clone().flatten();
codex-rs/thread-store/src/local/create_thread.rs
+3
View File
@@ -2,6 +2,7 @@ use super::LocalThreadStore;
use crate::CreateThreadParams;
use crate::ThreadStoreError;
use crate::ThreadStoreResult;
use crate::error::reject_paginated_history_mode;
use codex_protocol::protocol::ThreadMemoryMode;
use codex_rollout::RolloutConfig;
use codex_rollout::RolloutRecorder;
@@ -11,6 +12,7 @@ pub(super) async fn create_thread(
store: &LocalThreadStore,
params: CreateThreadParams,
) -> ThreadStoreResult<RolloutRecorder> {
reject_paginated_history_mode(params.history_mode)?;
let cwd = params
.metadata
.cwd
@@ -40,6 +42,7 @@ pub(super) async fn create_thread(
.with_session_id(params.session_id)
.with_selected_capability_roots(params.selected_capability_roots)
.with_multi_agent_version(params.multi_agent_version)
.with_history_mode(params.history_mode)
.with_initial_window_id(params.initial_window_id),
)
.await
codex-rs/thread-store/src/local/helpers.rs
+1
View File
@@ -142,6 +142,7 @@ pub(super) fn stored_thread_from_rollout_item(
cwd: item.cwd.unwrap_or_default(),
cli_version: item.cli_version.unwrap_or_default(),
source,
history_mode: item.history_mode,
thread_source: None,
agent_nickname: item.agent_nickname,
agent_role: item.agent_role,
codex-rs/thread-store/src/local/list_threads.rs
+2
View File
@@ -219,6 +219,7 @@ mod tests {
use chrono::Utc;
use codex_protocol::ThreadId;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadHistoryMode;
use pretty_assertions::assert_eq;
use std::fs;
use tempfile::TempDir;
@@ -243,6 +244,7 @@ mod tests {
Uuid::from_u128(102),
"Hello from user",
/*model_provider*/ None,
ThreadHistoryMode::Legacy,
)
.expect("session file");
codex-rs/thread-store/src/local/live_writer.rs
+26 -1
View File
@@ -16,6 +16,8 @@ use crate::ReadThreadParams;
use crate::ResumeThreadParams;
use crate::ThreadStoreError;
use crate::ThreadStoreResult;
use crate::error::reject_paginated_history_mode;
use crate::types::canonical_history_mode_from_rollout_items;
const ROLLOUT_SIZE_BYTES_METRIC: &str = "codex.rollout.size_bytes";
@@ -34,6 +36,30 @@ pub(super) async fn resume_thread(
params: ResumeThreadParams,
) -> ThreadStoreResult<()> {
store.ensure_live_recorder_absent(params.thread_id).await?;
let history_mode = if let Some(history) = params.history.as_deref() {
canonical_history_mode_from_rollout_items(history)
} else if let Some(rollout_path) = params.rollout_path.as_ref() {
super::read_thread::read_thread_by_rollout_path(
store,
rollout_path.clone(),
params.include_archived,
/*include_history*/ false,
)
.await?
.history_mode
} else {
super::read_thread::read_thread(
store,
ReadThreadParams {
thread_id: params.thread_id,
include_archived: params.include_archived,
include_history: false,
},
)
.await?
.history_mode
};
reject_paginated_history_mode(history_mode)?;
let rollout_path = match (params.rollout_path, params.history) {
(Some(rollout_path), _history) => rollout_path,
(None, history) => {
@@ -46,7 +72,6 @@ pub(super) async fn resume_thread(
},
)
.await?;
thread
.rollout_path
.ok_or_else(|| ThreadStoreError::Internal {
codex-rs/thread-store/src/local/mod.rs
+103 -1
View File
@@ -321,6 +321,7 @@ mod tests {
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadMemoryMode;
use codex_protocol::protocol::TurnCompleteEvent;
use codex_protocol::protocol::TurnStartedEvent;
@@ -333,6 +334,7 @@ mod tests {
use crate::local::test_support::test_config;
use crate::local::test_support::write_archived_session_file;
use crate::local::test_support::write_session_file;
use crate::local::test_support::write_session_file_with_history_mode;
#[tokio::test]
async fn live_writer_lifecycle_writes_and_closes() {
@@ -1109,18 +1111,108 @@ mod tests {
.expect("read thread by rollout path");
assert_eq!(thread.thread_id, thread_id);
assert_eq!(thread.history_mode, ThreadHistoryMode::Legacy);
assert_eq!(
thread
.history
.as_ref()
.expect("history")
.items
.into_iter()
.iter()
.filter(|item| matches!(item, RolloutItem::EventMsg(EventMsg::UserMessage(_))))
.count(),
1
);
}
#[tokio::test]
async fn paginated_threads_allow_metadata_reads_and_reject_legacy_history_paths() {
let home = TempDir::new().expect("temp dir");
let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
let uuid = uuid::Uuid::from_u128(408);
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
let rollout_path = write_session_file_with_history_mode(
home.path(),
"2025-01-04T12-00-00",
uuid,
ThreadHistoryMode::Paginated,
)
.expect("session file");
let thread = store
.read_thread(ReadThreadParams {
thread_id,
include_archived: false,
include_history: false,
})
.await
.expect("metadata read");
assert_eq!(thread.history_mode, ThreadHistoryMode::Paginated);
assert!(thread.history.is_none());
let thread = store
.read_thread_by_rollout_path(
rollout_path.clone(),
/*include_archived*/ true,
/*include_history*/ false,
)
.await
.expect("metadata path read");
assert_eq!(thread.history_mode, ThreadHistoryMode::Paginated);
assert!(thread.history.is_none());
assert_paginated_threads_unsupported(
store
.read_thread(ReadThreadParams {
thread_id,
include_archived: false,
include_history: true,
})
.await
.expect_err("full history read should fail"),
);
assert_paginated_threads_unsupported(
store
.read_thread_by_rollout_path(
rollout_path.clone(),
/*include_archived*/ true,
/*include_history*/ true,
)
.await
.expect_err("full history path read should fail"),
);
assert_paginated_threads_unsupported(
store
.load_history(LoadThreadHistoryParams {
thread_id,
include_archived: false,
})
.await
.expect_err("history load should fail"),
);
assert_paginated_threads_unsupported(
store
.resume_thread(ResumeThreadParams {
thread_id,
rollout_path: Some(rollout_path),
history: None,
include_archived: false,
metadata: thread_metadata(),
})
.await
.expect_err("resume should fail"),
);
let mut create_params = create_thread_params(ThreadId::default());
create_params.history_mode = ThreadHistoryMode::Paginated;
assert_paginated_threads_unsupported(
store
.create_thread(create_params)
.await
.expect_err("paginated create should fail"),
);
}
fn create_thread_params(thread_id: ThreadId) -> CreateThreadParams {
CreateThreadParams {
session_id: thread_id.into(),
@@ -1135,11 +1227,21 @@ mod tests {
dynamic_tools: Vec::new(),
selected_capability_roots: Vec::new(),
multi_agent_version: None,
history_mode: ThreadHistoryMode::Legacy,
initial_window_id: uuid::Uuid::now_v7().to_string(),
metadata: thread_metadata(),
}
}
fn assert_paginated_threads_unsupported(err: ThreadStoreError) {
assert!(matches!(
err,
ThreadStoreError::Unsupported {
operation: "paginated_threads"
}
));
}
fn thread_metadata() -> ThreadPersistenceMetadata {
ThreadPersistenceMetadata {
cwd: Some(std::env::current_dir().expect("cwd")),
codex-rs/thread-store/src/local/read_thread.rs
+63 -23
View File
@@ -25,6 +25,7 @@ use crate::StoredThread;
use crate::StoredThreadHistory;
use crate::ThreadStoreError;
use crate::ThreadStoreResult;
use crate::error::reject_paginated_history_mode;
pub(super) async fn read_thread(
store: &LocalThreadStore,
@@ -47,7 +48,7 @@ pub(super) async fn read_thread(
.await)
{
let metadata_sandbox_policy = metadata.sandbox_policy.clone();
let mut thread = stored_thread_from_sqlite_metadata(store, metadata).await;
let mut thread = stored_thread_from_sqlite_metadata(store, metadata).await?;
if !params.include_history
&& let Some(rollout_path) = thread.rollout_path.clone()
&& let Ok(mut rollout_thread) = read_thread_from_rollout_path(store, rollout_path).await
@@ -66,6 +67,7 @@ pub(super) async fn read_thread(
);
thread = rollout_thread;
}
reject_paginated_history(&thread, params.include_history)?;
attach_history_if_requested(&mut thread, params.include_history).await?;
return Ok(thread);
}
@@ -82,6 +84,7 @@ pub(super) async fn read_thread(
message: format!("thread {} is archived", thread.thread_id),
});
}
reject_paginated_history(&thread, params.include_history)?;
attach_history_if_requested(&mut thread, params.include_history).await?;
Ok(thread)
}
@@ -132,10 +135,18 @@ pub(super) async fn read_thread_by_rollout_path(
metadata.git_origin_url.or(fallback_origin_url),
);
}
reject_paginated_history(&thread, include_history)?;
attach_history_if_requested(&mut thread, include_history).await?;
Ok(thread)
}
fn reject_paginated_history(thread: &StoredThread, include_history: bool) -> ThreadStoreResult<()> {
if include_history {
reject_paginated_history_mode(thread.history_mode)?;
}
Ok(())
}
async fn resolve_requested_rollout_path(
store: &LocalThreadStore,
rollout_path: std::path::PathBuf,
@@ -261,16 +272,16 @@ async fn read_thread_from_rollout_path(
message: format!("failed to read thread id from {}", path.display()),
})?;
thread.rollout_path = Some(codex_rollout::plain_rollout_path(path.as_path()));
if let Ok(meta_line) = read_session_meta_line(path.as_path()).await {
thread.forked_from_id = meta_line.meta.forked_from_id;
thread.parent_thread_id = meta_line.meta.parent_thread_id;
if let Some(model_provider) = meta_line
.meta
.model_provider
.filter(|provider| !provider.is_empty())
{
thread.model_provider = model_provider;
}
let meta_line = read_required_session_meta_line(path.as_path()).await?;
thread.forked_from_id = meta_line.meta.forked_from_id;
thread.parent_thread_id = meta_line.meta.parent_thread_id;
thread.history_mode = meta_line.meta.history_mode;
if let Some(model_provider) = meta_line
.meta
.model_provider
.filter(|provider| !provider.is_empty())
{
thread.model_provider = model_provider;
}
if let Ok(Some(title)) =
find_thread_name_by_id(store.config.codex_home.as_path(), &thread.thread_id).await
@@ -302,7 +313,7 @@ async fn read_sqlite_metadata(
async fn stored_thread_from_sqlite_metadata(
store: &LocalThreadStore,
metadata: ThreadMetadata,
) -> StoredThread {
) -> ThreadStoreResult<StoredThread> {
let name = match distinct_thread_metadata_title(&metadata) {
Some(title) => Some(title),
None => find_thread_name_by_id(store.config.codex_home.as_path(), &metadata.id)
@@ -311,13 +322,32 @@ async fn stored_thread_from_sqlite_metadata(
.flatten()
.filter(|title| !title.trim().is_empty()),
};
let session_meta = read_session_meta_line(metadata.rollout_path.as_path())
.await
.ok()
.map(|meta_line| meta_line.meta);
let session_meta = match read_required_session_meta_line(metadata.rollout_path.as_path()).await
{
Ok(meta_line) => Some(meta_line.meta),
Err(_)
if codex_rollout::existing_rollout_path(metadata.rollout_path.as_path())
.await
.is_none() =>
{
None
}
Err(err) => {
return Err(ThreadStoreError::Internal {
message: format!(
"failed to read session metadata {}: {err}",
metadata.rollout_path.display()
),
});
}
};
let rollout_path = codex_rollout::plain_rollout_path(metadata.rollout_path.as_path());
let forked_from_id = session_meta.as_ref().and_then(|meta| meta.forked_from_id);
let parent_thread_id = session_meta.as_ref().and_then(|meta| meta.parent_thread_id);
let history_mode = session_meta
.as_ref()
.map(|meta| meta.history_mode)
.unwrap_or(metadata.history_mode);
let preview = metadata
.preview
.clone()
@@ -325,7 +355,7 @@ async fn stored_thread_from_sqlite_metadata(
.unwrap_or_default();
let permission_profile =
permission_profile_from_metadata_value(&metadata.sandbox_policy, metadata.cwd.as_path());
StoredThread {
Ok(StoredThread {
thread_id: metadata.id,
extra_config: None,
rollout_path: Some(rollout_path),
@@ -347,6 +377,7 @@ async fn stored_thread_from_sqlite_metadata(
cwd: metadata.cwd,
cli_version: metadata.cli_version,
source: parse_session_source(&metadata.source),
history_mode,
thread_source: metadata.thread_source,
agent_nickname: metadata.agent_nickname,
agent_role: metadata.agent_role,
@@ -361,24 +392,30 @@ async fn stored_thread_from_sqlite_metadata(
token_usage: None,
first_user_message: metadata.first_user_message,
history: None,
}
})
}
async fn stored_thread_from_session_meta(
store: &LocalThreadStore,
path: std::path::PathBuf,
) -> ThreadStoreResult<StoredThread> {
let meta_line = read_session_meta_line(path.as_path())
.await
.map_err(|err| ThreadStoreError::Internal {
message: format!("failed to read thread {}: {err}", path.display()),
})?;
let meta_line = read_required_session_meta_line(path.as_path()).await?;
let archived = rollout_path_is_archived(store.config.codex_home.as_path(), path.as_path());
Ok(stored_thread_from_meta_line(
store, meta_line, path, archived,
))
}
async fn read_required_session_meta_line(
path: &std::path::Path,
) -> ThreadStoreResult<SessionMetaLine> {
read_session_meta_line(path)
.await
.map_err(|err| ThreadStoreError::Internal {
message: format!("failed to read session metadata {}: {err}", path.display()),
})
}
fn stored_thread_from_meta_line(
store: &LocalThreadStore,
meta_line: SessionMetaLine,
@@ -414,6 +451,7 @@ fn stored_thread_from_meta_line(
cwd: meta_line.meta.cwd,
cli_version: meta_line.meta.cli_version,
source: meta_line.meta.source,
history_mode: meta_line.meta.history_mode,
thread_source: meta_line.meta.thread_source,
agent_nickname: meta_line.meta.agent_nickname,
agent_role: meta_line.meta.agent_role,
@@ -457,6 +495,7 @@ mod tests {
use codex_protocol::ThreadId;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_state::ThreadMetadataBuilder;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
@@ -665,6 +704,7 @@ mod tests {
"Forked user message",
Some("test-provider"),
Some(parent_uuid),
ThreadHistoryMode::Legacy,
)
.expect("forked session file");
codex-rs/thread-store/src/local/test_support.rs
+17
View File
@@ -3,6 +3,7 @@ use std::io::Write;
use std::path::Path;
use std::path::PathBuf;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_rollout::ARCHIVED_SESSIONS_SUBDIR;
use uuid::Uuid;
@@ -17,6 +18,15 @@ pub(super) fn test_config(codex_home: &Path) -> LocalThreadStoreConfig {
}
pub(super) fn write_session_file(root: &Path, ts: &str, uuid: Uuid) -> std::io::Result<PathBuf> {
write_session_file_with_history_mode(root, ts, uuid, ThreadHistoryMode::Legacy)
}
pub(super) fn write_session_file_with_history_mode(
root: &Path,
ts: &str,
uuid: Uuid,
history_mode: ThreadHistoryMode,
) -> std::io::Result<PathBuf> {
write_session_file_with(
root,
root.join("sessions/2025/01/03"),
@@ -24,6 +34,7 @@ pub(super) fn write_session_file(root: &Path, ts: &str, uuid: Uuid) -> std::io::
uuid,
"Hello from user",
Some("test-provider"),
history_mode,
)
}
@@ -39,6 +50,7 @@ pub(super) fn write_archived_session_file(
uuid,
"Archived user message",
Some("test-provider"),
ThreadHistoryMode::Legacy,
)
}
@@ -49,6 +61,7 @@ pub(super) fn write_session_file_with(
uuid: Uuid,
first_user_message: &str,
model_provider: Option<&str>,
history_mode: ThreadHistoryMode,
) -> std::io::Result<PathBuf> {
write_session_file_with_fork(
root,
@@ -58,9 +71,11 @@ pub(super) fn write_session_file_with(
first_user_message,
model_provider,
/*forked_from_id*/ None,
history_mode,
)
}
#[allow(clippy::too_many_arguments)]
pub(super) fn write_session_file_with_fork(
root: &Path,
day_dir: PathBuf,
@@ -69,6 +84,7 @@ pub(super) fn write_session_file_with_fork(
first_user_message: &str,
model_provider: Option<&str>,
forked_from_id: Option<Uuid>,
history_mode: ThreadHistoryMode,
) -> std::io::Result<PathBuf> {
fs::create_dir_all(&day_dir)?;
let path = day_dir.join(format!("rollout-{ts}-{uuid}.jsonl"));
@@ -86,6 +102,7 @@ pub(super) fn write_session_file_with_fork(
"cli_version": "test_version",
"source": "cli",
"model_provider": model_provider,
"history_mode": history_mode,
"git": {
"commit_hash": "abcdef",
"branch": "main",
codex-rs/thread-store/src/local/update_thread_metadata.rs
+71
View File
@@ -27,6 +27,7 @@ use crate::ThreadMetadataPatch;
use crate::ThreadStoreError;
use crate::ThreadStoreResult;
use crate::UpdateThreadMetadataParams;
use crate::error::reject_paginated_history_mode;
use crate::local::read_thread;
struct ResolvedRolloutPath {
@@ -53,6 +54,20 @@ pub(super) async fn update_thread_metadata(
}
let needs_rollout_compat = needs_rollout_compatibility_update(&patch);
if needs_rollout_compat {
// These explicit patches still write legacy rollout/name-index state after the
// SQLite update. Paginated threads must fail before either side is mutated.
let thread = read_thread::read_thread(
store,
ReadThreadParams {
thread_id,
include_archived: params.include_archived,
include_history: false,
},
)
.await?;
reject_paginated_history_mode(thread.history_mode)?;
}
let require_sqlite_write = sqlite_write_failure_should_block(&patch);
let updated = apply_metadata_update(
store,
@@ -222,6 +237,7 @@ async fn apply_metadata_update(
patch.source.clone().unwrap_or(SessionSource::Unknown),
);
builder.model_provider = patch.model_provider.clone();
builder.history_mode = patch.history_mode.unwrap_or_default();
builder.thread_source = patch.thread_source.clone().flatten();
builder.agent_nickname = patch.agent_nickname.clone().flatten();
builder.agent_role = patch.agent_role.clone().flatten();
@@ -269,6 +285,9 @@ async fn apply_metadata_update(
if let Some(source) = patch.source {
metadata.source = enum_to_string(&source);
}
if let Some(history_mode) = patch.history_mode {
metadata.history_mode = history_mode;
}
if let Some(thread_source) = patch.thread_source {
metadata.thread_source = thread_source;
}
@@ -412,6 +431,7 @@ fn has_observed_metadata_facts(patch: &ThreadMetadataPatch) -> bool {
|| patch.permission_profile.is_some()
|| patch.token_usage.is_some()
|| patch.first_user_message.is_some()
|| patch.history_mode.is_some()
}
fn enum_to_string<T: serde::Serialize>(value: &T) -> String {
@@ -632,6 +652,7 @@ fn rollout_path_is_archived(store: &LocalThreadStore, path: &Path) -> bool {
#[cfg(test)]
mod tests {
use codex_protocol::models::PermissionProfile;
use codex_protocol::protocol::ThreadHistoryMode;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
@@ -651,6 +672,7 @@ mod tests {
use crate::local::test_support::test_config;
use crate::local::test_support::write_archived_session_file;
use crate::local::test_support::write_session_file;
use crate::local::test_support::write_session_file_with_history_mode;
#[tokio::test]
async fn update_thread_metadata_sets_name_on_active_rollout_and_indexes_name() {
@@ -719,6 +741,55 @@ mod tests {
assert_eq!(memory_mode.as_deref(), Some("disabled"));
}
#[tokio::test]
async fn update_thread_metadata_rejects_paginated_rollout_compatibility_writes() {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let uuid = Uuid::from_u128(303);
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
let path = write_session_file_with_history_mode(
home.path(),
"2025-01-03T14-35-00",
uuid,
ThreadHistoryMode::Paginated,
)
.expect("session file");
let runtime = codex_state::StateRuntime::init(
home.path().to_path_buf(),
config.default_model_provider_id.clone(),
)
.await
.expect("state db should initialize");
let store = LocalThreadStore::new(config, Some(runtime.clone()));
assert!(matches!(
store
.update_thread_metadata(UpdateThreadMetadataParams {
thread_id,
patch: ThreadMetadataPatch {
memory_mode: Some(ThreadMemoryMode::Disabled),
..Default::default()
},
include_archived: false,
})
.await
.expect_err("paginated rollout compatibility write should fail"),
ThreadStoreError::Unsupported {
operation: "paginated_threads"
}
));
assert_eq!(last_rollout_item(path.as_path())["type"], "event_msg");
assert_eq!(
runtime
.get_thread_memory_mode(thread_id)
.await
.expect("thread memory mode should be readable")
.as_deref(),
Some("enabled")
);
}
#[tokio::test]
async fn update_thread_metadata_preserves_memory_mode_when_updating_git_info() {
let home = TempDir::new().expect("temp dir");
codex-rs/thread-store/src/store.rs
+9
View File
@@ -1,4 +1,5 @@
use codex_protocol::ThreadId;
use codex_protocol::protocol::ThreadHistoryMode;
use std::any::Any;
use std::future::Future;
use std::pin::Pin;
@@ -33,6 +34,14 @@ pub trait ThreadStore: Any + Send + Sync {
/// Return this store as [`Any`] for implementation-owned escape hatches.
fn as_any(&self) -> &dyn Any;
/// Returns the history mode to use when history does not carry a persisted mode.
///
/// The default is legacy so existing stores stay compatible. Stores whose durable contract is
/// already paginated should override this instead of relying on core to infer storage behavior.
fn default_history_mode(&self) -> ThreadHistoryMode {
ThreadHistoryMode::Legacy
}
/// Creates a new live thread.
fn create_thread(&self, params: CreateThreadParams) -> ThreadStoreFuture<'_, ()>;
codex-rs/thread-store/src/thread_metadata_sync.rs
+3
View File
@@ -76,6 +76,7 @@ impl ThreadMetadataSync {
cli_version: Some(env!("CARGO_PKG_VERSION").to_string()),
git_info: git_info.map(git_info_patch_from_observation),
memory_mode: Some(params.metadata.memory_mode),
history_mode: Some(params.history_mode),
..Default::default()
};
Self {
@@ -237,6 +238,7 @@ impl ThreadMetadataSync {
{
update.memory_mode = Some(memory_mode);
}
update.history_mode = Some(meta_line.meta.history_mode);
}
RolloutItem::TurnContext(turn_ctx) => {
if !self.cwd_seen {
@@ -375,6 +377,7 @@ fn update_has_metadata_facts(update: &ThreadMetadataPatch) -> bool {
|| update.first_user_message.is_some()
|| update.git_info.is_some()
|| update.memory_mode.is_some()
|| update.history_mode.is_some()
}
fn git_info_patch_from_observation(git_info: GitInfo) -> GitInfoPatch {
codex-rs/thread-store/src/types.rs
+46
View File
@@ -15,6 +15,7 @@ use codex_protocol::protocol::GitInfo;
use codex_protocol::protocol::MultiAgentVersion;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadMemoryMode as MemoryMode;
use codex_protocol::protocol::ThreadSource;
use codex_protocol::protocol::TokenUsage;
@@ -91,6 +92,8 @@ pub struct CreateThreadParams {
pub selected_capability_roots: Vec<SelectedCapabilityRoot>,
/// Multi-agent runtime selected when the thread was created.
pub multi_agent_version: Option<MultiAgentVersion>,
/// Persisted thread history contract selected when the thread was created.
pub history_mode: ThreadHistoryMode,
/// Initial context-window identity captured when the thread was created.
pub initial_window_id: String,
/// Metadata captured for the newly created thread.
@@ -112,6 +115,20 @@ pub struct ResumeThreadParams {
pub metadata: ThreadPersistenceMetadata,
}
pub(crate) fn canonical_history_mode_from_rollout_items(
items: &[RolloutItem],
) -> ThreadHistoryMode {
// Forked rollouts keep copied source SessionMeta items after the new thread's
// canonical SessionMeta, so the thread contract comes from the first one.
items
.iter()
.find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.history_mode),
_ => None,
})
.unwrap_or_default()
}
/// Parameters for appending rollout items to a live thread.
#[derive(Clone, Debug)]
pub struct AppendThreadItemsParams {
@@ -430,6 +447,8 @@ pub struct StoredThread {
pub cli_version: String,
/// Runtime source for the thread.
pub source: SessionSource,
/// Persisted thread history contract selected when this thread was created.
pub history_mode: ThreadHistoryMode,
/// Optional analytics source classification for this thread.
pub thread_source: Option<ThreadSource>,
/// Optional random nickname for thread-spawn sub-agents.
@@ -577,6 +596,8 @@ pub struct ThreadMetadataPatch {
pub git_info: Option<GitInfoPatch>,
/// Thread memory behavior.
pub memory_mode: Option<MemoryMode>,
/// Persisted thread history contract.
pub history_mode: Option<ThreadHistoryMode>,
}
impl ThreadMetadataPatch {
@@ -657,6 +678,9 @@ impl ThreadMetadataPatch {
if next.memory_mode.is_some() {
self.memory_mode = next.memory_mode;
}
if next.history_mode.is_some() {
self.history_mode = next.history_mode;
}
}
pub fn is_empty(&self) -> bool {
@@ -683,6 +707,7 @@ impl ThreadMetadataPatch {
&& self.first_user_message.is_none()
&& self.git_info.is_none()
&& self.memory_mode.is_none()
&& self.history_mode.is_none()
}
}
@@ -785,6 +810,17 @@ mod tests {
assert!(decoded.is_empty());
}
#[test]
fn canonical_history_mode_uses_first_session_meta() {
assert_eq!(
canonical_history_mode_from_rollout_items(&[
session_meta(ThreadHistoryMode::Legacy),
session_meta(ThreadHistoryMode::Paginated),
]),
ThreadHistoryMode::Legacy
);
}
#[test]
fn thread_metadata_patch_merge_uses_presence_semantics() {
let mut current = ThreadMetadataPatch {
@@ -822,4 +858,14 @@ mod tests {
})
);
}
fn session_meta(history_mode: ThreadHistoryMode) -> RolloutItem {
RolloutItem::SessionMeta(codex_protocol::protocol::SessionMetaLine {
meta: codex_protocol::protocol::SessionMeta {
history_mode,
..Default::default()
},
git: None,
})
}
}