openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-09 20:50:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: unify conversation with thread name (#8830)

Browse Source 
Done and verified by Codex + refactor feature of RustRover
This commit is contained in:
jif-oai
2026-01-09 11:37:07 +00:00
committed by pap
parent 137b26bd46
commit ef39b121f2
83 changed files with 1094 additions and 1203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control.rs
+29 -30
View File
@@ -1,9 +1,9 @@
use crate::CodexConversation;
use crate::CodexThread;
use crate::agent::AgentStatus;
use crate::conversation_manager::ConversationManagerState;
use crate::error::CodexErr;
use crate::error::Result as CodexResult;
use codex_protocol::ConversationId;
use crate::thread_manager::ThreadManagerState;
use codex_protocol::ThreadId;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::user_input::UserInput;
@@ -15,47 +15,46 @@ use std::sync::Weak;
/// spawn new agents and the inter-agent communication layer.
#[derive(Clone, Default)]
pub(crate) struct AgentControl {
/// Weak handle back to the global conversation registry/state.
/// Weak handle back to the global thread registry/state.
/// This is `Weak` to avoid reference cycles and shadow persistence of the form
/// `ConversationManagerState -> CodexConversation -> Session -> SessionServices -> ConversationManagerState`.
manager: Weak<ConversationManagerState>,
/// `ThreadManagerState -> CodexThread -> Session -> SessionServices -> ThreadManagerState`.
manager: Weak<ThreadManagerState>,
}
impl AgentControl {
/// Construct a new `AgentControl` that can spawn/message agents via the given manager state.
pub(crate) fn new(manager: Weak<ConversationManagerState>) -> Self {
pub(crate) fn new(manager: Weak<ThreadManagerState>) -> Self {
Self { manager }
}
#[allow(dead_code)] // Used by upcoming multi-agent tooling.
/// Spawn a new agent conversation and submit the initial prompt.
/// Spawn a new agent thread and submit the initial prompt.
///
/// If `headless` is true, a background drain task is spawned to prevent unbounded event growth
/// of the channel queue when there is no client actively reading the conversation events.
/// of the channel queue when there is no client actively reading the thread events.
pub(crate) async fn spawn_agent(
&self,
config: crate::config::Config,
prompt: String,
headless: bool,
) -> CodexResult<ConversationId> {
) -> CodexResult<ThreadId> {
let state = self.upgrade()?;
let new_conversation = state.spawn_new_conversation(config, self.clone()).await?;
let new_thread = state.spawn_new_thread(config, self.clone()).await?;
if headless {
spawn_headless_drain(Arc::clone(&new_conversation.conversation));
spawn_headless_drain(Arc::clone(&new_thread.thread));
}
self.send_prompt(new_conversation.conversation_id, prompt)
.await?;
self.send_prompt(new_thread.thread_id, prompt).await?;
Ok(new_conversation.conversation_id)
Ok(new_thread.thread_id)
}
#[allow(dead_code)] // Used by upcoming multi-agent tooling.
/// Send a `user` prompt to an existing agent conversation.
/// Send a `user` prompt to an existing agent thread.
pub(crate) async fn send_prompt(
&self,
agent_id: ConversationId,
agent_id: ThreadId,
prompt: String,
) -> CodexResult<String> {
let state = self.upgrade()?;
@@ -72,32 +71,32 @@ impl AgentControl {
#[allow(dead_code)] // Used by upcoming multi-agent tooling.
/// Fetch the last known status for `agent_id`, returning `NotFound` when unavailable.
pub(crate) async fn get_status(&self, agent_id: ConversationId) -> AgentStatus {
pub(crate) async fn get_status(&self, agent_id: ThreadId) -> AgentStatus {
let Ok(state) = self.upgrade() else {
// No agent available if upgrade fails.
return AgentStatus::NotFound;
};
let Ok(conversation) = state.get_conversation(agent_id).await else {
let Ok(thread) = state.get_thread(agent_id).await else {
return AgentStatus::NotFound;
};
conversation.agent_status().await
thread.agent_status().await
}
fn upgrade(&self) -> CodexResult<Arc<ConversationManagerState>> {
self.manager.upgrade().ok_or_else(|| {
CodexErr::UnsupportedOperation("conversation manager dropped".to_string())
})
fn upgrade(&self) -> CodexResult<Arc<ThreadManagerState>> {
self.manager
.upgrade()
.ok_or_else(|| CodexErr::UnsupportedOperation("thread manager dropped".to_string()))
}
}
/// When an agent is spawned "headless" (no UI/view attached), there may be no consumer polling
/// `CodexConversation::next_event()`. The underlying event channel is unbounded, so the producer can
/// `CodexThread::next_event()`. The underlying event channel is unbounded, so the producer can
/// accumulate events indefinitely. This drain task prevents that memory growth by polling and
/// discarding events until shutdown.
fn spawn_headless_drain(conversation: Arc<CodexConversation>) {
fn spawn_headless_drain(thread: Arc<CodexThread>) {
tokio::spawn(async move {
loop {
match conversation.next_event().await {
match thread.next_event().await {
Ok(event) => {
if matches!(event.msg, EventMsg::ShutdownComplete) {
break;
@@ -127,19 +126,19 @@ mod tests {
async fn send_prompt_errors_when_manager_dropped() {
let control = AgentControl::default();
let err = control
.send_prompt(ConversationId::new(), "hello".to_string())
.send_prompt(ThreadId::new(), "hello".to_string())
.await
.expect_err("send_prompt should fail without a manager");
assert_eq!(
err.to_string(),
"unsupported operation: conversation manager dropped"
"unsupported operation: thread manager dropped"
);
}
#[tokio::test]
async fn get_status_returns_not_found_without_manager() {
let control = AgentControl::default();
let got = control.get_status(ConversationId::new()).await;
let got = control.get_status(ThreadId::new()).await;
assert_eq!(got, AgentStatus::NotFound);
}
codex-rs/core/src/client.rs
+3 -3
View File
@@ -19,7 +19,7 @@ use codex_api::create_text_param_for_request;
use codex_api::error::ApiError;
use codex_app_server_protocol::AuthMode;
use codex_otel::otel_manager::OtelManager;
use codex_protocol::ConversationId;
use codex_protocol::ThreadId;
use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig;
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;
@@ -60,7 +60,7 @@ pub struct ModelClient {
model_family: ModelFamily,
otel_manager: OtelManager,
provider: ModelProviderInfo,
conversation_id: ConversationId,
conversation_id: ThreadId,
effort: Option<ReasoningEffortConfig>,
summary: ReasoningSummaryConfig,
session_source: SessionSource,
@@ -76,7 +76,7 @@ impl ModelClient {
provider: ModelProviderInfo,
effort: Option<ReasoningEffortConfig>,
summary: ReasoningSummaryConfig,
conversation_id: ConversationId,
conversation_id: ThreadId,
session_source: SessionSource,
) -> Self {
Self {
codex-rs/core/src/codex.rs
+17 -13
View File
@@ -32,7 +32,7 @@ use crate::user_notification::UserNotifier;
use crate::util::error_or_panic;
use async_channel::Receiver;
use async_channel::Sender;
use codex_protocol::ConversationId;
use codex_protocol::ThreadId;
use codex_protocol::approvals::ExecPolicyAmendment;
use codex_protocol::items::TurnItem;
use codex_protocol::protocol::FileChange;
@@ -179,7 +179,9 @@ pub struct Codex {
/// unique session id.
pub struct CodexSpawnOk {
pub codex: Codex,
pub conversation_id: ConversationId,
pub thread_id: ThreadId,
#[deprecated(note = "use thread_id")]
pub conversation_id: ThreadId,
}
pub(crate) const INITIAL_SUBMIT_ID: &str = "";
@@ -299,7 +301,7 @@ impl Codex {
error!("Failed to create session: {e:#}");
map_session_init_error(&e, &config.codex_home)
})?;
let conversation_id = session.conversation_id;
let thread_id = session.conversation_id;
// This task will run until Op::Shutdown is received.
tokio::spawn(submission_loop(session, config, rx_sub));
@@ -310,9 +312,11 @@ impl Codex {
agent_status,
};
#[allow(deprecated)]
Ok(CodexSpawnOk {
codex,
conversation_id,
thread_id,
conversation_id: thread_id,
})
}
@@ -356,7 +360,7 @@ impl Codex {
///
/// A session has at most 1 running task at a time, and can be interrupted by user input.
pub(crate) struct Session {
conversation_id: ConversationId,
conversation_id: ThreadId,
tx_event: Sender<Event>,
agent_status: Arc<RwLock<AgentStatus>>,
state: Mutex<SessionState>,
@@ -368,7 +372,7 @@ pub(crate) struct Session {
next_internal_sub_id: AtomicU64,
}
/// The context needed for a single turn of the conversation.
/// The context needed for a single turn of the thread.
#[derive(Debug)]
pub(crate) struct TurnContext {
pub(crate) sub_id: String,
@@ -505,7 +509,7 @@ impl Session {
session_configuration: &SessionConfiguration,
per_turn_config: Config,
model_family: ModelFamily,
conversation_id: ConversationId,
conversation_id: ThreadId,
sub_id: String,
) -> TurnContext {
let otel_manager = otel_manager.clone().with_model(
@@ -581,7 +585,7 @@ impl Session {
let (conversation_id, rollout_params) = match &initial_history {
InitialHistory::New | InitialHistory::Forked(_) => {
let conversation_id = ConversationId::default();
let conversation_id = ThreadId::default();
(
conversation_id,
RolloutRecorderParams::new(
@@ -2958,7 +2962,7 @@ mod tests {
session
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ConversationId::default(),
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
}))
@@ -3035,7 +3039,7 @@ mod tests {
session
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ConversationId::default(),
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
}))
@@ -3471,7 +3475,7 @@ mod tests {
}
fn otel_manager(
conversation_id: ConversationId,
conversation_id: ThreadId,
config: &Config,
model_family: &ModelFamily,
session_source: SessionSource,
@@ -3494,7 +3498,7 @@ mod tests {
let codex_home = tempfile::tempdir().expect("create temp dir");
let config = build_test_config(codex_home.path()).await;
let config = Arc::new(config);
let conversation_id = ConversationId::default();
let conversation_id = ThreadId::default();
let auth_manager =
AuthManager::from_auth_for_testing(CodexAuth::from_api_key("Test API Key"));
let models_manager = Arc::new(ModelsManager::new(auth_manager.clone()));
@@ -3585,7 +3589,7 @@ mod tests {
let codex_home = tempfile::tempdir().expect("create temp dir");
let config = build_test_config(codex_home.path()).await;
let config = Arc::new(config);
let conversation_id = ConversationId::default();
let conversation_id = ThreadId::default();
let auth_manager =
AuthManager::from_auth_for_testing(CodexAuth::from_api_key("Test API Key"));
let models_manager = Arc::new(ModelsManager::new(auth_manager.clone()));
codex-rs/core/src/codex_delegate.rs
+4 -4
View File
@@ -28,12 +28,12 @@ use crate::error::CodexErr;
use crate::models_manager::manager::ModelsManager;
use codex_protocol::protocol::InitialHistory;
/// Start an interactive sub-Codex conversation and return IO channels.
/// Start an interactive sub-Codex thread and return IO channels.
///
/// The returned `events_rx` yields non-approval events emitted by the sub-agent.
/// Approval requests are handled via `parent_session` and are not surfaced.
/// The returned `ops_tx` allows the caller to submit additional `Op`s to the sub-agent.
pub(crate) async fn run_codex_conversation_interactive(
pub(crate) async fn run_codex_thread_interactive(
config: Config,
auth_manager: Arc<AuthManager>,
models_manager: Arc<ModelsManager>,
@@ -95,7 +95,7 @@ pub(crate) async fn run_codex_conversation_interactive(
///
/// Internally calls the interactive variant, then immediately submits the provided input.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn run_codex_conversation_one_shot(
pub(crate) async fn run_codex_thread_one_shot(
config: Config,
auth_manager: Arc<AuthManager>,
models_manager: Arc<ModelsManager>,
@@ -108,7 +108,7 @@ pub(crate) async fn run_codex_conversation_one_shot(
// Use a child token so we can stop the delegate after completion without
// requiring the caller to cancel the parent token.
let child_cancel = cancel_token.child_token();
let io = run_codex_conversation_interactive(
let io = run_codex_thread_interactive(
config,
auth_manager,
models_manager,
codex-rs/core/src/codex_conversation.rs → codex-rs/core/src/codex_thread.rs
+4 -4
View File
@@ -6,14 +6,14 @@ use crate::protocol::Op;
use crate::protocol::Submission;
use std::path::PathBuf;
pub struct CodexConversation {
pub struct CodexThread {
codex: Codex,
rollout_path: PathBuf,
}
/// Conduit for the bidirectional stream of messages that compose a conversation
/// in Codex.
impl CodexConversation {
/// Conduit for the bidirectional stream of messages that compose a thread
/// (formerly called a conversation) in Codex.
impl CodexThread {
pub(crate) fn new(codex: Codex, rollout_path: PathBuf) -> Self {
Self {
codex,
codex-rs/core/src/compact.rs
+2 -2
View File
@@ -108,7 +108,7 @@ async fn run_compact_task_inner(
sess.notify_background_event(
turn_context.as_ref(),
format!(
"Trimmed {truncated_count} older conversation item(s) before compacting so the prompt fits the model context window."
"Trimmed {truncated_count} older thread item(s) before compacting so the prompt fits the model context window."
),
)
.await;
@@ -182,7 +182,7 @@ async fn run_compact_task_inner(
sess.send_event(&turn_context, event).await;
let warning = EventMsg::Warning(WarningEvent {
message: "Heads up: Long conversations and multiple compactions can cause the model to be less accurate. Start a new conversation when possible to keep conversations small and targeted.".to_string(),
message: "Heads up: Long threads and multiple compactions can cause the model to be less accurate. Start a new thread when possible to keep threads small and targeted.".to_string(),
});
sess.send_event(&turn_context, warning).await;
}
codex-rs/core/src/context_manager/history.rs
+1 -1
View File
@@ -16,7 +16,7 @@ use codex_protocol::protocol::TokenUsage;
use codex_protocol::protocol::TokenUsageInfo;
use std::ops::Deref;
/// Transcript of conversation history
/// Transcript of thread history
#[derive(Debug, Clone, Default)]
pub(crate) struct ContextManager {
/// The oldest items are at the beginning of the vector.
codex-rs/core/src/error.rs
+5 -5
View File
@@ -8,7 +8,7 @@ use chrono::Datelike;
use chrono::Local;
use chrono::Utc;
use codex_async_utils::CancelErr;
use codex_protocol::ConversationId;
use codex_protocol::ThreadId;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::ErrorEvent;
use codex_protocol::protocol::RateLimitSnapshot;
@@ -71,12 +71,12 @@ pub enum CodexErr {
Stream(String, Option<Duration>),
#[error(
"Codex ran out of room in the model's context window. Start a new conversation or clear earlier history before retrying."
"Codex ran out of room in the model's context window. Start a new thread or clear earlier history before retrying."
)]
ContextWindowExceeded,
#[error("no conversation with id: {0}")]
ConversationNotFound(ConversationId),
#[error("no thread with id: {0}")]
ThreadNotFound(ThreadId),
#[error("session configured event was not the first event in the stream")]
SessionConfiguredNotFirstEvent,
@@ -455,7 +455,7 @@ impl CodexErr {
CodexErr::SessionConfiguredNotFirstEvent
| CodexErr::InternalServerError
| CodexErr::InternalAgentDied => CodexErrorInfo::InternalServerError,
CodexErr::UnsupportedOperation(_) | CodexErr::ConversationNotFound(_) => {
CodexErr::UnsupportedOperation(_) | CodexErr::ThreadNotFound(_) => {
CodexErrorInfo::BadRequest
}
CodexErr::Sandbox(_) => CodexErrorInfo::SandboxError,
codex-rs/core/src/lib.rs
+15 -7
View File
@@ -12,9 +12,9 @@ pub mod bash;
mod client;
mod client_common;
pub mod codex;
mod codex_conversation;
mod codex_thread;
mod compact_remote;
pub use codex_conversation::CodexConversation;
pub use codex_thread::CodexThread;
mod agent;
mod codex_delegate;
mod command_safety;
@@ -60,13 +60,19 @@ pub use model_provider_info::OLLAMA_OSS_PROVIDER_ID;
pub use model_provider_info::WireApi;
pub use model_provider_info::built_in_model_providers;
pub use model_provider_info::create_oss_provider_with_base_url;
mod conversation_manager;
mod event_mapping;
pub mod review_format;
pub mod review_prompts;
mod thread_manager;
pub use codex_protocol::protocol::InitialHistory;
pub use conversation_manager::ConversationManager;
pub use conversation_manager::NewConversation;
pub use thread_manager::NewThread;
pub use thread_manager::ThreadManager;
#[deprecated(note = "use ThreadManager")]
pub type ConversationManager = ThreadManager;
#[deprecated(note = "use NewThread")]
pub type NewConversation = NewThread;
#[deprecated(note = "use CodexThread")]
pub type CodexConversation = CodexThread;
// Re-export common auth types for workspace consumers
pub use auth::AuthManager;
pub use auth::CodexAuth;
@@ -87,10 +93,12 @@ pub use rollout::INTERACTIVE_SESSION_SOURCES;
pub use rollout::RolloutRecorder;
pub use rollout::SESSIONS_SUBDIR;
pub use rollout::SessionMeta;
#[deprecated(note = "use find_thread_path_by_id_str")]
pub use rollout::find_conversation_path_by_id_str;
pub use rollout::list::ConversationItem;
pub use rollout::list::ConversationsPage;
pub use rollout::find_thread_path_by_id_str;
pub use rollout::list::Cursor;
pub use rollout::list::ThreadItem;
pub use rollout::list::ThreadsPage;
pub use rollout::list::parse_cursor;
pub use rollout::list::read_head_for_summary;
mod function_tool;
codex-rs/core/src/message_history.rs
+7 -5
View File
@@ -13,6 +13,8 @@
//! trailing `\n`) and write it with a **single `write(2)` system call** while
//! the file descriptor is opened with the `O_APPEND` flag. POSIX guarantees
//! that writes up to `PIPE_BUF` bytes are atomic in that case.
//! Note: `conversation_id` stores the thread id; the field name is preserved for
//! backwards compatibility with existing history files.
use std::fs::File;
use std::fs::OpenOptions;
@@ -36,7 +38,7 @@ use tokio::io::AsyncReadExt;
use crate::config::Config;
use crate::config::types::HistoryPersistence;
use codex_protocol::ConversationId;
use codex_protocol::ThreadId;
#[cfg(unix)]
use std::os::unix::fs::OpenOptionsExt;
#[cfg(unix)]
@@ -69,7 +71,7 @@ fn history_filepath(config: &Config) -> PathBuf {
/// which entails a small amount of blocking I/O internally.
pub(crate) async fn append_entry(
text: &str,
conversation_id: &ConversationId,
conversation_id: &ThreadId,
config: &Config,
) -> Result<()> {
match config.history.persistence {
@@ -402,7 +404,7 @@ fn history_log_id(_metadata: &std::fs::Metadata) -> Option<u64> {
mod tests {
use super::*;
use crate::config::ConfigBuilder;
use codex_protocol::ConversationId;
use codex_protocol::ThreadId;
use pretty_assertions::assert_eq;
use std::fs::File;
use std::io::Write;
@@ -497,7 +499,7 @@ mod tests {
.await
.expect("load config");
let conversation_id = ConversationId::new();
let conversation_id = ThreadId::new();
let entry_one = "a".repeat(200);
let entry_two = "b".repeat(200);
@@ -544,7 +546,7 @@ mod tests {
.await
.expect("load config");
let conversation_id = ConversationId::new();
let conversation_id = ThreadId::new();
let short_entry = "a".repeat(200);
let long_entry = "b".repeat(400);
codex-rs/core/src/rollout/error.rs
+1 -1
View File
@@ -33,7 +33,7 @@ fn map_rollout_io_error(io_err: &std::io::Error, codex_home: &Path) -> Option<Co
sessions_dir.display()
),
ErrorKind::InvalidData | ErrorKind::InvalidInput => format!(
"Session data under {} looks corrupt or unreadable. Clearing the sessions directory may help (this will remove saved conversations).",
"Session data under {} looks corrupt or unreadable. Clearing the sessions directory may help (this will remove saved threads).",
sessions_dir.display()
),
ErrorKind::IsADirectory | ErrorKind::NotADirectory => format!(
codex-rs/core/src/rollout/list.rs
+25 -18
View File
@@ -20,11 +20,11 @@ use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::SessionSource;
/// Returned page of conversation summaries.
/// Returned page of thread (thread) summaries.
#[derive(Debug, Default, PartialEq)]
pub struct ConversationsPage {
/// Conversation summaries ordered newest first.
pub items: Vec<ConversationItem>,
pub struct ThreadsPage {
/// Thread summaries ordered newest first.
pub items: Vec<ThreadItem>,
/// Opaque pagination token to resume after the last item, or `None` if end.
pub next_cursor: Option<Cursor>,
/// Total number of files touched while scanning this request.
@@ -33,9 +33,9 @@ pub struct ConversationsPage {
pub reached_scan_cap: bool,
}
/// Summary information for a conversation rollout file.
/// Summary information for a thread rollout file.
#[derive(Debug, PartialEq)]
pub struct ConversationItem {
pub struct ThreadItem {
/// Absolute path to the rollout file.
pub path: PathBuf,
/// First up to `HEAD_RECORD_LIMIT` JSONL records parsed as JSON (includes meta line).
@@ -46,6 +46,13 @@ pub struct ConversationItem {
pub updated_at: Option<String>,
}
#[allow(dead_code)]
#[deprecated(note = "use ThreadItem")]
pub type ConversationItem = ThreadItem;
#[allow(dead_code)]
#[deprecated(note = "use ThreadsPage")]
pub type ConversationsPage = ThreadsPage;
#[derive(Default)]
struct HeadTailSummary {
head: Vec<serde_json::Value>,
@@ -99,22 +106,22 @@ impl<'de> serde::Deserialize<'de> for Cursor {
}
}
/// Retrieve recorded conversation file paths with token pagination. The returned `next_cursor`
/// Retrieve recorded thread file paths with token pagination. The returned `next_cursor`
/// can be supplied on the next call to resume after the last returned item, resilient to
/// concurrent new sessions being appended. Ordering is stable by timestamp desc, then UUID desc.
pub(crate) async fn get_conversations(
pub(crate) async fn get_threads(
codex_home: &Path,
page_size: usize,
cursor: Option<&Cursor>,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
default_provider: &str,
) -> io::Result<ConversationsPage> {
) -> io::Result<ThreadsPage> {
let mut root = codex_home.to_path_buf();
root.push(SESSIONS_SUBDIR);
if !root.exists() {
return Ok(ConversationsPage {
return Ok(ThreadsPage {
items: Vec::new(),
next_cursor: None,
num_scanned_files: 0,
@@ -138,7 +145,7 @@ pub(crate) async fn get_conversations(
Ok(result)
}
/// Load conversation file paths from disk using directory traversal.
/// Load thread file paths from disk using directory traversal.
///
/// Directory layout: `~/.codex/sessions/YYYY/MM/DD/rollout-YYYY-MM-DDThh-mm-ss-<uuid>.jsonl`
/// Returned newest (latest) first.
@@ -148,8 +155,8 @@ async fn traverse_directories_for_paths(
anchor: Option<Cursor>,
allowed_sources: &[SessionSource],
provider_matcher: Option<&ProviderMatcher<'_>>,
) -> io::Result<ConversationsPage> {
let mut items: Vec<ConversationItem> = Vec::with_capacity(page_size);
) -> io::Result<ThreadsPage> {
let mut items: Vec<ThreadItem> = Vec::with_capacity(page_size);
let mut scanned_files = 0usize;
let mut anchor_passed = anchor.is_none();
let (anchor_ts, anchor_id) = match anchor {
@@ -232,7 +239,7 @@ async fn traverse_directories_for_paths(
.unwrap_or(None)
.or_else(|| created_at.clone());
}
items.push(ConversationItem {
items.push(ThreadItem {
path,
head,
created_at,
@@ -254,7 +261,7 @@ async fn traverse_directories_for_paths(
} else {
None
};
Ok(ConversationsPage {
Ok(ThreadsPage {
items,
next_cursor: next,
num_scanned_files: scanned_files,
@@ -279,7 +286,7 @@ pub fn parse_cursor(token: &str) -> Option<Cursor> {
Some(Cursor::new(ts, uuid))
}
fn build_next_cursor(items: &[ConversationItem]) -> Option<Cursor> {
fn build_next_cursor(items: &[ThreadItem]) -> Option<Cursor> {
let last = items.last()?;
let file_name = last.path.file_name()?.to_string_lossy();
let (ts, id) = parse_timestamp_uuid_from_filename(&file_name)?;
@@ -455,10 +462,10 @@ async fn file_modified_rfc3339(path: &Path) -> io::Result<Option<String>> {
Ok(dt.format(&Rfc3339).ok())
}
/// Locate a recorded conversation rollout file by its UUID string using the existing
/// Locate a recorded thread rollout file by its UUID string using the existing
/// paginated listing implementation. Returns `Ok(Some(path))` if found, `Ok(None)` if not present
/// or the id is invalid.
pub async fn find_conversation_path_by_id_str(
pub async fn find_thread_path_by_id_str(
codex_home: &Path,
id_str: &str,
) -> io::Result<Option<PathBuf>> {
codex-rs/core/src/rollout/mod.rs
+3 -1
View File
@@ -15,7 +15,9 @@ pub(crate) mod truncation;
pub use codex_protocol::protocol::SessionMeta;
pub(crate) use error::map_session_init_error;
pub use list::find_conversation_path_by_id_str;
pub use list::find_thread_path_by_id_str;
#[deprecated(note = "use find_thread_path_by_id_str")]
pub use list::find_thread_path_by_id_str as find_conversation_path_by_id_str;
pub use recorder::RolloutRecorder;
pub use recorder::RolloutRecorderParams;
codex-rs/core/src/rollout/recorder.rs
+19 -22
View File
@@ -6,7 +6,7 @@ use std::io::Error as IoError;
use std::path::Path;
use std::path::PathBuf;
use codex_protocol::ConversationId;
use codex_protocol::ThreadId;
use serde_json::Value;
use time::OffsetDateTime;
use time::format_description::FormatItem;
@@ -19,9 +19,9 @@ use tracing::info;
use tracing::warn;
use super::SESSIONS_SUBDIR;
use super::list::ConversationsPage;
use super::list::Cursor;
use super::list::get_conversations;
use super::list::ThreadsPage;
use super::list::get_threads;
use super::policy::is_persisted_response_item;
use crate::config::Config;
use crate::default_client::originator;
@@ -52,7 +52,7 @@ pub struct RolloutRecorder {
#[derive(Clone)]
pub enum RolloutRecorderParams {
Create {
conversation_id: ConversationId,
conversation_id: ThreadId,
instructions: Option<String>,
source: SessionSource,
},
@@ -79,7 +79,7 @@ enum RolloutCmd {
impl RolloutRecorderParams {
pub fn new(
conversation_id: ConversationId,
conversation_id: ThreadId,
instructions: Option<String>,
source: SessionSource,
) -> Self {
@@ -96,16 +96,16 @@ impl RolloutRecorderParams {
}
impl RolloutRecorder {
/// List conversations (rollout files) under the provided Codex home directory.
pub async fn list_conversations(
/// List threads (rollout files) under the provided Codex home directory.
pub async fn list_threads(
codex_home: &Path,
page_size: usize,
cursor: Option<&Cursor>,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
default_provider: &str,
) -> std::io::Result<ConversationsPage> {
get_conversations(
) -> std::io::Result<ThreadsPage> {
get_threads(
codex_home,
page_size,
cursor,
@@ -231,7 +231,7 @@ impl RolloutRecorder {
}
let mut items: Vec<RolloutItem> = Vec::new();
let mut conversation_id: Option<ConversationId> = None;
let mut thread_id: Option<ThreadId> = None;
for line in text.lines() {
if line.trim().is_empty() {
continue;
@@ -249,9 +249,9 @@ impl RolloutRecorder {
Ok(rollout_line) => match rollout_line.item {
RolloutItem::SessionMeta(session_meta_line) => {
// Use the FIRST SessionMeta encountered in the file as the canonical
// conversation id and main session information. Keep all items intact.
if conversation_id.is_none() {
conversation_id = Some(session_meta_line.meta.id);
// thread id and main session information. Keep all items intact.
if thread_id.is_none() {
thread_id = Some(session_meta_line.meta.id);
}
items.push(RolloutItem::SessionMeta(session_meta_line));
}
@@ -275,12 +275,12 @@ impl RolloutRecorder {
}
info!(
"Resumed rollout with {} items, conversation ID: {:?}",
"Resumed rollout with {} items, thread ID: {:?}",
items.len(),
conversation_id
thread_id
);
let conversation_id = conversation_id
.ok_or_else(|| IoError::other("failed to parse conversation ID from rollout file"))?;
let conversation_id = thread_id
.ok_or_else(|| IoError::other("failed to parse thread ID from rollout file"))?;
if items.is_empty() {
return Ok(InitialHistory::New);
@@ -318,16 +318,13 @@ struct LogFileInfo {
path: PathBuf,
/// Session ID (also embedded in filename).
conversation_id: ConversationId,
conversation_id: ThreadId,
/// Timestamp for the start of the session.
timestamp: OffsetDateTime,
}
fn create_log_file(
config: &Config,
conversation_id: ConversationId,
) -> std::io::Result<LogFileInfo> {
fn create_log_file(config: &Config, conversation_id: ThreadId) -> std::io::Result<LogFileInfo> {
// Resolve ~/.codex/sessions/YYYY/MM/DD and create it if missing.
let timestamp = OffsetDateTime::now_local()
.map_err(|e| IoError::other(format!("failed to get local time: {e}")))?;
codex-rs/core/src/rollout/tests.rs
+39 -39
View File
@@ -13,12 +13,12 @@ use time::macros::format_description;
use uuid::Uuid;
use crate::rollout::INTERACTIVE_SESSION_SOURCES;
use crate::rollout::list::ConversationItem;
use crate::rollout::list::ConversationsPage;
use crate::rollout::list::Cursor;
use crate::rollout::list::get_conversations;
use crate::rollout::list::ThreadItem;
use crate::rollout::list::ThreadsPage;
use crate::rollout::list::get_threads;
use anyhow::Result;
use codex_protocol::ConversationId;
use codex_protocol::ThreadId;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EventMsg;
@@ -162,7 +162,7 @@ async fn test_list_conversations_latest_first() {
.unwrap();
let provider_filter = provider_vec(&[TEST_PROVIDER]);
let page = get_conversations(
let page = get_threads(
home,
10,
None,
@@ -227,21 +227,21 @@ async fn test_list_conversations_latest_first() {
let updated_times: Vec<Option<String>> =
page.items.iter().map(|i| i.updated_at.clone()).collect();
let expected = ConversationsPage {
let expected = ThreadsPage {
items: vec![
ConversationItem {
ThreadItem {
path: p1,
head: head_3,
created_at: Some("2025-01-03T12-00-00".into()),
updated_at: updated_times.first().cloned().flatten(),
},
ConversationItem {
ThreadItem {
path: p2,
head: head_2,
created_at: Some("2025-01-02T12-00-00".into()),
updated_at: updated_times.get(1).cloned().flatten(),
},
ConversationItem {
ThreadItem {
path: p3,
head: head_1,
created_at: Some("2025-01-01T12-00-00".into()),
@@ -311,7 +311,7 @@ async fn test_pagination_cursor() {
.unwrap();
let provider_filter = provider_vec(&[TEST_PROVIDER]);
let page1 = get_conversations(
let page1 = get_threads(
home,
2,
None,
@@ -357,15 +357,15 @@ async fn test_pagination_cursor() {
page1.items.iter().map(|i| i.updated_at.clone()).collect();
let expected_cursor1: Cursor =
serde_json::from_str(&format!("\"2025-03-04T09-00-00|{u4}\"")).unwrap();
let expected_page1 = ConversationsPage {
let expected_page1 = ThreadsPage {
items: vec![
ConversationItem {
ThreadItem {
path: p5,
head: head_5,
created_at: Some("2025-03-05T09-00-00".into()),
updated_at: updated_page1.first().cloned().flatten(),
},
ConversationItem {
ThreadItem {
path: p4,
head: head_4,
created_at: Some("2025-03-04T09-00-00".into()),
@@ -378,7 +378,7 @@ async fn test_pagination_cursor() {
};
assert_eq!(page1, expected_page1);
let page2 = get_conversations(
let page2 = get_threads(
home,
2,
page1.next_cursor.as_ref(),
@@ -424,15 +424,15 @@ async fn test_pagination_cursor() {
page2.items.iter().map(|i| i.updated_at.clone()).collect();
let expected_cursor2: Cursor =
serde_json::from_str(&format!("\"2025-03-02T09-00-00|{u2}\"")).unwrap();
let expected_page2 = ConversationsPage {
let expected_page2 = ThreadsPage {
items: vec![
ConversationItem {
ThreadItem {
path: p3,
head: head_3,
created_at: Some("2025-03-03T09-00-00".into()),
updated_at: updated_page2.first().cloned().flatten(),
},
ConversationItem {
ThreadItem {
path: p2,
head: head_2,
created_at: Some("2025-03-02T09-00-00".into()),
@@ -445,7 +445,7 @@ async fn test_pagination_cursor() {
};
assert_eq!(page2, expected_page2);
let page3 = get_conversations(
let page3 = get_threads(
home,
2,
page2.next_cursor.as_ref(),
@@ -473,8 +473,8 @@ async fn test_pagination_cursor() {
})];
let updated_page3: Vec<Option<String>> =
page3.items.iter().map(|i| i.updated_at.clone()).collect();
let expected_page3 = ConversationsPage {
items: vec![ConversationItem {
let expected_page3 = ThreadsPage {
items: vec![ThreadItem {
path: p1,
head: head_1,
created_at: Some("2025-03-01T09-00-00".into()),
@@ -488,7 +488,7 @@ async fn test_pagination_cursor() {
}
#[tokio::test]
async fn test_get_conversation_contents() {
async fn test_get_thread_contents() {
let temp = TempDir::new().unwrap();
let home = temp.path();
@@ -497,7 +497,7 @@ async fn test_get_conversation_contents() {
write_session_file(home, ts, uuid, 2, Some(SessionSource::VSCode)).unwrap();
let provider_filter = provider_vec(&[TEST_PROVIDER]);
let page = get_conversations(
let page = get_threads(
home,
1,
None,
@@ -528,8 +528,8 @@ async fn test_get_conversation_contents() {
"source": "vscode",
"model_provider": "test-provider",
})];
let expected_page = ConversationsPage {
items: vec![ConversationItem {
let expected_page = ThreadsPage {
items: vec![ThreadItem {
path: expected_path,
head: expected_head,
created_at: Some(ts.into()),
@@ -579,7 +579,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> {
let file_path = day_dir.join(format!("rollout-{ts}-{uuid}.jsonl"));
let mut file = File::create(&file_path)?;
let conversation_id = ConversationId::from_string(&uuid.to_string())?;
let conversation_id = ThreadId::from_string(&uuid.to_string())?;
let meta_line = RolloutLine {
timestamp: ts.to_string(),
item: RolloutItem::SessionMeta(SessionMetaLine {
@@ -625,7 +625,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> {
drop(file);
let provider_filter = provider_vec(&[TEST_PROVIDER]);
let page = get_conversations(
let page = get_threads(
home,
1,
None,
@@ -664,7 +664,7 @@ async fn test_stable_ordering_same_second_pagination() {
write_session_file(home, ts, u3, 0, Some(SessionSource::VSCode)).unwrap();
let provider_filter = provider_vec(&[TEST_PROVIDER]);
let page1 = get_conversations(
let page1 = get_threads(
home,
2,
None,
@@ -702,15 +702,15 @@ async fn test_stable_ordering_same_second_pagination() {
let updated_page1: Vec<Option<String>> =
page1.items.iter().map(|i| i.updated_at.clone()).collect();
let expected_cursor1: Cursor = serde_json::from_str(&format!("\"{ts}|{u2}\"")).unwrap();
let expected_page1 = ConversationsPage {
let expected_page1 = ThreadsPage {
items: vec![
ConversationItem {
ThreadItem {
path: p3,
head: head(u3),
created_at: Some(ts.to_string()),
updated_at: updated_page1.first().cloned().flatten(),
},
ConversationItem {
ThreadItem {
path: p2,
head: head(u2),
created_at: Some(ts.to_string()),
@@ -723,7 +723,7 @@ async fn test_stable_ordering_same_second_pagination() {
};
assert_eq!(page1, expected_page1);
let page2 = get_conversations(
let page2 = get_threads(
home,
2,
page1.next_cursor.as_ref(),
@@ -741,8 +741,8 @@ async fn test_stable_ordering_same_second_pagination() {
.join(format!("rollout-2025-07-01T00-00-00-{u1}.jsonl"));
let updated_page2: Vec<Option<String>> =
page2.items.iter().map(|i| i.updated_at.clone()).collect();
let expected_page2 = ConversationsPage {
items: vec![ConversationItem {
let expected_page2 = ThreadsPage {
items: vec![ThreadItem {
path: p1,
head: head(u1),
created_at: Some(ts.to_string()),
@@ -781,7 +781,7 @@ async fn test_source_filter_excludes_non_matching_sessions() {
.unwrap();
let provider_filter = provider_vec(&[TEST_PROVIDER]);
let interactive_only = get_conversations(
let interactive_only = get_threads(
home,
10,
None,
@@ -802,7 +802,7 @@ async fn test_source_filter_excludes_non_matching_sessions() {
path.ends_with("rollout-2025-08-02T10-00-00-00000000-0000-0000-0000-00000000002a.jsonl")
}));
let all_sessions = get_conversations(home, 10, None, NO_SOURCE_FILTER, None, TEST_PROVIDER)
let all_sessions = get_threads(home, 10, None, NO_SOURCE_FILTER, None, TEST_PROVIDER)
.await
.unwrap();
let all_paths: Vec<_> = all_sessions
@@ -856,7 +856,7 @@ async fn test_model_provider_filter_selects_only_matching_sessions() -> Result<(
let openai_id_str = openai_id.to_string();
let none_id_str = none_id.to_string();
let openai_filter = provider_vec(&["openai"]);
let openai_sessions = get_conversations(
let openai_sessions = get_threads(
home,
10,
None,
@@ -881,7 +881,7 @@ async fn test_model_provider_filter_selects_only_matching_sessions() -> Result<(
assert!(openai_ids.contains(&none_id_str));
let beta_filter = provider_vec(&["beta"]);
let beta_sessions = get_conversations(
let beta_sessions = get_threads(
home,
10,
None,
@@ -901,7 +901,7 @@ async fn test_model_provider_filter_selects_only_matching_sessions() -> Result<(
assert_eq!(beta_head, Some(beta_id_str.as_str()));
let unknown_filter = provider_vec(&["unknown"]);
let unknown_sessions = get_conversations(
let unknown_sessions = get_threads(
home,
10,
None,
@@ -912,7 +912,7 @@ async fn test_model_provider_filter_selects_only_matching_sessions() -> Result<(
.await?;
assert!(unknown_sessions.items.is_empty());
let all_sessions = get_conversations(home, 10, None, NO_SOURCE_FILTER, None, "openai").await?;
let all_sessions = get_threads(home, 10, None, NO_SOURCE_FILTER, None, "openai").await?;
assert_eq!(all_sessions.items.len(), 3);
Ok(())
codex-rs/core/src/tasks/review.rs
+2 -2
View File
@@ -15,7 +15,7 @@ use tokio_util::sync::CancellationToken;
use crate::codex::Session;
use crate::codex::TurnContext;
use crate::codex_delegate::run_codex_conversation_one_shot;
use crate::codex_delegate::run_codex_thread_one_shot;
use crate::review_format::format_review_findings_block;
use crate::review_format::render_review_output_text;
use crate::state::TaskKind;
@@ -92,7 +92,7 @@ async fn start_review_conversation(
sub_agent_config.base_instructions = Some(crate::REVIEW_PROMPT.to_string());
sub_agent_config.model = Some(config.review_model.clone());
(run_codex_conversation_one_shot(
(run_codex_thread_one_shot(
sub_agent_config,
session.auth_manager(),
session.models_manager(),
codex-rs/core/src/conversation_manager.rs → codex-rs/core/src/thread_manager.rs
+110 -86
View File
@@ -7,7 +7,7 @@ use crate::agent::AgentControl;
use crate::codex::Codex;
use crate::codex::CodexSpawnOk;
use crate::codex::INITIAL_SUBMIT_ID;
use crate::codex_conversation::CodexConversation;
use crate::codex_thread::CodexThread;
use crate::config::Config;
use crate::error::CodexErr;
use crate::error::Result as CodexResult;
@@ -18,7 +18,7 @@ use crate::protocol::SessionConfiguredEvent;
use crate::rollout::RolloutRecorder;
use crate::rollout::truncation;
use crate::skills::SkillsManager;
use codex_protocol::ConversationId;
use codex_protocol::ThreadId;
use codex_protocol::openai_models::ModelPreset;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::Op;
@@ -31,38 +31,38 @@ use std::sync::Arc;
use tempfile::TempDir;
use tokio::sync::RwLock;
/// Represents a newly created Codex conversation, including the first event
/// Represents a newly created Codex thread (formerly called a conversation), including the first event
/// (which is [`EventMsg::SessionConfigured`]).
pub struct NewConversation {
pub conversation_id: ConversationId,
pub conversation: Arc<CodexConversation>,
pub struct NewThread {
pub thread_id: ThreadId,
pub thread: Arc<CodexThread>,
pub session_configured: SessionConfiguredEvent,
}
/// [`ConversationManager`] is responsible for creating conversations and
/// maintaining them in memory.
pub struct ConversationManager {
state: Arc<ConversationManagerState>,
/// [`ThreadManager`] is responsible for creating threads and maintaining
/// them in memory.
pub struct ThreadManager {
state: Arc<ThreadManagerState>,
#[cfg(any(test, feature = "test-support"))]
_test_codex_home_guard: Option<TempDir>,
}
/// Shared, `Arc`-owned state for [`ConversationManager`]. This `Arc` is required to have a single
/// Shared, `Arc`-owned state for [`ThreadManager`]. This `Arc` is required to have a single
/// `Arc` reference that can be downgraded to by `AgentControl` while preventing every single
/// function to require an `Arc<&Self>`.
pub(crate) struct ConversationManagerState {
conversations: Arc<RwLock<HashMap<ConversationId, Arc<CodexConversation>>>>,
pub(crate) struct ThreadManagerState {
threads: Arc<RwLock<HashMap<ThreadId, Arc<CodexThread>>>>,
auth_manager: Arc<AuthManager>,
models_manager: Arc<ModelsManager>,
skills_manager: Arc<SkillsManager>,
session_source: SessionSource,
}
impl ConversationManager {
impl ThreadManager {
pub fn new(auth_manager: Arc<AuthManager>, session_source: SessionSource) -> Self {
Self {
state: Arc::new(ConversationManagerState {
conversations: Arc::new(RwLock::new(HashMap::new())),
state: Arc::new(ThreadManagerState {
threads: Arc::new(RwLock::new(HashMap::new())),
models_manager: Arc::new(ModelsManager::new(auth_manager.clone())),
skills_manager: Arc::new(SkillsManager::new(
auth_manager.codex_home().to_path_buf(),
@@ -96,8 +96,8 @@ impl ConversationManager {
) -> Self {
let auth_manager = AuthManager::from_auth_for_testing_with_home(auth, codex_home);
Self {
state: Arc::new(ConversationManagerState {
conversations: Arc::new(RwLock::new(HashMap::new())),
state: Arc::new(ThreadManagerState {
threads: Arc::new(RwLock::new(HashMap::new())),
models_manager: Arc::new(ModelsManager::with_provider(
auth_manager.clone(),
provider,
@@ -128,16 +128,13 @@ impl ConversationManager {
self.state.models_manager.list_models(config).await
}
pub async fn get_conversation(
&self,
conversation_id: ConversationId,
) -> CodexResult<Arc<CodexConversation>> {
self.state.get_conversation(conversation_id).await
pub async fn get_thread(&self, thread_id: ThreadId) -> CodexResult<Arc<CodexThread>> {
self.state.get_thread(thread_id).await
}
pub async fn new_conversation(&self, config: Config) -> CodexResult<NewConversation> {
pub async fn start_thread(&self, config: Config) -> CodexResult<NewThread> {
self.state
.spawn_conversation(
.spawn_thread(
config,
InitialHistory::New,
Arc::clone(&self.state.auth_manager),
@@ -146,56 +143,96 @@ impl ConversationManager {
.await
}
pub async fn resume_thread_from_rollout(
&self,
config: Config,
rollout_path: PathBuf,
auth_manager: Arc<AuthManager>,
) -> CodexResult<NewThread> {
let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?;
self.resume_thread_with_history(config, initial_history, auth_manager)
.await
}
pub async fn resume_thread_with_history(
&self,
config: Config,
initial_history: InitialHistory,
auth_manager: Arc<AuthManager>,
) -> CodexResult<NewThread> {
self.state
.spawn_thread(config, initial_history, auth_manager, self.agent_control())
.await
}
#[deprecated(note = "use get_thread")]
pub async fn get_conversation(&self, thread_id: ThreadId) -> CodexResult<Arc<CodexThread>> {
self.get_thread(thread_id).await
}
#[deprecated(note = "use start_thread")]
pub async fn new_conversation(&self, config: Config) -> CodexResult<NewThread> {
self.start_thread(config).await
}
#[deprecated(note = "use resume_thread_from_rollout")]
pub async fn resume_conversation_from_rollout(
&self,
config: Config,
rollout_path: PathBuf,
auth_manager: Arc<AuthManager>,
) -> CodexResult<NewConversation> {
let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?;
self.resume_conversation_with_history(config, initial_history, auth_manager)
) -> CodexResult<NewThread> {
self.resume_thread_from_rollout(config, rollout_path, auth_manager)
.await
}
#[deprecated(note = "use resume_thread_with_history")]
pub async fn resume_conversation_with_history(
&self,
config: Config,
initial_history: InitialHistory,
auth_manager: Arc<AuthManager>,
) -> CodexResult<NewConversation> {
self.state
.spawn_conversation(config, initial_history, auth_manager, self.agent_control())
) -> CodexResult<NewThread> {
self.resume_thread_with_history(config, initial_history, auth_manager)
.await
}
/// Removes the conversation from the manager's internal map, though the conversation is stored
/// as `Arc<CodexConversation>`, it is possible that other references to it exist elsewhere.
/// Returns the conversation if the conversation was found and removed.
pub async fn remove_conversation(
&self,
conversation_id: &ConversationId,
) -> Option<Arc<CodexConversation>> {
self.state
.conversations
.write()
.await
.remove(conversation_id)
#[deprecated(note = "use remove_thread")]
pub async fn remove_conversation(&self, thread_id: &ThreadId) -> Option<Arc<CodexThread>> {
self.remove_thread(thread_id).await
}
/// Fork an existing conversation by taking messages up to the given position (not including
/// the message at the given position) and starting a new conversation with identical
/// configuration (unless overridden by the caller's `config`). The new conversation will have
/// a fresh id.
#[deprecated(note = "use fork_thread")]
pub async fn fork_conversation(
&self,
nth_user_message: usize,
config: Config,
path: PathBuf,
) -> CodexResult<NewConversation> {
) -> CodexResult<NewThread> {
self.fork_thread(nth_user_message, config, path).await
}
/// Removes the thread from the manager's internal map, though the thread is stored
/// as `Arc<CodexThread>`, it is possible that other references to it exist elsewhere.
/// Returns the thread if the thread was found and removed.
pub async fn remove_thread(&self, thread_id: &ThreadId) -> Option<Arc<CodexThread>> {
self.state.threads.write().await.remove(thread_id)
}
/// Fork an existing thread by taking messages up to the given position (not including
/// the message at the given position) and starting a new thread with identical
/// configuration (unless overridden by the caller's `config`). The new thread will have
/// a fresh id.
pub async fn fork_thread(
&self,
nth_user_message: usize,
config: Config,
path: PathBuf,
) -> CodexResult<NewThread> {
let history = RolloutRecorder::get_rollout_history(&path).await?;
let history = truncate_before_nth_user_message(history, nth_user_message);
self.state
.spawn_conversation(
.spawn_thread(
config,
history,
Arc::clone(&self.state.auth_manager),
@@ -209,36 +246,26 @@ impl ConversationManager {
}
}
impl ConversationManagerState {
pub(crate) async fn get_conversation(
&self,
conversation_id: ConversationId,
) -> CodexResult<Arc<CodexConversation>> {
let conversations = self.conversations.read().await;
conversations
.get(&conversation_id)
impl ThreadManagerState {
pub(crate) async fn get_thread(&self, thread_id: ThreadId) -> CodexResult<Arc<CodexThread>> {
let threads = self.threads.read().await;
threads
.get(&thread_id)
.cloned()
.ok_or_else(|| CodexErr::ConversationNotFound(conversation_id))
.ok_or_else(|| CodexErr::ThreadNotFound(thread_id))
}
pub(crate) async fn send_op(
&self,
conversation_id: ConversationId,
op: Op,
) -> CodexResult<String> {
self.get_conversation(conversation_id)
.await?
.submit(op)
.await
pub(crate) async fn send_op(&self, thread_id: ThreadId, op: Op) -> CodexResult<String> {
self.get_thread(thread_id).await?.submit(op).await
}
#[allow(dead_code)] // Used by upcoming multi-agent tooling.
pub(crate) async fn spawn_new_conversation(
pub(crate) async fn spawn_new_thread(
&self,
config: Config,
agent_control: AgentControl,
) -> CodexResult<NewConversation> {
self.spawn_conversation(
) -> CodexResult<NewThread> {
self.spawn_thread(
config,
InitialHistory::New,
Arc::clone(&self.auth_manager),
@@ -247,16 +274,15 @@ impl ConversationManagerState {
.await
}
pub(crate) async fn spawn_conversation(
pub(crate) async fn spawn_thread(
&self,
config: Config,
initial_history: InitialHistory,
auth_manager: Arc<AuthManager>,
agent_control: AgentControl,
) -> CodexResult<NewConversation> {
) -> CodexResult<NewThread> {
let CodexSpawnOk {
codex,
conversation_id,
codex, thread_id, ..
} = Codex::spawn(
config,
auth_manager,
@@ -267,14 +293,14 @@ impl ConversationManagerState {
agent_control,
)
.await?;
self.finalize_spawn(codex, conversation_id).await
self.finalize_thread_spawn(codex, thread_id).await
}
async fn finalize_spawn(
async fn finalize_thread_spawn(
&self,
codex: Codex,
conversation_id: ConversationId,
) -> CodexResult<NewConversation> {
thread_id: ThreadId,
) -> CodexResult<NewThread> {
let event = codex.next_event().await?;
let session_configured = match event {
Event {
@@ -286,18 +312,16 @@ impl ConversationManagerState {
}
};
let conversation = Arc::new(CodexConversation::new(
let thread = Arc::new(CodexThread::new(
codex,
session_configured.rollout_path.clone(),
));
self.conversations
.write()
.await
.insert(conversation_id, conversation.clone());
self.threads.write().await.insert(thread_id, thread.clone());
Ok(NewConversation {
conversation_id,
conversation,
#[allow(deprecated)]
Ok(NewThread {
thread_id,
thread,
session_configured,
})
}
codex-rs/core/src/tools/spec.rs
+1 -1
View File
@@ -408,7 +408,7 @@ fn create_view_image_tool() -> ToolSpec {
ToolSpec::Function(ResponsesApiTool {
name: "view_image".to_string(),
description:
"Attach a local image (by filesystem path) to the conversation context for this turn."
"Attach a local image (by filesystem path) to the thread context for this turn."
.to_string(),
strict: false,
parameters: JsonSchema::Object {