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

[codex] Remove remote thread store implementation (#21596)

Browse Source 
Remove the remote thread-store backend and checked-in protobuf
artifacts. We've moved these into another crate that link against this
one.

Also remove the config settings for thread store backend selection,
since we'll instead pass an instantiated thread store into the core-api
crate's main entrypoint.
This commit is contained in:
Tom
2026-05-07 17:02:46 -07:00
committed by GitHub
Unverified
parent a3de5bde6e
commit 79ad209ce6
22 changed files with 42 additions and 2901 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/thread-store/src/lib.rs
-2
View File
@@ -8,7 +8,6 @@ mod error;
mod in_memory;
mod live_thread;
mod local;
mod remote;
mod store;
mod types;
@@ -20,7 +19,6 @@ pub use live_thread::LiveThread;
pub use live_thread::LiveThreadInitGuard;
pub use local::LocalThreadStore;
pub use local::LocalThreadStoreConfig;
pub use remote::RemoteThreadStore;
pub use store::ThreadStore;
pub use types::AppendThreadItemsParams;
pub use types::ArchiveThreadParams;
codex-rs/thread-store/src/remote/AGENTS.md
-13
View File
@@ -1,13 +0,0 @@
# Remote Thread Store
- The Rust protobuf output in `proto/codex.thread_store.v1.rs` is checked in.
- Do not add build-time protobuf generation to `codex-thread-store` unless the Bazel/Cargo story is intentionally changed.
- When `proto/codex.thread_store.v1.proto` changes, regenerate the Rust file manually and include both files in the same commit.
Run this from the repository root:
```sh
./codex-rs/thread-store/scripts/generate-proto.sh
```
The command requires `protoc` to be available on `PATH`.
codex-rs/thread-store/src/remote/helpers.rs
-457
View File
@@ -1,457 +0,0 @@
use std::path::PathBuf;
use std::str::FromStr;
use chrono::DateTime;
use chrono::Utc;
use codex_git_utils::GitSha;
use codex_protocol::AgentPath;
use codex_protocol::ThreadId;
use codex_protocol::dynamic_tools::DynamicToolSpec;
use codex_protocol::models::BaseInstructions;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::GitInfo;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::ThreadMemoryMode;
use codex_protocol::protocol::ThreadSource;
use super::proto;
use crate::GitInfoPatch;
use crate::OptionalStringPatch;
use crate::SortDirection;
use crate::StoredThread;
use crate::StoredThreadHistory;
use crate::ThreadEventPersistenceMode;
use crate::ThreadMetadataPatch;
use crate::ThreadPersistenceMetadata;
use crate::ThreadSortKey;
use crate::ThreadStoreError;
use crate::ThreadStoreResult;
pub(super) fn remote_status_to_error(status: tonic::Status) -> ThreadStoreError {
match status.code() {
tonic::Code::InvalidArgument => ThreadStoreError::InvalidRequest {
message: status.message().to_string(),
},
tonic::Code::AlreadyExists | tonic::Code::FailedPrecondition | tonic::Code::Aborted => {
ThreadStoreError::Conflict {
message: status.message().to_string(),
}
}
_ => ThreadStoreError::Internal {
message: format!("remote thread store request failed: {status}"),
},
}
}
pub(super) fn remote_status_to_thread_error(
status: tonic::Status,
thread_id: ThreadId,
) -> ThreadStoreError {
if status.code() == tonic::Code::NotFound {
return ThreadStoreError::ThreadNotFound { thread_id };
}
remote_status_to_error(status)
}
pub(super) fn proto_thread_id_request(thread_id: ThreadId) -> proto::ThreadIdRequest {
proto::ThreadIdRequest {
thread_id: thread_id.to_string(),
}
}
pub(super) fn proto_sort_key(sort_key: ThreadSortKey) -> proto::ThreadSortKey {
match sort_key {
ThreadSortKey::CreatedAt => proto::ThreadSortKey::CreatedAt,
ThreadSortKey::UpdatedAt => proto::ThreadSortKey::UpdatedAt,
}
}
pub(super) fn proto_sort_direction(sort_direction: SortDirection) -> proto::SortDirection {
match sort_direction {
SortDirection::Asc => proto::SortDirection::Asc,
SortDirection::Desc => proto::SortDirection::Desc,
}
}
pub(super) fn proto_event_persistence_mode(
mode: ThreadEventPersistenceMode,
) -> proto::ThreadEventPersistenceMode {
match mode {
ThreadEventPersistenceMode::Limited => proto::ThreadEventPersistenceMode::Limited,
ThreadEventPersistenceMode::Extended => proto::ThreadEventPersistenceMode::Extended,
}
}
pub(super) fn proto_session_source(source: &SessionSource) -> proto::SessionSource {
match source {
SessionSource::Cli => proto_source(proto::SessionSourceKind::Cli),
SessionSource::VSCode => proto_source(proto::SessionSourceKind::Vscode),
SessionSource::Exec => proto_source(proto::SessionSourceKind::Exec),
SessionSource::Mcp => proto_source(proto::SessionSourceKind::AppServer),
SessionSource::Custom(custom) => proto::SessionSource {
kind: proto::SessionSourceKind::Custom.into(),
custom: Some(custom.clone()),
..Default::default()
},
SessionSource::SubAgent(SubAgentSource::Review) => {
proto_source(proto::SessionSourceKind::SubAgentReview)
}
SessionSource::SubAgent(SubAgentSource::Compact) => {
proto_source(proto::SessionSourceKind::SubAgentCompact)
}
SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth,
agent_path,
agent_nickname,
agent_role,
}) => proto::SessionSource {
kind: proto::SessionSourceKind::SubAgentThreadSpawn.into(),
sub_agent_parent_thread_id: Some(parent_thread_id.to_string()),
sub_agent_depth: Some(*depth),
sub_agent_path: agent_path.as_ref().map(|path| path.as_str().to_string()),
sub_agent_nickname: agent_nickname.clone(),
sub_agent_role: agent_role.clone(),
..Default::default()
},
SessionSource::SubAgent(SubAgentSource::MemoryConsolidation) => {
proto_source(proto::SessionSourceKind::SubAgentMemoryConsolidation)
}
SessionSource::SubAgent(SubAgentSource::Other(other)) => proto::SessionSource {
kind: proto::SessionSourceKind::SubAgentOther.into(),
sub_agent_other: Some(other.clone()),
..Default::default()
},
SessionSource::Internal(_) => proto_source(proto::SessionSourceKind::Unknown),
SessionSource::Unknown => proto_source(proto::SessionSourceKind::Unknown),
}
}
fn proto_source(kind: proto::SessionSourceKind) -> proto::SessionSource {
proto::SessionSource {
kind: kind.into(),
..Default::default()
}
}
pub(super) fn serialize_json<T: serde::Serialize>(
value: &T,
field_name: &str,
) -> ThreadStoreResult<String> {
serde_json::to_string(value).map_err(|err| ThreadStoreError::InvalidRequest {
message: format!("failed to serialize {field_name} for remote thread store: {err}"),
})
}
fn deserialize_json<T: serde::de::DeserializeOwned>(
json: &str,
field_name: &str,
) -> ThreadStoreResult<T> {
serde_json::from_str(json).map_err(|err| ThreadStoreError::InvalidRequest {
message: format!("remote thread store returned invalid {field_name}: {err}"),
})
}
pub(super) fn serialize_json_vec<T: serde::Serialize>(
values: &[T],
field_name: &str,
) -> ThreadStoreResult<Vec<String>> {
values
.iter()
.map(|value| serialize_json(value, field_name))
.collect()
}
fn deserialize_json_vec<T: serde::de::DeserializeOwned>(
values: &[String],
field_name: &str,
) -> ThreadStoreResult<Vec<T>> {
values
.iter()
.map(|value| deserialize_json(value, field_name))
.collect()
}
pub(super) fn base_instructions_json(
base_instructions: &BaseInstructions,
) -> ThreadStoreResult<String> {
serialize_json(base_instructions, "base_instructions")
}
pub(super) fn dynamic_tools_json(
dynamic_tools: &[DynamicToolSpec],
) -> ThreadStoreResult<Vec<String>> {
serialize_json_vec(dynamic_tools, "dynamic_tool")
}
pub(super) fn thread_persistence_metadata_json(
metadata: &ThreadPersistenceMetadata,
) -> ThreadStoreResult<String> {
serialize_json(metadata, "thread_persistence_metadata")
}
pub(super) fn rollout_items_json(items: &[RolloutItem]) -> ThreadStoreResult<Vec<String>> {
serialize_json_vec(items, "rollout_item")
}
pub(super) fn stored_thread_history_from_proto(
history: proto::StoredThreadHistory,
) -> ThreadStoreResult<StoredThreadHistory> {
let thread_id = ThreadId::from_string(&history.thread_id).map_err(|err| {
ThreadStoreError::InvalidRequest {
message: format!("remote thread store returned invalid history thread_id: {err}"),
}
})?;
Ok(StoredThreadHistory {
thread_id,
items: deserialize_json_vec(&history.items_json, "rollout_item")?,
})
}
pub(super) fn proto_metadata_patch(patch: ThreadMetadataPatch) -> proto::ThreadMetadataPatch {
proto::ThreadMetadataPatch {
name: patch.name,
memory_mode: patch.memory_mode.map(proto_memory_mode).map(Into::into),
git_info: patch.git_info.map(proto_git_info_patch),
}
}
fn proto_memory_mode(memory_mode: ThreadMemoryMode) -> proto::ThreadMemoryMode {
match memory_mode {
ThreadMemoryMode::Enabled => proto::ThreadMemoryMode::Enabled,
ThreadMemoryMode::Disabled => proto::ThreadMemoryMode::Disabled,
}
}
fn proto_git_info_patch(patch: GitInfoPatch) -> proto::GitInfoPatch {
proto::GitInfoPatch {
sha: Some(proto_optional_string_patch(patch.sha)),
branch: Some(proto_optional_string_patch(patch.branch)),
origin_url: Some(proto_optional_string_patch(patch.origin_url)),
}
}
fn proto_optional_string_patch(patch: OptionalStringPatch) -> proto::OptionalStringPatch {
match patch {
None => proto::OptionalStringPatch {
kind: proto::OptionalStringPatchKind::Unset.into(),
value: None,
},
Some(None) => proto::OptionalStringPatch {
kind: proto::OptionalStringPatchKind::Clear.into(),
value: None,
},
Some(Some(value)) => proto::OptionalStringPatch {
kind: proto::OptionalStringPatchKind::Set.into(),
value: Some(value),
},
}
}
pub(super) fn stored_thread_from_proto(
thread: proto::StoredThread,
) -> ThreadStoreResult<StoredThread> {
// Keep this mapping boring: the proto mirrors StoredThread for remote-readable
// summary fields, except for Rust domain types that cross gRPC as stable scalar
// values. Local-only fields such as rollout_path intentionally stay local.
let source = thread
.source
.as_ref()
.map(session_source_from_proto)
.transpose()?
.unwrap_or(SessionSource::Unknown);
let thread_id = ThreadId::from_string(&thread.thread_id).map_err(|err| {
ThreadStoreError::InvalidRequest {
message: format!("remote thread store returned invalid thread_id: {err}"),
}
})?;
let forked_from_id = thread
.forked_from_id
.as_deref()
.map(ThreadId::from_string)
.transpose()
.map_err(|err| ThreadStoreError::InvalidRequest {
message: format!("remote thread store returned invalid forked_from_id: {err}"),
})?;
Ok(StoredThread {
thread_id,
rollout_path: thread.rollout_path.map(PathBuf::from),
forked_from_id,
preview: thread.preview,
name: thread.name,
model_provider: thread.model_provider,
model: thread.model,
reasoning_effort: thread
.reasoning_effort
.as_deref()
.map(parse_reasoning_effort)
.transpose()?,
created_at: datetime_from_unix(thread.created_at)?,
updated_at: datetime_from_unix(thread.updated_at)?,
archived_at: thread.archived_at.map(datetime_from_unix).transpose()?,
cwd: PathBuf::from(thread.cwd),
cli_version: thread.cli_version,
source,
thread_source: thread
.thread_source
.map(|thread_source| thread_source.parse::<ThreadSource>())
.transpose()
.map_err(|error| ThreadStoreError::Internal { message: error })?,
agent_nickname: thread.agent_nickname,
agent_role: thread.agent_role,
agent_path: thread.agent_path,
git_info: thread.git_info.map(git_info_from_proto),
approval_mode: thread
.approval_mode_json
.as_deref()
.map(|json| deserialize_json(json, "approval_mode"))
.transpose()?
.unwrap_or(AskForApproval::OnRequest),
sandbox_policy: thread
.sandbox_policy_json
.as_deref()
.map(|json| deserialize_json(json, "sandbox_policy"))
.transpose()?
.unwrap_or_else(SandboxPolicy::new_read_only_policy),
token_usage: thread
.token_usage_json
.as_deref()
.map(|json| deserialize_json(json, "token_usage"))
.transpose()?,
first_user_message: thread.first_user_message,
history: thread
.history
.map(stored_thread_history_from_proto)
.transpose()?,
})
}
#[cfg(test)]
pub(super) fn stored_thread_to_proto(thread: StoredThread) -> proto::StoredThread {
proto::StoredThread {
thread_id: thread.thread_id.to_string(),
forked_from_id: thread.forked_from_id.map(|thread_id| thread_id.to_string()),
preview: thread.preview,
name: thread.name,
model_provider: thread.model_provider,
model: thread.model,
created_at: thread.created_at.timestamp(),
updated_at: thread.updated_at.timestamp(),
archived_at: thread.archived_at.map(|timestamp| timestamp.timestamp()),
cwd: thread.cwd.to_string_lossy().into_owned(),
cli_version: thread.cli_version,
source: Some(proto_session_source(&thread.source)),
thread_source: thread.thread_source.map(|source| source.to_string()),
git_info: thread.git_info.map(git_info_to_proto),
agent_nickname: thread.agent_nickname,
agent_role: thread.agent_role,
agent_path: thread.agent_path,
reasoning_effort: thread.reasoning_effort.map(|effort| effort.to_string()),
first_user_message: thread.first_user_message,
rollout_path: thread
.rollout_path
.map(|path| path.to_string_lossy().into_owned()),
approval_mode_json: Some(serialize_json(&thread.approval_mode, "approval_mode").unwrap()),
sandbox_policy_json: Some(
serialize_json(&thread.sandbox_policy, "sandbox_policy").unwrap(),
),
token_usage_json: thread
.token_usage
.as_ref()
.map(|usage| serialize_json(usage, "token_usage").unwrap()),
history: thread.history.map(stored_thread_history_to_proto),
}
}
#[cfg(test)]
fn stored_thread_history_to_proto(history: StoredThreadHistory) -> proto::StoredThreadHistory {
proto::StoredThreadHistory {
thread_id: history.thread_id.to_string(),
items_json: rollout_items_json(&history.items).unwrap(),
}
}
fn datetime_from_unix(timestamp: i64) -> ThreadStoreResult<DateTime<Utc>> {
DateTime::from_timestamp(timestamp, 0).ok_or_else(|| ThreadStoreError::InvalidRequest {
message: format!("remote thread store returned invalid timestamp: {timestamp}"),
})
}
fn session_source_from_proto(source: &proto::SessionSource) -> ThreadStoreResult<SessionSource> {
let kind = proto::SessionSourceKind::try_from(source.kind).unwrap_or_default();
Ok(match kind {
proto::SessionSourceKind::Unknown => SessionSource::Unknown,
proto::SessionSourceKind::Cli => SessionSource::Cli,
proto::SessionSourceKind::Vscode => SessionSource::VSCode,
proto::SessionSourceKind::Exec => SessionSource::Exec,
proto::SessionSourceKind::AppServer => SessionSource::Mcp,
proto::SessionSourceKind::Custom => {
SessionSource::Custom(source.custom.clone().unwrap_or_default())
}
proto::SessionSourceKind::SubAgentReview => SessionSource::SubAgent(SubAgentSource::Review),
proto::SessionSourceKind::SubAgentCompact => {
SessionSource::SubAgent(SubAgentSource::Compact)
}
proto::SessionSourceKind::SubAgentThreadSpawn => {
let parent_thread_id = source
.sub_agent_parent_thread_id
.as_deref()
.map(ThreadId::from_string)
.transpose()
.map_err(|err| ThreadStoreError::InvalidRequest {
message: format!(
"remote thread store returned invalid sub-agent parent thread id: {err}"
),
})?
.ok_or_else(|| ThreadStoreError::InvalidRequest {
message: "remote thread store omitted sub-agent parent thread id".to_string(),
})?;
SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: source.sub_agent_depth.unwrap_or_default(),
agent_path: source
.sub_agent_path
.clone()
.map(AgentPath::from_string)
.transpose()
.map_err(|message| ThreadStoreError::InvalidRequest { message })?,
agent_nickname: source.sub_agent_nickname.clone(),
agent_role: source.sub_agent_role.clone(),
})
}
proto::SessionSourceKind::SubAgentMemoryConsolidation => {
SessionSource::SubAgent(SubAgentSource::MemoryConsolidation)
}
proto::SessionSourceKind::SubAgentOther => SessionSource::SubAgent(SubAgentSource::Other(
source.sub_agent_other.clone().unwrap_or_default(),
)),
})
}
fn git_info_from_proto(info: proto::GitInfo) -> GitInfo {
GitInfo {
commit_hash: info.sha.as_deref().map(GitSha::new),
branch: info.branch,
repository_url: info.origin_url,
}
}
#[cfg(test)]
fn git_info_to_proto(info: GitInfo) -> proto::GitInfo {
proto::GitInfo {
sha: info.commit_hash.map(|sha| sha.0),
branch: info.branch,
origin_url: info.repository_url,
}
}
fn parse_reasoning_effort(value: &str) -> ThreadStoreResult<ReasoningEffort> {
ReasoningEffort::from_str(value).map_err(|message| ThreadStoreError::InvalidRequest {
message: format!("remote thread store returned {message}"),
})
}
codex-rs/thread-store/src/remote/list_threads.rs
-282
View File
@@ -1,282 +0,0 @@
use super::RemoteThreadStore;
use super::helpers::proto_session_source;
use super::helpers::proto_sort_direction;
use super::helpers::proto_sort_key;
use super::helpers::remote_status_to_error;
use super::helpers::stored_thread_from_proto;
use super::proto;
use crate::ListThreadsParams;
use crate::ThreadPage;
use crate::ThreadStoreError;
use crate::ThreadStoreResult;
pub(super) async fn list_threads(
store: &RemoteThreadStore,
params: ListThreadsParams,
) -> ThreadStoreResult<ThreadPage> {
let request = proto::ListThreadsRequest {
page_size: params
.page_size
.try_into()
.map_err(|_| ThreadStoreError::InvalidRequest {
message: format!("page_size is too large: {}", params.page_size),
})?,
cursor: params.cursor,
sort_key: proto_sort_key(params.sort_key).into(),
sort_direction: proto_sort_direction(params.sort_direction).into(),
allowed_sources: params
.allowed_sources
.iter()
.map(proto_session_source)
.collect(),
model_provider_filter: params
.model_providers
.map(|values| proto::ModelProviderFilter { values }),
cwd_filter: params.cwd_filters.map(|values| proto::CwdFilter {
values: values
.into_iter()
.map(|cwd| cwd.display().to_string())
.collect(),
}),
archived: params.archived,
search_term: params.search_term,
use_state_db_only: params.use_state_db_only,
};
let response = store
.client()
.await?
.list_threads(request)
.await
.map_err(remote_status_to_error)?
.into_inner();
let items = response
.threads
.into_iter()
.map(stored_thread_from_proto)
.collect::<ThreadStoreResult<Vec<_>>>()?;
Ok(ThreadPage {
items,
next_cursor: response.next_cursor,
})
}
#[cfg(test)]
mod tests {
use std::path::PathBuf;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::SessionSource;
use pretty_assertions::assert_eq;
use tonic::Request;
use tonic::Response;
use tonic::Status;
use tonic::transport::Server;
use super::super::helpers::stored_thread_to_proto;
use super::super::proto::thread_store_server;
use super::super::proto::thread_store_server::ThreadStoreServer;
use super::*;
use crate::ThreadSortKey;
use crate::ThreadStore;
#[derive(Default)]
struct TestServer;
#[tonic::async_trait]
impl thread_store_server::ThreadStore for TestServer {
async fn list_threads(
&self,
request: Request<proto::ListThreadsRequest>,
) -> Result<Response<proto::ListThreadsResponse>, Status> {
let request = request.into_inner();
assert_eq!(request.page_size, 2);
assert_eq!(request.cursor.as_deref(), Some("cursor-1"));
assert_eq!(
proto::ThreadSortKey::try_from(request.sort_key),
Ok(proto::ThreadSortKey::UpdatedAt)
);
assert_eq!(
proto::SortDirection::try_from(request.sort_direction),
Ok(proto::SortDirection::Desc)
);
assert_eq!(request.archived, true);
assert_eq!(request.search_term.as_deref(), Some("needle"));
assert!(request.use_state_db_only);
assert_eq!(
request.model_provider_filter,
Some(proto::ModelProviderFilter {
values: vec!["openai".to_string()],
})
);
assert_eq!(
request.cwd_filter,
Some(proto::CwdFilter {
values: vec!["/workspace".to_string()],
})
);
assert_eq!(request.allowed_sources.len(), 1);
assert_eq!(
proto::SessionSourceKind::try_from(request.allowed_sources[0].kind),
Ok(proto::SessionSourceKind::Cli)
);
Ok(Response::new(proto::ListThreadsResponse {
threads: vec![proto::StoredThread {
thread_id: "11111111-1111-1111-1111-111111111111".to_string(),
forked_from_id: None,
preview: "hello".to_string(),
name: Some("named thread".to_string()),
model_provider: "openai".to_string(),
model: Some("gpt-5".to_string()),
created_at: 100,
updated_at: 200,
archived_at: Some(300),
cwd: "/workspace".to_string(),
cli_version: "1.2.3".to_string(),
source: Some(proto::SessionSource {
kind: proto::SessionSourceKind::Cli.into(),
..Default::default()
}),
thread_source: Some("user".to_string()),
git_info: Some(proto::GitInfo {
sha: Some("abc123".to_string()),
branch: Some("main".to_string()),
origin_url: Some("https://example.test/repo.git".to_string()),
}),
agent_nickname: None,
agent_role: None,
agent_path: None,
reasoning_effort: Some("medium".to_string()),
first_user_message: Some("hello".to_string()),
rollout_path: None,
approval_mode_json: None,
sandbox_policy_json: None,
token_usage_json: None,
history: None,
}],
next_cursor: Some("cursor-2".to_string()),
}))
}
}
#[tokio::test]
async fn list_threads_calls_remote_service() {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind test server");
let addr = listener.local_addr().expect("test server addr");
let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
let server = tokio::spawn(async move {
Server::builder()
.add_service(ThreadStoreServer::new(TestServer))
.serve_with_incoming_shutdown(
tokio_stream::wrappers::TcpListenerStream::new(listener),
async {
let _ = shutdown_rx.await;
},
)
.await
});
let store = RemoteThreadStore::new(format!("http://{addr}"));
let page = store
.list_threads(ListThreadsParams {
page_size: 2,
cursor: Some("cursor-1".to_string()),
sort_key: ThreadSortKey::UpdatedAt,
sort_direction: crate::SortDirection::Desc,
allowed_sources: vec![SessionSource::Cli],
model_providers: Some(vec!["openai".to_string()]),
cwd_filters: Some(vec![PathBuf::from("/workspace")]),
archived: true,
search_term: Some("needle".to_string()),
use_state_db_only: true,
})
.await
.expect("list threads");
assert_eq!(page.next_cursor.as_deref(), Some("cursor-2"));
assert_eq!(page.items.len(), 1);
let item = &page.items[0];
assert_eq!(
item.thread_id.to_string(),
"11111111-1111-1111-1111-111111111111"
);
assert_eq!(item.name.as_deref(), Some("named thread"));
assert_eq!(item.preview, "hello");
assert_eq!(item.first_user_message.as_deref(), Some("hello"));
assert_eq!(item.model_provider, "openai");
assert_eq!(item.model.as_deref(), Some("gpt-5"));
assert_eq!(item.created_at.timestamp(), 100);
assert_eq!(item.updated_at.timestamp(), 200);
assert_eq!(item.archived_at.map(|ts| ts.timestamp()), Some(300));
assert_eq!(item.cwd, PathBuf::from("/workspace"));
assert_eq!(item.cli_version, "1.2.3");
assert_eq!(item.source, SessionSource::Cli);
assert_eq!(item.reasoning_effort, Some(ReasoningEffort::Medium));
assert_eq!(
item.git_info.as_ref().and_then(|git| git.branch.as_deref()),
Some("main")
);
let _ = shutdown_tx.send(());
server.await.expect("join server").expect("server");
}
#[test]
fn stored_thread_proto_roundtrips_through_domain_type() {
let thread = proto::StoredThread {
thread_id: "11111111-1111-1111-1111-111111111111".to_string(),
forked_from_id: Some("22222222-2222-2222-2222-222222222222".to_string()),
preview: "preview text".to_string(),
name: Some("named thread".to_string()),
model_provider: "openai".to_string(),
model: Some("gpt-5".to_string()),
created_at: 100,
updated_at: 200,
archived_at: Some(300),
cwd: "/workspace/project".to_string(),
cli_version: "1.2.3".to_string(),
source: Some(proto::SessionSource {
kind: proto::SessionSourceKind::SubAgentThreadSpawn.into(),
sub_agent_parent_thread_id: Some(
"33333333-3333-3333-3333-333333333333".to_string(),
),
sub_agent_depth: Some(2),
sub_agent_path: Some("/root/review/backend".to_string()),
sub_agent_nickname: Some("Navigator".to_string()),
sub_agent_role: Some("explorer".to_string()),
..Default::default()
}),
thread_source: Some("subagent".to_string()),
git_info: Some(proto::GitInfo {
sha: Some("abc123".to_string()),
branch: Some("main".to_string()),
origin_url: Some("https://example.test/repo.git".to_string()),
}),
agent_nickname: Some("Navigator".to_string()),
agent_role: Some("explorer".to_string()),
agent_path: Some("/root/review/backend".to_string()),
reasoning_effort: Some("high".to_string()),
first_user_message: Some("first message".to_string()),
rollout_path: None,
approval_mode_json: None,
sandbox_policy_json: None,
token_usage_json: None,
history: None,
};
let stored = stored_thread_from_proto(thread.clone()).expect("proto to stored thread");
assert_eq!(stored.rollout_path, None);
assert!(stored.history.is_none());
let roundtripped = stored_thread_to_proto(stored);
assert_eq!(roundtripped.thread_id, thread.thread_id);
assert_eq!(roundtripped.forked_from_id, thread.forked_from_id);
assert_eq!(roundtripped.source, thread.source);
assert_eq!(roundtripped.git_info, thread.git_info);
}
}
codex-rs/thread-store/src/remote/mod.rs
-410
View File
@@ -1,410 +0,0 @@
mod helpers;
mod list_threads;
use async_trait::async_trait;
use codex_protocol::ThreadId;
use crate::AppendThreadItemsParams;
use crate::ArchiveThreadParams;
use crate::CreateThreadParams;
use crate::ListThreadsParams;
use crate::LoadThreadHistoryParams;
use crate::ReadThreadByRolloutPathParams;
use crate::ReadThreadParams;
use crate::ResumeThreadParams;
use crate::StoredThread;
use crate::StoredThreadHistory;
use crate::ThreadPage;
use crate::ThreadStore;
use crate::ThreadStoreError;
use crate::ThreadStoreResult;
use crate::UpdateThreadMetadataParams;
use proto::thread_store_client::ThreadStoreClient;
#[path = "proto/codex.thread_store.v1.rs"]
mod proto;
/// gRPC-backed [`ThreadStore`] implementation for deployments whose durable thread data lives
/// outside the app-server process.
///
/// This store is still a work in progress: app-server code should call the generic
/// [`ThreadStore`] methods, and unsupported remote operations will return explicit
/// `not_implemented` errors until the remote API catches up.
#[derive(Clone, Debug)]
pub struct RemoteThreadStore {
endpoint: String,
}
impl RemoteThreadStore {
pub fn new(endpoint: impl Into<String>) -> Self {
Self {
endpoint: endpoint.into(),
}
}
async fn client(&self) -> ThreadStoreResult<ThreadStoreClient<tonic::transport::Channel>> {
ThreadStoreClient::connect(self.endpoint.clone())
.await
.map_err(|err| ThreadStoreError::Internal {
message: format!("failed to connect to remote thread store: {err}"),
})
}
}
#[async_trait]
impl ThreadStore for RemoteThreadStore {
fn as_any(&self) -> &dyn std::any::Any {
self
}
async fn create_thread(&self, params: CreateThreadParams) -> ThreadStoreResult<()> {
let thread_id = params.thread_id;
let request = proto::CreateThreadRequest {
thread_id: thread_id.to_string(),
forked_from_id: params.forked_from_id.map(|thread_id| thread_id.to_string()),
source: Some(helpers::proto_session_source(&params.source)),
base_instructions_json: helpers::base_instructions_json(&params.base_instructions)?,
dynamic_tools_json: helpers::dynamic_tools_json(&params.dynamic_tools)?,
event_persistence_mode: helpers::proto_event_persistence_mode(
params.event_persistence_mode,
)
.into(),
metadata_json: helpers::thread_persistence_metadata_json(&params.metadata)?,
};
self.client()
.await?
.create_thread(request)
.await
.map_err(|status| helpers::remote_status_to_thread_error(status, thread_id))?;
Ok(())
}
async fn resume_thread(&self, params: ResumeThreadParams) -> ThreadStoreResult<()> {
let thread_id = params.thread_id;
let (has_history, history_json) = match params.history {
Some(history) => (true, helpers::rollout_items_json(&history)?),
None => (false, Vec::new()),
};
let request = proto::ResumeThreadRequest {
thread_id: thread_id.to_string(),
rollout_path: params
.rollout_path
.map(|path| path.to_string_lossy().into_owned()),
history_json,
has_history,
include_archived: params.include_archived,
event_persistence_mode: helpers::proto_event_persistence_mode(
params.event_persistence_mode,
)
.into(),
metadata_json: helpers::thread_persistence_metadata_json(&params.metadata)?,
};
self.client()
.await?
.resume_thread(request)
.await
.map_err(|status| helpers::remote_status_to_thread_error(status, thread_id))?;
Ok(())
}
async fn append_items(&self, params: AppendThreadItemsParams) -> ThreadStoreResult<()> {
let thread_id = params.thread_id;
let request = proto::AppendThreadItemsRequest {
thread_id: thread_id.to_string(),
items_json: helpers::rollout_items_json(&params.items)?,
};
self.client()
.await?
.append_items(request)
.await
.map_err(|status| helpers::remote_status_to_thread_error(status, thread_id))?;
Ok(())
}
async fn persist_thread(&self, thread_id: ThreadId) -> ThreadStoreResult<()> {
self.client()
.await?
.persist_thread(helpers::proto_thread_id_request(thread_id))
.await
.map_err(|status| helpers::remote_status_to_thread_error(status, thread_id))?;
Ok(())
}
async fn flush_thread(&self, thread_id: ThreadId) -> ThreadStoreResult<()> {
self.client()
.await?
.flush_thread(helpers::proto_thread_id_request(thread_id))
.await
.map_err(|status| helpers::remote_status_to_thread_error(status, thread_id))?;
Ok(())
}
async fn shutdown_thread(&self, thread_id: ThreadId) -> ThreadStoreResult<()> {
self.client()
.await?
.shutdown_thread(helpers::proto_thread_id_request(thread_id))
.await
.map_err(|status| helpers::remote_status_to_thread_error(status, thread_id))?;
Ok(())
}
async fn discard_thread(&self, thread_id: ThreadId) -> ThreadStoreResult<()> {
self.client()
.await?
.discard_thread(helpers::proto_thread_id_request(thread_id))
.await
.map_err(|status| helpers::remote_status_to_thread_error(status, thread_id))?;
Ok(())
}
async fn load_history(
&self,
params: LoadThreadHistoryParams,
) -> ThreadStoreResult<StoredThreadHistory> {
let thread_id = params.thread_id;
let response = self
.client()
.await?
.load_history(proto::LoadThreadHistoryRequest {
thread_id: thread_id.to_string(),
include_archived: params.include_archived,
})
.await
.map_err(|status| helpers::remote_status_to_thread_error(status, thread_id))?
.into_inner();
helpers::stored_thread_history_from_proto(response)
}
async fn read_thread(&self, params: ReadThreadParams) -> ThreadStoreResult<StoredThread> {
let thread_id = params.thread_id;
let response = self
.client()
.await?
.read_thread(proto::ReadThreadRequest {
thread_id: thread_id.to_string(),
include_archived: params.include_archived,
include_history: params.include_history,
})
.await
.map_err(|status| helpers::remote_status_to_thread_error(status, thread_id))?
.into_inner();
let thread = response.thread.ok_or_else(|| ThreadStoreError::Internal {
message: "remote thread store omitted read_thread response thread".to_string(),
})?;
helpers::stored_thread_from_proto(thread)
}
async fn read_thread_by_rollout_path(
&self,
_params: ReadThreadByRolloutPathParams,
) -> ThreadStoreResult<StoredThread> {
Err(ThreadStoreError::Internal {
message: "remote thread store does not support read_thread_by_rollout_path".to_string(),
})
}
async fn list_threads(&self, params: ListThreadsParams) -> ThreadStoreResult<ThreadPage> {
list_threads::list_threads(self, params).await
}
async fn update_thread_metadata(
&self,
params: UpdateThreadMetadataParams,
) -> ThreadStoreResult<StoredThread> {
let thread_id = params.thread_id;
let response = self
.client()
.await?
.update_thread_metadata(proto::UpdateThreadMetadataRequest {
thread_id: thread_id.to_string(),
patch: Some(helpers::proto_metadata_patch(params.patch)),
include_archived: params.include_archived,
})
.await
.map_err(|status| helpers::remote_status_to_thread_error(status, thread_id))?
.into_inner();
let thread = response.thread.ok_or_else(|| ThreadStoreError::Internal {
message: "remote thread store omitted update_thread_metadata response thread"
.to_string(),
})?;
helpers::stored_thread_from_proto(thread)
}
async fn archive_thread(&self, params: ArchiveThreadParams) -> ThreadStoreResult<()> {
let thread_id = params.thread_id;
self.client()
.await?
.archive_thread(proto::ArchiveThreadRequest {
thread_id: thread_id.to_string(),
})
.await
.map_err(|status| helpers::remote_status_to_thread_error(status, thread_id))?;
Ok(())
}
async fn unarchive_thread(
&self,
params: ArchiveThreadParams,
) -> ThreadStoreResult<StoredThread> {
let thread_id = params.thread_id;
let response = self
.client()
.await?
.unarchive_thread(proto::ArchiveThreadRequest {
thread_id: thread_id.to_string(),
})
.await
.map_err(|status| helpers::remote_status_to_thread_error(status, thread_id))?
.into_inner();
let thread = response.thread.ok_or_else(|| ThreadStoreError::Internal {
message: "remote thread store omitted unarchive_thread response thread".to_string(),
})?;
helpers::stored_thread_from_proto(thread)
}
}
#[cfg(test)]
mod tests {
use std::path::PathBuf;
use codex_protocol::ThreadId;
use codex_protocol::models::BaseInstructions;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadMemoryMode;
use pretty_assertions::assert_eq;
use tokio::sync::mpsc;
use tonic::Request;
use tonic::Response;
use tonic::Status;
use tonic::transport::Server;
use super::*;
use crate::ThreadEventPersistenceMode;
use crate::ThreadPersistenceMetadata;
use proto::thread_store_server;
use proto::thread_store_server::ThreadStoreServer;
enum RecordedRequest {
Create(proto::CreateThreadRequest),
Resume(proto::ResumeThreadRequest),
}
struct TestServer {
requests_tx: mpsc::UnboundedSender<RecordedRequest>,
}
#[tonic::async_trait]
impl thread_store_server::ThreadStore for TestServer {
async fn create_thread(
&self,
request: Request<proto::CreateThreadRequest>,
) -> Result<Response<proto::Empty>, Status> {
self.requests_tx
.send(RecordedRequest::Create(request.into_inner()))
.expect("record create request");
Ok(Response::new(proto::Empty {}))
}
async fn resume_thread(
&self,
request: Request<proto::ResumeThreadRequest>,
) -> Result<Response<proto::Empty>, Status> {
self.requests_tx
.send(RecordedRequest::Resume(request.into_inner()))
.expect("record resume request");
Ok(Response::new(proto::Empty {}))
}
async fn list_threads(
&self,
_request: Request<proto::ListThreadsRequest>,
) -> Result<Response<proto::ListThreadsResponse>, Status> {
Err(Status::unimplemented("not implemented"))
}
}
async fn test_store() -> (RemoteThreadStore, mpsc::UnboundedReceiver<RecordedRequest>) {
let (requests_tx, requests_rx) = mpsc::unbounded_channel();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind test server");
let addr = listener.local_addr().expect("test server addr");
tokio::spawn(async move {
Server::builder()
.add_service(ThreadStoreServer::new(TestServer { requests_tx }))
.serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
.await
.expect("test server");
});
(
RemoteThreadStore::new(format!("http://{addr}")),
requests_rx,
)
}
#[tokio::test]
async fn create_thread_forwards_metadata() {
let (store, mut requests_rx) = test_store().await;
let metadata = ThreadPersistenceMetadata {
cwd: Some(PathBuf::from("/workspace")),
model_provider: "test-provider".to_string(),
memory_mode: ThreadMemoryMode::Enabled,
};
store
.create_thread(CreateThreadParams {
thread_id: ThreadId::new(),
forked_from_id: None,
source: SessionSource::Exec,
thread_source: None,
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
metadata: metadata.clone(),
event_persistence_mode: ThreadEventPersistenceMode::Limited,
})
.await
.expect("create thread");
let Some(RecordedRequest::Create(request)) = requests_rx.recv().await else {
panic!("expected create request");
};
assert_eq!(
serde_json::from_str::<ThreadPersistenceMetadata>(&request.metadata_json)
.expect("metadata json"),
metadata
);
}
#[tokio::test]
async fn resume_thread_forwards_metadata() {
let (store, mut requests_rx) = test_store().await;
let metadata = ThreadPersistenceMetadata {
cwd: Some(PathBuf::from("/workspace")),
model_provider: "test-provider".to_string(),
memory_mode: ThreadMemoryMode::Disabled,
};
store
.resume_thread(ResumeThreadParams {
thread_id: ThreadId::new(),
rollout_path: None,
history: None,
include_archived: false,
metadata: metadata.clone(),
event_persistence_mode: ThreadEventPersistenceMode::Limited,
})
.await
.expect("resume thread");
let Some(RecordedRequest::Resume(request)) = requests_rx.recv().await else {
panic!("expected resume request");
};
assert_eq!(
serde_json::from_str::<ThreadPersistenceMetadata>(&request.metadata_json)
.expect("metadata json"),
metadata
);
}
}
codex-rs/thread-store/src/remote/proto/codex.thread_store.v1.proto
-210
View File
@@ -1,210 +0,0 @@
syntax = "proto3";
package codex.thread_store.v1;
service ThreadStore {
rpc CreateThread(CreateThreadRequest) returns (Empty);
rpc ResumeThread(ResumeThreadRequest) returns (Empty);
rpc AppendItems(AppendThreadItemsRequest) returns (Empty);
rpc PersistThread(ThreadIdRequest) returns (Empty);
rpc FlushThread(ThreadIdRequest) returns (Empty);
rpc ShutdownThread(ThreadIdRequest) returns (Empty);
rpc DiscardThread(ThreadIdRequest) returns (Empty);
rpc LoadHistory(LoadThreadHistoryRequest) returns (StoredThreadHistory);
rpc ReadThread(ReadThreadRequest) returns (StoredThreadResponse);
rpc ListThreads(ListThreadsRequest) returns (ListThreadsResponse);
rpc UpdateThreadMetadata(UpdateThreadMetadataRequest) returns (StoredThreadResponse);
rpc ArchiveThread(ArchiveThreadRequest) returns (Empty);
rpc UnarchiveThread(ArchiveThreadRequest) returns (StoredThreadResponse);
}
message Empty {}
message ThreadIdRequest {
string thread_id = 1;
}
message CreateThreadRequest {
string thread_id = 1;
optional string forked_from_id = 2;
SessionSource source = 3;
string base_instructions_json = 4;
repeated string dynamic_tools_json = 5;
ThreadEventPersistenceMode event_persistence_mode = 6;
string metadata_json = 7;
}
message ResumeThreadRequest {
string thread_id = 1;
optional string rollout_path = 2;
repeated string history_json = 3;
bool has_history = 4;
bool include_archived = 5;
ThreadEventPersistenceMode event_persistence_mode = 6;
string metadata_json = 7;
}
message AppendThreadItemsRequest {
string thread_id = 1;
repeated string items_json = 2;
}
message LoadThreadHistoryRequest {
string thread_id = 1;
bool include_archived = 2;
}
message ReadThreadRequest {
string thread_id = 1;
bool include_archived = 2;
bool include_history = 3;
}
message ListThreadsRequest {
uint32 page_size = 1;
optional string cursor = 2;
ThreadSortKey sort_key = 3;
repeated SessionSource allowed_sources = 4;
optional ModelProviderFilter model_provider_filter = 5;
bool archived = 6;
optional string search_term = 7;
optional CwdFilter cwd_filter = 8;
bool use_state_db_only = 9;
SortDirection sort_direction = 10;
}
message ModelProviderFilter {
repeated string values = 1;
}
message CwdFilter {
repeated string values = 1;
}
enum ThreadSortKey {
THREAD_SORT_KEY_CREATED_AT = 0;
THREAD_SORT_KEY_UPDATED_AT = 1;
}
enum SortDirection {
SORT_DIRECTION_ASC = 0;
SORT_DIRECTION_DESC = 1;
}
message ListThreadsResponse {
repeated StoredThread threads = 1;
optional string next_cursor = 2;
}
message StoredThreadResponse {
StoredThread thread = 1;
}
message StoredThreadHistory {
string thread_id = 1;
repeated string items_json = 2;
}
message StoredThread {
// Mirrors Rust's StoredThread. Domain types that are not protobuf-native,
// such as ThreadId, DateTime<Utc>, and PathBuf, are represented as their
// stable scalar forms on the wire.
string thread_id = 1;
optional string forked_from_id = 2;
string preview = 3;
optional string name = 4;
string model_provider = 5;
optional string model = 6;
int64 created_at = 7;
int64 updated_at = 8;
optional int64 archived_at = 9;
string cwd = 10;
string cli_version = 11;
SessionSource source = 12;
optional GitInfo git_info = 13;
optional string agent_nickname = 14;
optional string agent_role = 15;
optional string agent_path = 16;
optional string reasoning_effort = 17;
optional string first_user_message = 18;
optional string rollout_path = 19;
optional string approval_mode_json = 20;
optional string sandbox_policy_json = 21;
optional string token_usage_json = 22;
optional StoredThreadHistory history = 23;
optional string thread_source = 24;
}
message SessionSource {
SessionSourceKind kind = 1;
optional string custom = 2;
optional string sub_agent_parent_thread_id = 3;
optional int32 sub_agent_depth = 4;
optional string sub_agent_other = 5;
optional string sub_agent_path = 6;
optional string sub_agent_nickname = 7;
optional string sub_agent_role = 8;
}
enum SessionSourceKind {
SESSION_SOURCE_KIND_UNKNOWN = 0;
SESSION_SOURCE_KIND_CLI = 1;
SESSION_SOURCE_KIND_VSCODE = 2;
SESSION_SOURCE_KIND_EXEC = 3;
SESSION_SOURCE_KIND_APP_SERVER = 4;
SESSION_SOURCE_KIND_CUSTOM = 5;
SESSION_SOURCE_KIND_SUB_AGENT_REVIEW = 6;
SESSION_SOURCE_KIND_SUB_AGENT_COMPACT = 7;
SESSION_SOURCE_KIND_SUB_AGENT_THREAD_SPAWN = 8;
SESSION_SOURCE_KIND_SUB_AGENT_MEMORY_CONSOLIDATION = 9;
SESSION_SOURCE_KIND_SUB_AGENT_OTHER = 10;
}
message GitInfo {
optional string sha = 1;
optional string branch = 2;
optional string origin_url = 3;
}
message UpdateThreadMetadataRequest {
string thread_id = 1;
ThreadMetadataPatch patch = 2;
bool include_archived = 3;
}
message ThreadMetadataPatch {
optional string name = 1;
optional ThreadMemoryMode memory_mode = 2;
optional GitInfoPatch git_info = 3;
}
enum ThreadMemoryMode {
THREAD_MEMORY_MODE_ENABLED = 0;
THREAD_MEMORY_MODE_DISABLED = 1;
}
message GitInfoPatch {
OptionalStringPatch sha = 1;
OptionalStringPatch branch = 2;
OptionalStringPatch origin_url = 3;
}
message OptionalStringPatch {
OptionalStringPatchKind kind = 1;
optional string value = 2;
}
enum OptionalStringPatchKind {
OPTIONAL_STRING_PATCH_KIND_UNSET = 0;
OPTIONAL_STRING_PATCH_KIND_CLEAR = 1;
OPTIONAL_STRING_PATCH_KIND_SET = 2;
}
message ArchiveThreadRequest {
string thread_id = 1;
}
enum ThreadEventPersistenceMode {
THREAD_EVENT_PERSISTENCE_MODE_LIMITED = 0;
THREAD_EVENT_PERSISTENCE_MODE_EXTENDED = 1;
}
codex-rs/thread-store/src/remote/proto/codex.thread_store.v1.rs
-1122
View File
File diff suppressed because it is too large Load Diff