mirror of
https://github.com/openai/codex.git
synced 2026-09-04 15:08:45 +00:00
5267e805fb
## 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.
1367 lines
53 KiB
Rust
1367 lines
53 KiB
Rust
use chrono::DateTime;
|
|
use chrono::Utc;
|
|
use codex_protocol::models::PermissionProfile;
|
|
use codex_protocol::protocol::AskForApproval;
|
|
use codex_protocol::protocol::SessionMetaLine;
|
|
use codex_protocol::protocol::SessionSource;
|
|
use codex_rollout::RolloutRecorder;
|
|
use codex_rollout::find_archived_thread_path_by_id_str;
|
|
use codex_rollout::find_thread_name_by_id;
|
|
use codex_rollout::find_thread_path_by_id_str;
|
|
use codex_rollout::read_session_meta_line;
|
|
use codex_rollout::read_thread_item_from_rollout;
|
|
use codex_state::ThreadMetadata;
|
|
|
|
use super::LocalThreadStore;
|
|
use super::helpers::distinct_thread_metadata_title;
|
|
use super::helpers::git_info_from_parts;
|
|
use super::helpers::permission_profile_from_metadata_value;
|
|
use super::helpers::rollout_path_is_archived;
|
|
use super::helpers::set_thread_name_from_title;
|
|
use super::helpers::stored_thread_from_rollout_item;
|
|
use super::live_writer;
|
|
use crate::ReadThreadParams;
|
|
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,
|
|
params: ReadThreadParams,
|
|
) -> ThreadStoreResult<StoredThread> {
|
|
let thread_id = params.thread_id;
|
|
if let Some(metadata) = read_sqlite_metadata(store, thread_id).await
|
|
&& (params.include_archived
|
|
|| (metadata.archived_at.is_none()
|
|
&& !rollout_path_is_archived(
|
|
store.config.codex_home.as_path(),
|
|
metadata.rollout_path.as_path(),
|
|
)))
|
|
&& (!params.include_history
|
|
|| sqlite_rollout_path_can_load_history_for_thread(
|
|
store,
|
|
&metadata.rollout_path,
|
|
thread_id,
|
|
)
|
|
.await)
|
|
{
|
|
let metadata_sandbox_policy = metadata.sandbox_policy.clone();
|
|
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
|
|
&& rollout_thread.thread_id == thread_id
|
|
&& (params.include_archived || rollout_thread.archived_at.is_none())
|
|
&& !rollout_thread.preview.is_empty()
|
|
{
|
|
rollout_thread.recency_at = thread.recency_at;
|
|
if thread.name.is_some() {
|
|
rollout_thread.name = thread.name;
|
|
}
|
|
rollout_thread.git_info = thread.git_info;
|
|
rollout_thread.permission_profile = permission_profile_from_metadata_value(
|
|
&metadata_sandbox_policy,
|
|
rollout_thread.cwd.as_path(),
|
|
);
|
|
thread = rollout_thread;
|
|
}
|
|
reject_paginated_history(&thread, params.include_history)?;
|
|
attach_history_if_requested(&mut thread, params.include_history).await?;
|
|
return Ok(thread);
|
|
}
|
|
|
|
let path = resolve_rollout_path(store, thread_id, params.include_archived)
|
|
.await?
|
|
.ok_or_else(|| ThreadStoreError::InvalidRequest {
|
|
message: format!("no rollout found for thread id {thread_id}"),
|
|
})?;
|
|
|
|
let mut thread = read_thread_from_rollout_path(store, path).await?;
|
|
if !params.include_archived && thread.archived_at.is_some() {
|
|
return Err(ThreadStoreError::InvalidRequest {
|
|
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)
|
|
}
|
|
|
|
async fn sqlite_rollout_path_can_load_history_for_thread(
|
|
store: &LocalThreadStore,
|
|
path: &std::path::Path,
|
|
thread_id: codex_protocol::ThreadId,
|
|
) -> bool {
|
|
if codex_rollout::existing_rollout_path(path).await.is_none() {
|
|
return false;
|
|
}
|
|
// SQLite metadata can outlive a moved/recreated rollout path. When history is
|
|
// requested, verify the path still resolves to the requested thread before
|
|
// trusting it as the source replay.
|
|
read_thread_from_rollout_path(store, path.to_path_buf())
|
|
.await
|
|
.is_ok_and(|thread| thread.thread_id == thread_id)
|
|
}
|
|
|
|
pub(super) async fn read_thread_by_rollout_path(
|
|
store: &LocalThreadStore,
|
|
rollout_path: std::path::PathBuf,
|
|
include_archived: bool,
|
|
include_history: bool,
|
|
) -> ThreadStoreResult<StoredThread> {
|
|
let path = resolve_requested_rollout_path(store, rollout_path).await?;
|
|
let mut thread = read_thread_from_rollout_path(store, path).await?;
|
|
if !include_archived && thread.archived_at.is_some() {
|
|
return Err(ThreadStoreError::InvalidRequest {
|
|
message: format!("thread {} is archived", thread.thread_id),
|
|
});
|
|
}
|
|
if let Some(metadata) = read_sqlite_metadata(store, thread.thread_id).await {
|
|
thread.recency_at = metadata.recency_at;
|
|
let existing_git_info = thread.git_info.take();
|
|
let (fallback_sha, fallback_branch, fallback_origin_url) = match existing_git_info {
|
|
Some(info) => (
|
|
info.commit_hash.map(|sha| sha.0),
|
|
info.branch,
|
|
info.repository_url,
|
|
),
|
|
None => (None, None, None),
|
|
};
|
|
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)?;
|
|
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,
|
|
) -> ThreadStoreResult<std::path::PathBuf> {
|
|
let path = if rollout_path.is_relative() {
|
|
store.config.codex_home.join(rollout_path)
|
|
} else {
|
|
rollout_path
|
|
};
|
|
match tokio::fs::metadata(path.as_path()).await {
|
|
Ok(metadata) if metadata.is_dir() => {
|
|
return Err(ThreadStoreError::InvalidRequest {
|
|
message: format!(
|
|
"failed to resolve rollout path `{}`: path is a directory",
|
|
path.display()
|
|
),
|
|
});
|
|
}
|
|
Ok(metadata) if !metadata.is_file() => {
|
|
return Err(ThreadStoreError::InvalidRequest {
|
|
message: format!(
|
|
"failed to resolve rollout path `{}`: path is not a file",
|
|
path.display()
|
|
),
|
|
});
|
|
}
|
|
_ => {}
|
|
}
|
|
let Some(path) = codex_rollout::existing_rollout_path(path.as_path()).await else {
|
|
return Err(ThreadStoreError::InvalidRequest {
|
|
message: format!(
|
|
"failed to resolve rollout path `{}`: file does not exist",
|
|
path.display()
|
|
),
|
|
});
|
|
};
|
|
std::fs::canonicalize(path.as_path()).map_err(|err| ThreadStoreError::InvalidRequest {
|
|
message: format!("failed to resolve rollout path `{}`: {err}", path.display()),
|
|
})
|
|
}
|
|
|
|
async fn attach_history_if_requested(
|
|
thread: &mut StoredThread,
|
|
include_history: bool,
|
|
) -> ThreadStoreResult<()> {
|
|
if !include_history {
|
|
return Ok(());
|
|
}
|
|
let thread_id = thread.thread_id;
|
|
let Some(path) = thread.rollout_path.clone() else {
|
|
return Err(ThreadStoreError::Internal {
|
|
message: format!("failed to load thread history for thread {thread_id}"),
|
|
});
|
|
};
|
|
let items = load_history_items(&path).await?;
|
|
thread.history = Some(StoredThreadHistory { thread_id, items });
|
|
Ok(())
|
|
}
|
|
|
|
async fn resolve_rollout_path(
|
|
store: &LocalThreadStore,
|
|
thread_id: codex_protocol::ThreadId,
|
|
include_archived: bool,
|
|
) -> ThreadStoreResult<Option<std::path::PathBuf>> {
|
|
if let Ok(path) = live_writer::rollout_path(store, thread_id).await
|
|
&& codex_rollout::existing_rollout_path(path.as_path())
|
|
.await
|
|
.is_some()
|
|
&& (include_archived || !rollout_path_is_archived(store.config.codex_home.as_path(), &path))
|
|
{
|
|
return Ok(Some(path));
|
|
}
|
|
|
|
let state_db_ctx = store.state_db().await;
|
|
if include_archived {
|
|
match find_thread_path_by_id_str(
|
|
store.config.codex_home.as_path(),
|
|
&thread_id.to_string(),
|
|
state_db_ctx.as_deref(),
|
|
)
|
|
.await
|
|
.map_err(|err| ThreadStoreError::InvalidRequest {
|
|
message: format!("failed to locate thread id {thread_id}: {err}"),
|
|
})? {
|
|
Some(path) => Ok(Some(path)),
|
|
None => find_archived_thread_path_by_id_str(
|
|
store.config.codex_home.as_path(),
|
|
&thread_id.to_string(),
|
|
state_db_ctx.as_deref(),
|
|
)
|
|
.await
|
|
.map_err(|err| ThreadStoreError::InvalidRequest {
|
|
message: format!("failed to locate archived thread id {thread_id}: {err}"),
|
|
}),
|
|
}
|
|
} else {
|
|
find_thread_path_by_id_str(
|
|
store.config.codex_home.as_path(),
|
|
&thread_id.to_string(),
|
|
state_db_ctx.as_deref(),
|
|
)
|
|
.await
|
|
.map_err(|err| ThreadStoreError::InvalidRequest {
|
|
message: format!("failed to locate thread id {thread_id}: {err}"),
|
|
})
|
|
}
|
|
}
|
|
|
|
async fn read_thread_from_rollout_path(
|
|
store: &LocalThreadStore,
|
|
path: std::path::PathBuf,
|
|
) -> ThreadStoreResult<StoredThread> {
|
|
let Some(item) = read_thread_item_from_rollout(path.clone()).await else {
|
|
return stored_thread_from_session_meta(store, path).await;
|
|
};
|
|
let archived = rollout_path_is_archived(store.config.codex_home.as_path(), path.as_path());
|
|
let mut thread = stored_thread_from_rollout_item(
|
|
item,
|
|
archived,
|
|
store.config.default_model_provider_id.as_str(),
|
|
)
|
|
.ok_or_else(|| ThreadStoreError::Internal {
|
|
message: format!("failed to read thread id from {}", path.display()),
|
|
})?;
|
|
thread.rollout_path = Some(codex_rollout::plain_rollout_path(path.as_path()));
|
|
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
|
|
{
|
|
set_thread_name_from_title(&mut thread, title);
|
|
}
|
|
Ok(thread)
|
|
}
|
|
|
|
async fn load_history_items(
|
|
path: &std::path::Path,
|
|
) -> ThreadStoreResult<Vec<codex_protocol::protocol::RolloutItem>> {
|
|
let (items, _, _) = RolloutRecorder::load_rollout_items(path)
|
|
.await
|
|
.map_err(|err| ThreadStoreError::Internal {
|
|
message: format!("failed to load thread history {}: {err}", path.display()),
|
|
})?;
|
|
Ok(items)
|
|
}
|
|
|
|
async fn read_sqlite_metadata(
|
|
store: &LocalThreadStore,
|
|
thread_id: codex_protocol::ThreadId,
|
|
) -> Option<ThreadMetadata> {
|
|
let runtime = store.state_db().await?;
|
|
runtime.get_thread(thread_id).await.ok().flatten()
|
|
}
|
|
|
|
async fn stored_thread_from_sqlite_metadata(
|
|
store: &LocalThreadStore,
|
|
metadata: ThreadMetadata,
|
|
) -> 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)
|
|
.await
|
|
.ok()
|
|
.flatten()
|
|
.filter(|title| !title.trim().is_empty()),
|
|
};
|
|
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()
|
|
.or_else(|| metadata.first_user_message.clone())
|
|
.unwrap_or_default();
|
|
let permission_profile =
|
|
permission_profile_from_metadata_value(&metadata.sandbox_policy, metadata.cwd.as_path());
|
|
Ok(StoredThread {
|
|
thread_id: metadata.id,
|
|
extra_config: None,
|
|
rollout_path: Some(rollout_path),
|
|
forked_from_id,
|
|
parent_thread_id,
|
|
preview,
|
|
name,
|
|
model_provider: if metadata.model_provider.is_empty() {
|
|
store.config.default_model_provider_id.clone()
|
|
} else {
|
|
metadata.model_provider
|
|
},
|
|
model: metadata.model,
|
|
reasoning_effort: metadata.reasoning_effort,
|
|
created_at: metadata.created_at,
|
|
updated_at: metadata.updated_at,
|
|
recency_at: metadata.recency_at,
|
|
archived_at: metadata.archived_at,
|
|
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,
|
|
agent_path: metadata.agent_path,
|
|
git_info: git_info_from_parts(
|
|
metadata.git_sha,
|
|
metadata.git_branch,
|
|
metadata.git_origin_url,
|
|
),
|
|
approval_mode: parse_or_default(&metadata.approval_mode, AskForApproval::OnRequest),
|
|
permission_profile,
|
|
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_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,
|
|
path: std::path::PathBuf,
|
|
archived: bool,
|
|
) -> StoredThread {
|
|
let created_at = parse_rfc3339_non_optional(&meta_line.meta.timestamp).unwrap_or_else(Utc::now);
|
|
let updated_at = std::fs::metadata(path.as_path())
|
|
.ok()
|
|
.and_then(|meta| meta.modified().ok())
|
|
.map(DateTime::<Utc>::from)
|
|
.unwrap_or(created_at);
|
|
let rollout_path = codex_rollout::plain_rollout_path(path.as_path());
|
|
StoredThread {
|
|
thread_id: meta_line.meta.id,
|
|
extra_config: None,
|
|
rollout_path: Some(rollout_path),
|
|
forked_from_id: meta_line.meta.forked_from_id,
|
|
parent_thread_id: meta_line.meta.parent_thread_id,
|
|
preview: String::new(),
|
|
name: None,
|
|
model_provider: meta_line
|
|
.meta
|
|
.model_provider
|
|
.filter(|provider| !provider.is_empty())
|
|
.unwrap_or_else(|| store.config.default_model_provider_id.clone()),
|
|
model: None,
|
|
reasoning_effort: None,
|
|
created_at,
|
|
updated_at,
|
|
recency_at: updated_at,
|
|
archived_at: archived.then_some(updated_at),
|
|
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,
|
|
agent_path: meta_line.meta.agent_path,
|
|
git_info: meta_line.git,
|
|
approval_mode: AskForApproval::OnRequest,
|
|
permission_profile: PermissionProfile::read_only(),
|
|
token_usage: None,
|
|
first_user_message: None,
|
|
history: None,
|
|
}
|
|
}
|
|
|
|
fn parse_session_source(source: &str) -> SessionSource {
|
|
serde_json::from_str(source)
|
|
.or_else(|_| serde_json::from_value(serde_json::Value::String(source.to_string())))
|
|
.unwrap_or(SessionSource::Unknown)
|
|
}
|
|
|
|
fn parse_or_default<T>(value: &str, default: T) -> T
|
|
where
|
|
T: serde::de::DeserializeOwned,
|
|
{
|
|
serde_json::from_str(value)
|
|
.or_else(|_| serde_json::from_value(serde_json::Value::String(value.to_string())))
|
|
.unwrap_or(default)
|
|
}
|
|
|
|
fn parse_rfc3339_non_optional(value: &str) -> Option<DateTime<Utc>> {
|
|
DateTime::parse_from_rfc3339(value)
|
|
.ok()
|
|
.map(|dt| dt.with_timezone(&Utc))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use std::io::Write;
|
|
use std::path::PathBuf;
|
|
|
|
use chrono::Utc;
|
|
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;
|
|
use uuid::Uuid;
|
|
|
|
use super::*;
|
|
use crate::ThreadStore;
|
|
use crate::local::LocalThreadStore;
|
|
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_fork;
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_returns_active_rollout_summary() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
|
|
let uuid = Uuid::from_u128(205);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let active_path =
|
|
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: false,
|
|
include_history: true,
|
|
})
|
|
.await
|
|
.expect("read thread");
|
|
|
|
assert_eq!(thread.thread_id, thread_id);
|
|
assert_eq!(thread.rollout_path, Some(active_path));
|
|
assert_eq!(thread.archived_at, None);
|
|
assert_eq!(thread.preview, "Hello from user");
|
|
assert_eq!(
|
|
thread.history.expect("history should load").thread_id,
|
|
thread_id
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_returns_rollout_path_summary() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
|
|
let uuid = Uuid::from_u128(211);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let active_path =
|
|
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
|
|
let relative_path = active_path
|
|
.strip_prefix(home.path())
|
|
.expect("path should be under codex home")
|
|
.to_path_buf();
|
|
|
|
let thread = store
|
|
.read_thread_by_rollout_path(
|
|
relative_path,
|
|
/*include_archived*/ false,
|
|
/*include_history*/ false,
|
|
)
|
|
.await
|
|
.expect("read thread by rollout path");
|
|
|
|
assert_eq!(thread.thread_id, thread_id);
|
|
assert_eq!(
|
|
thread.rollout_path,
|
|
Some(std::fs::canonicalize(active_path).expect("canonical path"))
|
|
);
|
|
assert_eq!(thread.preview, "Hello from user");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_by_rollout_path_prefers_sqlite_git_info() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let config = test_config(home.path());
|
|
let uuid = Uuid::from_u128(223);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let active_path =
|
|
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
|
|
let runtime = codex_state::StateRuntime::init(
|
|
config.sqlite_home.clone(),
|
|
config.default_model_provider_id.clone(),
|
|
)
|
|
.await
|
|
.expect("state db should initialize");
|
|
let store = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
|
|
let mut builder = ThreadMetadataBuilder::new(
|
|
thread_id,
|
|
active_path.clone(),
|
|
Utc::now(),
|
|
SessionSource::Cli,
|
|
);
|
|
builder.model_provider = Some(config.default_model_provider_id.clone());
|
|
builder.git_branch = Some("sqlite-branch".to_string());
|
|
let recency_at = chrono::DateTime::parse_from_rfc3339("2026-01-03T12:00:00Z")
|
|
.expect("timestamp should parse")
|
|
.with_timezone(&Utc);
|
|
builder.recency_at = Some(recency_at);
|
|
runtime
|
|
.upsert_thread(&builder.build(config.default_model_provider_id.as_str()))
|
|
.await
|
|
.expect("state db upsert should succeed");
|
|
|
|
let thread = store
|
|
.read_thread_by_rollout_path(
|
|
active_path,
|
|
/*include_archived*/ false,
|
|
/*include_history*/ false,
|
|
)
|
|
.await
|
|
.expect("read thread by rollout path");
|
|
|
|
let git_info = thread.git_info.expect("git info should be present");
|
|
assert_eq!(thread.recency_at, recency_at);
|
|
assert_eq!(git_info.branch.as_deref(), Some("sqlite-branch"));
|
|
assert_eq!(
|
|
git_info.commit_hash.as_ref().map(|sha| sha.0.as_str()),
|
|
Some("abcdef")
|
|
);
|
|
assert_eq!(
|
|
git_info.repository_url.as_deref(),
|
|
Some("https://example.com/repo.git")
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_returns_archived_rollout_when_requested() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
|
|
let uuid = Uuid::from_u128(207);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let archived_path = write_archived_session_file(home.path(), "2025-01-03T12-00-00", uuid)
|
|
.expect("archived session file");
|
|
|
|
let active_only_err = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: false,
|
|
include_history: false,
|
|
})
|
|
.await
|
|
.expect_err("active-only read should fail for archived rollout");
|
|
let ThreadStoreError::InvalidRequest { message } = active_only_err else {
|
|
panic!("expected invalid request error");
|
|
};
|
|
assert_eq!(
|
|
message,
|
|
format!("no rollout found for thread id {thread_id}")
|
|
);
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: true,
|
|
include_history: false,
|
|
})
|
|
.await
|
|
.expect("read archived thread");
|
|
|
|
assert_eq!(thread.thread_id, thread_id);
|
|
assert_eq!(thread.rollout_path, Some(archived_path));
|
|
assert!(thread.archived_at.is_some());
|
|
assert_eq!(thread.preview, "Archived user message");
|
|
assert!(thread.history.is_none());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_prefers_active_rollout_over_archived() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
|
|
let uuid = Uuid::from_u128(208);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let active_path =
|
|
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
|
|
write_archived_session_file(home.path(), "2025-01-03T12-00-00", uuid)
|
|
.expect("archived session file");
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: true,
|
|
include_history: false,
|
|
})
|
|
.await
|
|
.expect("read thread");
|
|
|
|
assert_eq!(thread.rollout_path, Some(active_path));
|
|
assert_eq!(thread.archived_at, None);
|
|
assert_eq!(thread.preview, "Hello from user");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_returns_forked_from_id() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
|
|
let uuid = Uuid::from_u128(209);
|
|
let parent_uuid = Uuid::from_u128(210);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let parent_thread_id =
|
|
ThreadId::from_string(&parent_uuid.to_string()).expect("valid parent thread id");
|
|
write_session_file_with_fork(
|
|
home.path(),
|
|
home.path().join("sessions/2025/01/03"),
|
|
"2025-01-03T12-00-00",
|
|
uuid,
|
|
"Forked user message",
|
|
Some("test-provider"),
|
|
Some(parent_uuid),
|
|
ThreadHistoryMode::Legacy,
|
|
)
|
|
.expect("forked session file");
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: false,
|
|
include_history: false,
|
|
})
|
|
.await
|
|
.expect("read thread");
|
|
|
|
assert_eq!(thread.forked_from_id, Some(parent_thread_id));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_applies_sqlite_thread_name() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let config = test_config(home.path());
|
|
let uuid = Uuid::from_u128(212);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let rollout_path =
|
|
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
|
|
let runtime = codex_state::StateRuntime::init(
|
|
config.sqlite_home.clone(),
|
|
config.default_model_provider_id.clone(),
|
|
)
|
|
.await
|
|
.expect("state db should initialize");
|
|
let store = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
|
|
let mut builder =
|
|
ThreadMetadataBuilder::new(thread_id, rollout_path, Utc::now(), SessionSource::Cli);
|
|
builder.model_provider = Some(config.default_model_provider_id.clone());
|
|
builder.cwd = home.path().to_path_buf();
|
|
builder.cli_version = Some("test_version".to_string());
|
|
let mut metadata = builder.build(config.default_model_provider_id.as_str());
|
|
metadata.title = "Saved title".to_string();
|
|
metadata.first_user_message = Some("Hello from user".to_string());
|
|
runtime
|
|
.upsert_thread(&metadata)
|
|
.await
|
|
.expect("state db upsert should succeed");
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: false,
|
|
include_history: false,
|
|
})
|
|
.await
|
|
.expect("read thread");
|
|
|
|
assert_eq!(thread.name, Some("Saved title".to_string()));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_returns_permission_profile_from_sqlite_metadata() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let config = test_config(home.path());
|
|
let uuid = Uuid::from_u128(225);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let rollout_path =
|
|
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
|
|
let runtime = codex_state::StateRuntime::init(
|
|
config.sqlite_home.clone(),
|
|
config.default_model_provider_id.clone(),
|
|
)
|
|
.await
|
|
.expect("state db should initialize");
|
|
let store = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
|
|
let mut builder =
|
|
ThreadMetadataBuilder::new(thread_id, rollout_path, Utc::now(), SessionSource::Cli);
|
|
builder.model_provider = Some(config.default_model_provider_id.clone());
|
|
builder.cwd = home.path().to_path_buf();
|
|
let mut metadata = builder.build(config.default_model_provider_id.as_str());
|
|
let permission_profile: PermissionProfile = PermissionProfile::Disabled;
|
|
metadata.sandbox_policy =
|
|
serde_json::to_string(&permission_profile).expect("serialize profile");
|
|
runtime
|
|
.upsert_thread(&metadata)
|
|
.await
|
|
.expect("state db upsert should succeed");
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: false,
|
|
include_history: false,
|
|
})
|
|
.await
|
|
.expect("read thread");
|
|
|
|
assert_eq!(thread.preview, "Hello from user");
|
|
assert_eq!(thread.permission_profile, PermissionProfile::Disabled);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_accepts_legacy_sandbox_policy_metadata() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let config = test_config(home.path());
|
|
let uuid = Uuid::from_u128(226);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let rollout_path =
|
|
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
|
|
let runtime = codex_state::StateRuntime::init(
|
|
config.sqlite_home.clone(),
|
|
config.default_model_provider_id.clone(),
|
|
)
|
|
.await
|
|
.expect("state db should initialize");
|
|
let store = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
|
|
let mut builder =
|
|
ThreadMetadataBuilder::new(thread_id, rollout_path, Utc::now(), SessionSource::Cli);
|
|
builder.model_provider = Some(config.default_model_provider_id.clone());
|
|
builder.cwd = home.path().to_path_buf();
|
|
let mut metadata = builder.build(config.default_model_provider_id.as_str());
|
|
metadata.sandbox_policy = "danger-full-access".to_string();
|
|
runtime
|
|
.upsert_thread(&metadata)
|
|
.await
|
|
.expect("state db upsert should succeed");
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: false,
|
|
include_history: true,
|
|
})
|
|
.await
|
|
.expect("read thread");
|
|
|
|
assert_eq!(thread.permission_profile, PermissionProfile::Disabled);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_preserves_rollout_cwd_when_sqlite_metadata_exists() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let config = test_config(home.path());
|
|
let runtime = codex_state::StateRuntime::init(
|
|
config.sqlite_home.clone(),
|
|
config.default_model_provider_id.clone(),
|
|
)
|
|
.await
|
|
.expect("state db should initialize");
|
|
let store = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
|
|
let uuid = Uuid::from_u128(224);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let day_dir = home.path().join("sessions/2025/01/03");
|
|
std::fs::create_dir_all(&day_dir).expect("sessions dir");
|
|
let rollout_path = day_dir.join(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl"));
|
|
let mut file = std::fs::File::create(&rollout_path).expect("session file");
|
|
let rollout_cwd = PathBuf::from("/");
|
|
let meta = serde_json::json!({
|
|
"timestamp": "2025-01-03T12:00:00Z",
|
|
"type": "session_meta",
|
|
"payload": {
|
|
"session_id": uuid,
|
|
"id": uuid,
|
|
"timestamp": "2025-01-03T12:00:00Z",
|
|
"cwd": rollout_cwd,
|
|
"originator": "test_originator",
|
|
"cli_version": "test_version",
|
|
"source": "cli",
|
|
"model_provider": "rollout-provider"
|
|
},
|
|
});
|
|
writeln!(file, "{meta}").expect("write session meta");
|
|
let user_event = serde_json::json!({
|
|
"timestamp": "2025-01-03T12:00:00Z",
|
|
"type": "event_msg",
|
|
"payload": {
|
|
"type": "user_message",
|
|
"message": "Hello from rollout",
|
|
"kind": "plain",
|
|
},
|
|
});
|
|
writeln!(file, "{user_event}").expect("write user event");
|
|
|
|
let mut builder = ThreadMetadataBuilder::new(
|
|
thread_id,
|
|
rollout_path.clone(),
|
|
Utc::now(),
|
|
SessionSource::Cli,
|
|
);
|
|
builder.model_provider = Some(config.default_model_provider_id.clone());
|
|
builder.cwd = home.path().join("sqlite-workspace");
|
|
let mut metadata = builder.build(config.default_model_provider_id.as_str());
|
|
metadata.title = "Saved title".to_string();
|
|
metadata.first_user_message = Some("Hello from sqlite".to_string());
|
|
metadata.sandbox_policy = "workspace-write".to_string();
|
|
runtime
|
|
.upsert_thread(&metadata)
|
|
.await
|
|
.expect("state db upsert should succeed");
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: false,
|
|
include_history: false,
|
|
})
|
|
.await
|
|
.expect("read thread");
|
|
|
|
assert_eq!(thread.thread_id, thread_id);
|
|
assert_eq!(thread.rollout_path, Some(rollout_path));
|
|
assert_eq!(thread.preview, "Hello from rollout");
|
|
assert_eq!(thread.name, Some("Saved title".to_string()));
|
|
assert_eq!(thread.model_provider, "rollout-provider");
|
|
assert_eq!(thread.cwd, rollout_cwd);
|
|
let legacy_policy = SandboxPolicy::WorkspaceWrite {
|
|
writable_roots: Vec::new(),
|
|
network_access: false,
|
|
exclude_tmpdir_env_var: false,
|
|
exclude_slash_tmp: false,
|
|
};
|
|
assert_eq!(
|
|
thread.permission_profile,
|
|
PermissionProfile::from_legacy_sandbox_policy_for_cwd(
|
|
&legacy_policy,
|
|
rollout_cwd.as_path()
|
|
)
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_uses_legacy_thread_name_when_sqlite_title_is_missing() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
|
|
let uuid = Uuid::from_u128(213);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
|
|
codex_rollout::append_thread_name(home.path(), thread_id, "Legacy title")
|
|
.await
|
|
.expect("append legacy thread name");
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: false,
|
|
include_history: false,
|
|
})
|
|
.await
|
|
.expect("read thread");
|
|
|
|
assert_eq!(thread.name, Some("Legacy title".to_string()));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_uses_sqlite_metadata_for_rollout_without_user_preview() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let config = test_config(home.path());
|
|
let uuid = Uuid::from_u128(217);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let day_dir = home.path().join("sessions/2025/01/03");
|
|
std::fs::create_dir_all(&day_dir).expect("sessions dir");
|
|
let rollout_path = day_dir.join(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl"));
|
|
let mut file = std::fs::File::create(&rollout_path).expect("session file");
|
|
let meta = serde_json::json!({
|
|
"timestamp": "2025-01-03T12-00-00",
|
|
"type": "session_meta",
|
|
"payload": {
|
|
"session_id": uuid,
|
|
"id": uuid,
|
|
"timestamp": "2025-01-03T12-00-00",
|
|
"cwd": home.path(),
|
|
"originator": "test_originator",
|
|
"cli_version": "test_version",
|
|
"source": "cli",
|
|
"model_provider": "rollout-provider"
|
|
},
|
|
});
|
|
writeln!(file, "{meta}").expect("write session meta");
|
|
|
|
let runtime = codex_state::StateRuntime::init(
|
|
config.sqlite_home.clone(),
|
|
config.default_model_provider_id.clone(),
|
|
)
|
|
.await
|
|
.expect("state db should initialize");
|
|
let store = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
|
|
let mut builder = ThreadMetadataBuilder::new(
|
|
thread_id,
|
|
rollout_path.clone(),
|
|
Utc::now(),
|
|
SessionSource::Cli,
|
|
);
|
|
builder.model_provider = Some("sqlite-provider".to_string());
|
|
builder.cwd = home.path().join("workspace");
|
|
builder.cli_version = Some("sqlite-cli".to_string());
|
|
let mut metadata = builder.build(config.default_model_provider_id.as_str());
|
|
metadata.title = "Command-only thread".to_string();
|
|
runtime
|
|
.upsert_thread(&metadata)
|
|
.await
|
|
.expect("state db upsert should succeed");
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: false,
|
|
include_history: true,
|
|
})
|
|
.await
|
|
.expect("read thread");
|
|
|
|
assert_eq!(thread.thread_id, thread_id);
|
|
assert_eq!(thread.rollout_path, Some(rollout_path));
|
|
assert_eq!(thread.preview, "");
|
|
assert_eq!(thread.name.as_deref(), Some("Command-only thread"));
|
|
assert_eq!(thread.model_provider, "sqlite-provider");
|
|
assert_eq!(thread.cwd, home.path().join("workspace"));
|
|
assert_eq!(thread.cli_version, "sqlite-cli");
|
|
let history = thread.history.expect("history should load");
|
|
assert_eq!(history.thread_id, thread_id);
|
|
assert_eq!(history.items.len(), 1);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_falls_back_to_rollout_search_when_sqlite_path_is_stale() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let external = TempDir::new().expect("external temp dir");
|
|
let config = test_config(home.path());
|
|
let uuid = Uuid::from_u128(220);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let rollout_path =
|
|
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
|
|
let stale_path = external.path().join("missing-rollout.jsonl");
|
|
let runtime = codex_state::StateRuntime::init(
|
|
config.sqlite_home.clone(),
|
|
config.default_model_provider_id.clone(),
|
|
)
|
|
.await
|
|
.expect("state db should initialize");
|
|
let store = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
|
|
let mut builder = ThreadMetadataBuilder::new(
|
|
thread_id,
|
|
stale_path.clone(),
|
|
Utc::now(),
|
|
SessionSource::Cli,
|
|
);
|
|
builder.model_provider = Some("stale-sqlite-provider".to_string());
|
|
let mut metadata = builder.build(config.default_model_provider_id.as_str());
|
|
metadata.first_user_message = Some("stale sqlite preview".to_string());
|
|
runtime
|
|
.upsert_thread(&metadata)
|
|
.await
|
|
.expect("state db upsert should succeed");
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: true,
|
|
include_history: true,
|
|
})
|
|
.await
|
|
.expect("read thread");
|
|
|
|
assert_eq!(thread.thread_id, thread_id);
|
|
assert_eq!(thread.rollout_path, Some(rollout_path));
|
|
assert_eq!(thread.preview, "Hello from user");
|
|
assert_eq!(thread.model_provider, config.default_model_provider_id);
|
|
let history = thread.history.expect("history should load");
|
|
assert_eq!(history.thread_id, thread_id);
|
|
assert_eq!(history.items.len(), 2);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_falls_back_when_sqlite_path_points_to_another_thread() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let external = TempDir::new().expect("external temp dir");
|
|
let config = test_config(home.path());
|
|
let uuid = Uuid::from_u128(221);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let rollout_path =
|
|
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
|
|
let other_uuid = Uuid::from_u128(222);
|
|
let stale_path = write_session_file(external.path(), "2025-01-04T12-00-00", other_uuid)
|
|
.expect("other session file");
|
|
let runtime = codex_state::StateRuntime::init(
|
|
config.sqlite_home.clone(),
|
|
config.default_model_provider_id.clone(),
|
|
)
|
|
.await
|
|
.expect("state db should initialize");
|
|
let store = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
|
|
let mut builder =
|
|
ThreadMetadataBuilder::new(thread_id, stale_path, Utc::now(), SessionSource::Cli);
|
|
builder.model_provider = Some("wrong-sqlite-provider".to_string());
|
|
let mut metadata = builder.build(config.default_model_provider_id.as_str());
|
|
metadata.first_user_message = Some("wrong sqlite preview".to_string());
|
|
runtime
|
|
.upsert_thread(&metadata)
|
|
.await
|
|
.expect("state db upsert should succeed");
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: true,
|
|
include_history: true,
|
|
})
|
|
.await
|
|
.expect("read thread");
|
|
|
|
assert_eq!(thread.thread_id, thread_id);
|
|
assert_eq!(thread.rollout_path, Some(rollout_path));
|
|
assert_eq!(thread.preview, "Hello from user");
|
|
assert_eq!(thread.model_provider, config.default_model_provider_id);
|
|
let history = thread.history.expect("history should load");
|
|
assert_eq!(history.thread_id, thread_id);
|
|
assert_eq!(history.items.len(), 2);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_uses_session_meta_for_rollout_without_user_preview_or_sqlite_metadata() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
|
|
let uuid = Uuid::from_u128(218);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let day_dir = home.path().join("sessions/2025/01/03");
|
|
std::fs::create_dir_all(&day_dir).expect("sessions dir");
|
|
let rollout_path = day_dir.join(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl"));
|
|
let mut file = std::fs::File::create(&rollout_path).expect("session file");
|
|
let meta = serde_json::json!({
|
|
"timestamp": "2025-01-03T12:00:00Z",
|
|
"type": "session_meta",
|
|
"payload": {
|
|
"session_id": uuid,
|
|
"id": uuid,
|
|
"timestamp": "2025-01-03T12:00:00Z",
|
|
"cwd": home.path(),
|
|
"originator": "test_originator",
|
|
"cli_version": "test_version",
|
|
"source": "cli",
|
|
"model_provider": "rollout-provider"
|
|
},
|
|
});
|
|
writeln!(file, "{meta}").expect("write session meta");
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: false,
|
|
include_history: true,
|
|
})
|
|
.await
|
|
.expect("read thread");
|
|
|
|
assert_eq!(thread.thread_id, thread_id);
|
|
assert_eq!(thread.rollout_path, Some(rollout_path));
|
|
assert_eq!(thread.preview, "");
|
|
assert_eq!(thread.name, None);
|
|
assert_eq!(thread.model_provider, "rollout-provider");
|
|
assert_eq!(
|
|
thread.created_at,
|
|
parse_rfc3339_non_optional("2025-01-03T12:00:00Z").unwrap()
|
|
);
|
|
assert!(thread.updated_at >= thread.created_at);
|
|
assert_eq!(thread.archived_at, None);
|
|
assert_eq!(thread.cwd, home.path());
|
|
assert_eq!(thread.cli_version, "test_version");
|
|
assert_eq!(thread.source, SessionSource::Cli);
|
|
let history = thread.history.expect("history should load");
|
|
assert_eq!(history.thread_id, thread_id);
|
|
assert_eq!(history.items.len(), 1);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_falls_back_to_sqlite_summary() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let external = TempDir::new().expect("external temp dir");
|
|
let config = test_config(home.path());
|
|
let uuid = Uuid::from_u128(214);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let rollout_path = external
|
|
.path()
|
|
.join(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl"));
|
|
let runtime = codex_state::StateRuntime::init(
|
|
config.sqlite_home.clone(),
|
|
config.default_model_provider_id.clone(),
|
|
)
|
|
.await
|
|
.expect("state db should initialize");
|
|
let store = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
|
|
let mut builder = ThreadMetadataBuilder::new(
|
|
thread_id,
|
|
rollout_path.clone(),
|
|
Utc::now(),
|
|
SessionSource::Exec,
|
|
);
|
|
builder.model_provider = Some("sqlite-provider".to_string());
|
|
builder.cwd = external.path().join("workspace");
|
|
builder.cli_version = Some("sqlite-cli".to_string());
|
|
let mut metadata = builder.build(config.default_model_provider_id.as_str());
|
|
metadata.preview = Some("optimize the benchmark".to_string());
|
|
metadata.first_user_message = Some("next normal prompt".to_string());
|
|
metadata.title = "next normal prompt".to_string();
|
|
metadata.model = Some("sqlite-model".to_string());
|
|
runtime
|
|
.upsert_thread(&metadata)
|
|
.await
|
|
.expect("state db upsert should succeed");
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: false,
|
|
include_history: false,
|
|
})
|
|
.await
|
|
.expect("read thread");
|
|
|
|
assert_eq!(thread.thread_id, thread_id);
|
|
assert_eq!(thread.rollout_path, Some(rollout_path));
|
|
assert_eq!(thread.preview, "optimize the benchmark");
|
|
assert_eq!(
|
|
thread.first_user_message.as_deref(),
|
|
Some("next normal prompt")
|
|
);
|
|
assert_eq!(thread.name, None);
|
|
assert_eq!(thread.model_provider, "sqlite-provider");
|
|
assert_eq!(thread.model.as_deref(), Some("sqlite-model"));
|
|
assert_eq!(thread.cwd, external.path().join("workspace"));
|
|
assert_eq!(thread.cli_version, "sqlite-cli");
|
|
assert_eq!(thread.source, SessionSource::Exec);
|
|
assert_eq!(thread.archived_at, None);
|
|
assert!(thread.history.is_none());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_sqlite_fallback_respects_include_archived() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let external = TempDir::new().expect("external temp dir");
|
|
let config = test_config(home.path());
|
|
let uuid = Uuid::from_u128(216);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let rollout_path = external
|
|
.path()
|
|
.join(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl"));
|
|
let runtime = codex_state::StateRuntime::init(
|
|
config.sqlite_home.clone(),
|
|
config.default_model_provider_id.clone(),
|
|
)
|
|
.await
|
|
.expect("state db should initialize");
|
|
let mut builder =
|
|
ThreadMetadataBuilder::new(thread_id, rollout_path, Utc::now(), SessionSource::Cli);
|
|
let store = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
|
|
builder.archived_at = Some(Utc::now());
|
|
let mut metadata = builder.build(config.default_model_provider_id.as_str());
|
|
metadata.first_user_message = Some("Archived SQLite preview".to_string());
|
|
runtime
|
|
.upsert_thread(&metadata)
|
|
.await
|
|
.expect("state db upsert should succeed");
|
|
|
|
let active_only_err = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: false,
|
|
include_history: false,
|
|
})
|
|
.await
|
|
.expect_err("active-only read should fail for archived metadata");
|
|
let ThreadStoreError::InvalidRequest { message } = active_only_err else {
|
|
panic!("expected invalid request error");
|
|
};
|
|
assert_eq!(
|
|
message,
|
|
format!("no rollout found for thread id {thread_id}")
|
|
);
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: true,
|
|
include_history: false,
|
|
})
|
|
.await
|
|
.expect("read archived thread");
|
|
|
|
assert_eq!(thread.thread_id, thread_id);
|
|
assert_eq!(thread.preview, "Archived SQLite preview");
|
|
assert!(thread.archived_at.is_some());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_sqlite_fallback_loads_archived_history() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let config = test_config(home.path());
|
|
let uuid = Uuid::from_u128(219);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
let archived_path = write_archived_session_file(home.path(), "2025-01-03T12-00-00", uuid)
|
|
.expect("archived session file");
|
|
let runtime = codex_state::StateRuntime::init(
|
|
config.sqlite_home.clone(),
|
|
config.default_model_provider_id.clone(),
|
|
)
|
|
.await
|
|
.expect("state db should initialize");
|
|
let store = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
|
|
let mut builder = ThreadMetadataBuilder::new(
|
|
thread_id,
|
|
archived_path.clone(),
|
|
Utc::now(),
|
|
SessionSource::Cli,
|
|
);
|
|
builder.archived_at = Some(Utc::now());
|
|
let mut metadata = builder.build(config.default_model_provider_id.as_str());
|
|
metadata.first_user_message = Some("Archived SQLite preview".to_string());
|
|
runtime
|
|
.upsert_thread(&metadata)
|
|
.await
|
|
.expect("state db upsert should succeed");
|
|
|
|
let thread = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: true,
|
|
include_history: true,
|
|
})
|
|
.await
|
|
.expect("read archived thread");
|
|
|
|
assert_eq!(thread.thread_id, thread_id);
|
|
assert_eq!(thread.rollout_path, Some(archived_path));
|
|
assert_eq!(thread.preview, "Archived SQLite preview");
|
|
assert!(thread.archived_at.is_some());
|
|
let history = thread.history.expect("history should load");
|
|
assert_eq!(history.thread_id, thread_id);
|
|
assert_eq!(history.items.len(), 2);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_thread_fails_without_rollout() {
|
|
let home = TempDir::new().expect("temp dir");
|
|
let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
|
|
let uuid = Uuid::from_u128(206);
|
|
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
|
|
|
|
let err = store
|
|
.read_thread(ReadThreadParams {
|
|
thread_id,
|
|
include_archived: false,
|
|
include_history: false,
|
|
})
|
|
.await
|
|
.expect_err("read should fail without rollout");
|
|
|
|
let ThreadStoreError::InvalidRequest { message } = err else {
|
|
panic!("expected invalid request error");
|
|
};
|
|
assert_eq!(
|
|
message,
|
|
format!("no rollout found for thread id {thread_id}")
|
|
);
|
|
}
|
|
}
|