mirror of
https://github.com/pchuan98/codex.git
synced 2026-07-01 00:31:56 +08:00
jif
8f02973d25
## Why
`selectedCapabilityRoots` is durable thread intent: “use this capability
root from environment `worker`.”
The important product assumption is:
> One environment ID always names the same logical executor and stable
contents.
`worker` does not silently change from executor A to an unrelated
executor B. The process-local connection handle for `worker` can still
be replaced while Codex is running, though, for example when
`environment/add` registers a fresh handle for the same logical
environment.
The thread should persist only the stable selection. Each model step
should pair that selection with the exact ready handle captured for that
step.
## The boundary
```text
persisted thread intent
plugin@1 -> environment "worker"
|
| capture the current step
v
model-step view
unavailable, or
plugin@1 + worker's exact captured ready handle
```
The environment ID is the stable identity and cache key. The
`Arc<Environment>` is only a process-local handle retained so consumers
of one model step use the same captured environment. It is never
persisted and it does not imply different environment contents.
## What changes
### Persist the stable selection
Selected roots are written into `SessionMeta` and restored with the
thread. Forked subagents inherit the same selections, including
bounded-history forks.
Only stable data is persisted: root ID, environment ID, and root path.
### Capture readiness together with the exact handle
The environment snapshot records:
```rust
environment_id -> Some(Arc<Environment>) // ready in this step
environment_id -> None // still starting in this step
```
This prevents readiness and execution from coming from different
registry snapshots.
For example:
```text
step snapshot: worker -> handle A, ready
environment/add: worker -> fresh handle B for the same logical environment
current step: plugin@1 still uses captured handle A
```
Without carrying handle A in the snapshot, the resolver could combine “A
was ready” with handle B and treat B as ready before it had finished
starting.
This does not change cache invalidation. Stable capability metadata
remains identified by environment ID and capability root. Replacing a
process-local handle under the same stable environment ID does not
invalidate or rediscover that metadata.
### Resolve availability per model step
- A ready captured environment produces resolved roots using its
captured handle.
- A starting, missing, or failed environment is omitted from that step.
- A selected lazy environment that is outside the turn's captured
environment set is asked to start, and a later step can observe it as
ready.
- No capability files are scanned here.
Transient transport disconnects remain the remote client's reconnect
concern. This PR models initial attachment/readiness; it does not add
live socket-connectivity state.
## Example
```text
thread selection: plugin@1 -> environment "worker"
step 1: worker is starting -> plugin@1 unavailable
step 2: worker is ready -> plugin@1 resolves through worker's captured handle
step 3: fresh local handle -> current step remains pinned; a later step captures its own view
```
Temporary unavailability does not discard the durable selection. Later
PRs can retain stable metadata caches while projecting only currently
available capabilities into model-visible World State.
## Compatibility
The app-server request shape does not change. Older rollouts without
`selected_capability_roots` deserialize to an empty list.
## Stack
1. **This PR:** persist stable selected roots and resolve them through
an exact model-step handle.
2. #29960: cache stable skill metadata and project available skills into
World State.
3. #29946: cache stable plugin declarations and manage the separate live
MCP runtime.
1461 lines
50 KiB
Rust
1461 lines
50 KiB
Rust
use anyhow::Result;
|
|
use app_test_support::TestAppServer;
|
|
use app_test_support::create_fake_rollout_with_text_elements;
|
|
use app_test_support::create_mock_responses_server_repeating_assistant;
|
|
use app_test_support::rollout_path;
|
|
use app_test_support::test_absolute_path;
|
|
use app_test_support::to_response;
|
|
use codex_app_server::in_process;
|
|
use codex_app_server::in_process::InProcessStartArgs;
|
|
use codex_app_server_protocol::ClientInfo;
|
|
use codex_app_server_protocol::ClientRequest;
|
|
use codex_app_server_protocol::InitializeCapabilities;
|
|
use codex_app_server_protocol::InitializeParams;
|
|
use codex_app_server_protocol::JSONRPCError;
|
|
use codex_app_server_protocol::JSONRPCResponse;
|
|
use codex_app_server_protocol::RequestId;
|
|
use codex_app_server_protocol::SessionSource;
|
|
use codex_app_server_protocol::SortDirection;
|
|
use codex_app_server_protocol::ThreadForkParams;
|
|
use codex_app_server_protocol::ThreadForkResponse;
|
|
use codex_app_server_protocol::ThreadItem;
|
|
use codex_app_server_protocol::ThreadItemsListParams;
|
|
use codex_app_server_protocol::ThreadListParams;
|
|
use codex_app_server_protocol::ThreadListResponse;
|
|
use codex_app_server_protocol::ThreadNameUpdatedNotification;
|
|
use codex_app_server_protocol::ThreadReadParams;
|
|
use codex_app_server_protocol::ThreadReadResponse;
|
|
use codex_app_server_protocol::ThreadResumeInitialTurnsPageParams;
|
|
use codex_app_server_protocol::ThreadResumeParams;
|
|
use codex_app_server_protocol::ThreadResumeResponse;
|
|
use codex_app_server_protocol::ThreadSetNameParams;
|
|
use codex_app_server_protocol::ThreadSetNameResponse;
|
|
use codex_app_server_protocol::ThreadStartParams;
|
|
use codex_app_server_protocol::ThreadStartResponse;
|
|
use codex_app_server_protocol::ThreadStatus;
|
|
use codex_app_server_protocol::ThreadTurnsListParams;
|
|
use codex_app_server_protocol::ThreadTurnsListResponse;
|
|
use codex_app_server_protocol::TurnItemsView;
|
|
use codex_app_server_protocol::TurnStartParams;
|
|
use codex_app_server_protocol::TurnStartResponse;
|
|
use codex_app_server_protocol::TurnStatus;
|
|
use codex_app_server_protocol::UserInput;
|
|
use codex_arg0::Arg0DispatchPaths;
|
|
use codex_config::CloudConfigBundleLoader;
|
|
use codex_config::LoaderOverrides;
|
|
use codex_core::ARCHIVED_SESSIONS_SUBDIR;
|
|
use codex_core::config::ConfigBuilder;
|
|
use codex_exec_server::EnvironmentManager;
|
|
use codex_feedback::CodexFeedback;
|
|
use codex_protocol::models::BaseInstructions;
|
|
use codex_protocol::protocol::AgentMessageEvent;
|
|
use codex_protocol::protocol::EventMsg;
|
|
use codex_protocol::protocol::RolloutItem;
|
|
use codex_protocol::protocol::SessionSource as ProtocolSessionSource;
|
|
use codex_protocol::protocol::ThreadMemoryMode;
|
|
use codex_protocol::protocol::UserMessageEvent;
|
|
use codex_protocol::user_input::ByteRange;
|
|
use codex_protocol::user_input::TextElement;
|
|
use codex_thread_store::AppendThreadItemsParams;
|
|
use codex_thread_store::CreateThreadParams;
|
|
use codex_thread_store::InMemoryThreadStore;
|
|
use codex_thread_store::ThreadMetadataPatch;
|
|
use codex_thread_store::ThreadPersistenceMetadata;
|
|
use codex_thread_store::ThreadStore;
|
|
use codex_thread_store::UpdateThreadMetadataParams;
|
|
use core_test_support::responses;
|
|
use pretty_assertions::assert_eq;
|
|
use serde_json::Value;
|
|
use serde_json::json;
|
|
use std::io::Write;
|
|
use std::path::Path;
|
|
use std::sync::Arc;
|
|
use tempfile::TempDir;
|
|
use tokio::time::timeout;
|
|
use uuid::Uuid;
|
|
|
|
#[cfg(windows)]
|
|
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25);
|
|
#[cfg(not(windows))]
|
|
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
|
|
|
|
#[tokio::test]
|
|
async fn thread_read_returns_summary_without_turns() -> Result<()> {
|
|
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let preview = "Saved user message";
|
|
let text_elements = [TextElement::new(
|
|
ByteRange { start: 0, end: 5 },
|
|
Some("<note>".into()),
|
|
)];
|
|
let conversation_id = create_fake_rollout_with_text_elements(
|
|
codex_home.path(),
|
|
"2025-01-05T12-00-00",
|
|
"2025-01-05T12:00:00Z",
|
|
preview,
|
|
text_elements
|
|
.iter()
|
|
.map(|elem| serde_json::to_value(elem).expect("serialize text element"))
|
|
.collect(),
|
|
Some("mock_provider"),
|
|
/*git_info*/ None,
|
|
)?;
|
|
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
|
|
|
let read_id = mcp
|
|
.send_thread_read_request(ThreadReadParams {
|
|
thread_id: conversation_id.clone(),
|
|
include_turns: false,
|
|
})
|
|
.await?;
|
|
let read_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
let ThreadReadResponse { thread, .. } = to_response::<ThreadReadResponse>(read_resp)?;
|
|
|
|
assert_eq!(thread.id, conversation_id);
|
|
assert_eq!(thread.preview, preview);
|
|
assert_eq!(thread.model_provider, "mock_provider");
|
|
assert!(!thread.ephemeral, "stored rollouts should not be ephemeral");
|
|
assert!(thread.path.as_ref().expect("thread path").is_absolute());
|
|
assert_eq!(thread.cwd, test_absolute_path("/"));
|
|
assert_eq!(thread.cli_version, "0.0.0");
|
|
assert_eq!(thread.source, SessionSource::Cli);
|
|
assert_eq!(thread.git_info, None);
|
|
assert_eq!(thread.turns.len(), 0);
|
|
assert_eq!(thread.status, ThreadStatus::NotLoaded);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_read_can_include_turns() -> Result<()> {
|
|
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let preview = "Saved user message";
|
|
let text_elements = vec![TextElement::new(
|
|
ByteRange { start: 0, end: 5 },
|
|
Some("<note>".into()),
|
|
)];
|
|
let conversation_id = create_fake_rollout_with_text_elements(
|
|
codex_home.path(),
|
|
"2025-01-05T12-00-00",
|
|
"2025-01-05T12:00:00Z",
|
|
preview,
|
|
text_elements
|
|
.iter()
|
|
.map(|elem| serde_json::to_value(elem).expect("serialize text element"))
|
|
.collect(),
|
|
Some("mock_provider"),
|
|
/*git_info*/ None,
|
|
)?;
|
|
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
|
|
|
let read_id = mcp
|
|
.send_thread_read_request(ThreadReadParams {
|
|
thread_id: conversation_id.clone(),
|
|
include_turns: true,
|
|
})
|
|
.await?;
|
|
let read_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
let ThreadReadResponse { thread, .. } = to_response::<ThreadReadResponse>(read_resp)?;
|
|
|
|
assert_eq!(thread.turns.len(), 1);
|
|
let turn = &thread.turns[0];
|
|
assert_eq!(turn.status, TurnStatus::Completed);
|
|
assert_eq!(turn.items_view, TurnItemsView::Full);
|
|
assert_eq!(turn.items.len(), 1, "expected user message item");
|
|
match &turn.items[0] {
|
|
ThreadItem::UserMessage { content, .. } => {
|
|
assert_eq!(
|
|
content,
|
|
&vec![UserInput::Text {
|
|
text: preview.to_string(),
|
|
text_elements: text_elements.clone().into_iter().map(Into::into).collect(),
|
|
}]
|
|
);
|
|
}
|
|
other => panic!("expected user message item, got {other:?}"),
|
|
}
|
|
assert_eq!(thread.status, ThreadStatus::NotLoaded);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_turns_list_can_page_backward_and_forward() -> Result<()> {
|
|
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let filename_ts = "2025-01-05T12-00-00";
|
|
let conversation_id = create_fake_rollout_with_text_elements(
|
|
codex_home.path(),
|
|
filename_ts,
|
|
"2025-01-05T12:00:00Z",
|
|
"first",
|
|
vec![],
|
|
Some("mock_provider"),
|
|
/*git_info*/ None,
|
|
)?;
|
|
let rollout_path = rollout_path(codex_home.path(), filename_ts, &conversation_id);
|
|
append_user_message(rollout_path.as_path(), "2025-01-05T12:01:00Z", "second")?;
|
|
append_user_message(rollout_path.as_path(), "2025-01-05T12:02:00Z", "third")?;
|
|
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
|
|
|
let read_id = mcp
|
|
.send_thread_turns_list_request(ThreadTurnsListParams {
|
|
thread_id: conversation_id.clone(),
|
|
cursor: None,
|
|
limit: Some(2),
|
|
sort_direction: Some(SortDirection::Desc),
|
|
items_view: None,
|
|
})
|
|
.await?;
|
|
let read_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
let ThreadTurnsListResponse {
|
|
data,
|
|
next_cursor,
|
|
backwards_cursor,
|
|
} = to_response::<ThreadTurnsListResponse>(read_resp)?;
|
|
assert_eq!(turn_user_texts(&data), vec!["third", "second"]);
|
|
assert!(
|
|
data.iter()
|
|
.all(|turn| turn.items_view == TurnItemsView::Summary)
|
|
);
|
|
let next_cursor = next_cursor.expect("expected nextCursor for older turns");
|
|
let backwards_cursor = backwards_cursor.expect("expected backwardsCursor for newest turn");
|
|
|
|
let read_id = mcp
|
|
.send_thread_turns_list_request(ThreadTurnsListParams {
|
|
thread_id: conversation_id.clone(),
|
|
cursor: Some(next_cursor),
|
|
limit: Some(10),
|
|
sort_direction: Some(SortDirection::Desc),
|
|
items_view: None,
|
|
})
|
|
.await?;
|
|
let read_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
let ThreadTurnsListResponse { data, .. } = to_response::<ThreadTurnsListResponse>(read_resp)?;
|
|
assert_eq!(turn_user_texts(&data), vec!["first"]);
|
|
|
|
append_user_message(rollout_path.as_path(), "2025-01-05T12:03:00Z", "fourth")?;
|
|
|
|
let read_id = mcp
|
|
.send_thread_turns_list_request(ThreadTurnsListParams {
|
|
thread_id: conversation_id,
|
|
cursor: Some(backwards_cursor),
|
|
limit: Some(10),
|
|
sort_direction: Some(SortDirection::Asc),
|
|
items_view: None,
|
|
})
|
|
.await?;
|
|
let read_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
let ThreadTurnsListResponse { data, .. } = to_response::<ThreadTurnsListResponse>(read_resp)?;
|
|
assert_eq!(turn_user_texts(&data), vec!["third", "fourth"]);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_turns_list_supports_requested_items_view() -> Result<()> {
|
|
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let filename_ts = "2025-01-05T12-00-00";
|
|
let conversation_id = create_fake_rollout_with_text_elements(
|
|
codex_home.path(),
|
|
filename_ts,
|
|
"2025-01-05T12:00:00Z",
|
|
"first",
|
|
vec![],
|
|
Some("mock_provider"),
|
|
/*git_info*/ None,
|
|
)?;
|
|
let rollout_path = rollout_path(codex_home.path(), filename_ts, &conversation_id);
|
|
append_agent_message(rollout_path.as_path(), "2025-01-05T12:01:00Z", "draft")?;
|
|
append_agent_message(rollout_path.as_path(), "2025-01-05T12:02:00Z", "final")?;
|
|
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
|
|
|
let full = read_single_turn_items_view(
|
|
&mut mcp,
|
|
conversation_id.as_str(),
|
|
Some(TurnItemsView::Full),
|
|
)
|
|
.await?;
|
|
assert_eq!(full.items_view, TurnItemsView::Full);
|
|
assert_eq!(
|
|
turn_agent_texts(std::slice::from_ref(&full)),
|
|
vec!["draft", "final"]
|
|
);
|
|
|
|
let summary = read_single_turn_items_view(
|
|
&mut mcp,
|
|
conversation_id.as_str(),
|
|
Some(TurnItemsView::Summary),
|
|
)
|
|
.await?;
|
|
assert_eq!(summary.items_view, TurnItemsView::Summary);
|
|
assert_eq!(
|
|
turn_user_texts(std::slice::from_ref(&summary)),
|
|
vec!["first"]
|
|
);
|
|
assert_eq!(
|
|
turn_agent_texts(std::slice::from_ref(&summary)),
|
|
vec!["final"]
|
|
);
|
|
|
|
let not_loaded = read_single_turn_items_view(
|
|
&mut mcp,
|
|
conversation_id.as_str(),
|
|
Some(TurnItemsView::NotLoaded),
|
|
)
|
|
.await?;
|
|
assert_eq!(not_loaded.items_view, TurnItemsView::NotLoaded);
|
|
assert!(not_loaded.items.is_empty());
|
|
assert_eq!(not_loaded.id, full.id);
|
|
assert_eq!(not_loaded.status, full.status);
|
|
assert_eq!(not_loaded.started_at, full.started_at);
|
|
assert_eq!(not_loaded.completed_at, full.completed_at);
|
|
assert_eq!(not_loaded.duration_ms, full.duration_ms);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_turns_list_reads_store_history_without_rollout_path() -> Result<()> {
|
|
let codex_home = TempDir::new()?;
|
|
let thread_id = codex_protocol::ThreadId::from_string("00000000-0000-4000-8000-000000000123")?;
|
|
let store_id = Uuid::new_v4().to_string();
|
|
create_config_toml_with_thread_store(codex_home.path(), &store_id)?;
|
|
let store = InMemoryThreadStore::for_id(store_id.clone());
|
|
let _in_memory_store = InMemoryThreadStoreId { store_id };
|
|
seed_pathless_store_thread(&store, thread_id).await?;
|
|
|
|
let loader_overrides = LoaderOverrides::without_managed_config_for_tests();
|
|
let config = ConfigBuilder::default()
|
|
.codex_home(codex_home.path().to_path_buf())
|
|
.fallback_cwd(Some(codex_home.path().to_path_buf()))
|
|
.loader_overrides(loader_overrides.clone())
|
|
.build()
|
|
.await?;
|
|
let client = in_process::start(InProcessStartArgs {
|
|
arg0_paths: Arg0DispatchPaths::default(),
|
|
config: Arc::new(config),
|
|
cli_overrides: Vec::new(),
|
|
loader_overrides,
|
|
strict_config: false,
|
|
cloud_config_bundle: CloudConfigBundleLoader::default(),
|
|
thread_config_loader: Arc::new(codex_config::NoopThreadConfigLoader),
|
|
feedback: CodexFeedback::new(),
|
|
log_db: None,
|
|
state_db: None,
|
|
environment_manager: Arc::new(EnvironmentManager::default_for_tests()),
|
|
config_warnings: Vec::new(),
|
|
session_source: SessionSource::Cli.into(),
|
|
enable_codex_api_key_env: false,
|
|
initialize: InitializeParams {
|
|
client_info: ClientInfo {
|
|
name: "codex-app-server-tests".to_string(),
|
|
title: None,
|
|
version: "0.1.0".to_string(),
|
|
},
|
|
capabilities: Some(InitializeCapabilities {
|
|
experimental_api: true,
|
|
..Default::default()
|
|
}),
|
|
},
|
|
channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY,
|
|
})
|
|
.await?;
|
|
|
|
let result = client
|
|
.request(ClientRequest::ThreadTurnsList {
|
|
request_id: RequestId::Integer(1),
|
|
params: ThreadTurnsListParams {
|
|
thread_id: thread_id.to_string(),
|
|
cursor: None,
|
|
limit: Some(10),
|
|
sort_direction: Some(SortDirection::Asc),
|
|
items_view: None,
|
|
},
|
|
})
|
|
.await?
|
|
.expect("thread/turns/list should succeed");
|
|
let ThreadTurnsListResponse { data, .. } = serde_json::from_value(result)?;
|
|
|
|
assert_eq!(turn_user_texts(&data), vec!["history from store"]);
|
|
|
|
client.shutdown().await?;
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_read_loaded_include_turns_reads_store_history_without_rollout_path() -> Result<()> {
|
|
let codex_home = TempDir::new()?;
|
|
let store_id = Uuid::new_v4().to_string();
|
|
create_config_toml_with_thread_store(codex_home.path(), &store_id)?;
|
|
let store = InMemoryThreadStore::for_id(store_id.clone());
|
|
let _in_memory_store = InMemoryThreadStoreId { store_id };
|
|
|
|
let loader_overrides = LoaderOverrides::without_managed_config_for_tests();
|
|
let config = ConfigBuilder::default()
|
|
.codex_home(codex_home.path().to_path_buf())
|
|
.fallback_cwd(Some(codex_home.path().to_path_buf()))
|
|
.loader_overrides(loader_overrides.clone())
|
|
.build()
|
|
.await?;
|
|
let client = in_process::start(InProcessStartArgs {
|
|
arg0_paths: Arg0DispatchPaths::default(),
|
|
config: Arc::new(config),
|
|
cli_overrides: Vec::new(),
|
|
loader_overrides,
|
|
strict_config: false,
|
|
cloud_config_bundle: CloudConfigBundleLoader::default(),
|
|
thread_config_loader: Arc::new(codex_config::NoopThreadConfigLoader),
|
|
feedback: CodexFeedback::new(),
|
|
log_db: None,
|
|
state_db: None,
|
|
environment_manager: Arc::new(EnvironmentManager::default_for_tests()),
|
|
config_warnings: Vec::new(),
|
|
session_source: SessionSource::Cli.into(),
|
|
enable_codex_api_key_env: false,
|
|
initialize: InitializeParams {
|
|
client_info: ClientInfo {
|
|
name: "codex-app-server-tests".to_string(),
|
|
title: None,
|
|
version: "0.1.0".to_string(),
|
|
},
|
|
capabilities: Some(InitializeCapabilities {
|
|
experimental_api: true,
|
|
..Default::default()
|
|
}),
|
|
},
|
|
channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY,
|
|
})
|
|
.await?;
|
|
|
|
let result = client
|
|
.request(ClientRequest::ThreadStart {
|
|
request_id: RequestId::Integer(1),
|
|
params: ThreadStartParams {
|
|
model: Some("mock-model".to_string()),
|
|
..Default::default()
|
|
},
|
|
})
|
|
.await?
|
|
.expect("thread/start should succeed");
|
|
let ThreadStartResponse { thread, .. } = serde_json::from_value(result)?;
|
|
assert_eq!(thread.path, None);
|
|
|
|
let thread_id = codex_protocol::ThreadId::from_string(&thread.id)?;
|
|
store
|
|
.append_items(AppendThreadItemsParams {
|
|
thread_id,
|
|
items: store_history_items(),
|
|
})
|
|
.await?;
|
|
|
|
let result = client
|
|
.request(ClientRequest::ThreadRead {
|
|
request_id: RequestId::Integer(2),
|
|
params: ThreadReadParams {
|
|
thread_id: thread.id,
|
|
include_turns: true,
|
|
},
|
|
})
|
|
.await?
|
|
.expect("thread/read should succeed");
|
|
let ThreadReadResponse { thread, .. } = serde_json::from_value(result)?;
|
|
|
|
assert_eq!(turn_user_texts(&thread.turns), vec!["history from store"]);
|
|
|
|
client.shutdown().await?;
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_list_includes_store_thread_without_rollout_path() -> Result<()> {
|
|
let codex_home = TempDir::new()?;
|
|
let thread_id = codex_protocol::ThreadId::from_string("00000000-0000-4000-8000-000000000124")?;
|
|
let store_id = Uuid::new_v4().to_string();
|
|
create_config_toml_with_thread_store(codex_home.path(), &store_id)?;
|
|
let store = InMemoryThreadStore::for_id(store_id.clone());
|
|
let _in_memory_store = InMemoryThreadStoreId { store_id };
|
|
seed_pathless_store_thread(&store, thread_id).await?;
|
|
|
|
let loader_overrides = LoaderOverrides::without_managed_config_for_tests();
|
|
let config = ConfigBuilder::default()
|
|
.codex_home(codex_home.path().to_path_buf())
|
|
.fallback_cwd(Some(codex_home.path().to_path_buf()))
|
|
.loader_overrides(loader_overrides.clone())
|
|
.build()
|
|
.await?;
|
|
let client = in_process::start(InProcessStartArgs {
|
|
arg0_paths: Arg0DispatchPaths::default(),
|
|
config: Arc::new(config),
|
|
cli_overrides: Vec::new(),
|
|
loader_overrides,
|
|
strict_config: false,
|
|
cloud_config_bundle: CloudConfigBundleLoader::default(),
|
|
thread_config_loader: Arc::new(codex_config::NoopThreadConfigLoader),
|
|
feedback: CodexFeedback::new(),
|
|
log_db: None,
|
|
state_db: None,
|
|
environment_manager: Arc::new(EnvironmentManager::default_for_tests()),
|
|
config_warnings: Vec::new(),
|
|
session_source: SessionSource::Cli.into(),
|
|
enable_codex_api_key_env: false,
|
|
initialize: InitializeParams {
|
|
client_info: ClientInfo {
|
|
name: "codex-app-server-tests".to_string(),
|
|
title: None,
|
|
version: "0.1.0".to_string(),
|
|
},
|
|
capabilities: Some(InitializeCapabilities {
|
|
experimental_api: true,
|
|
..Default::default()
|
|
}),
|
|
},
|
|
channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY,
|
|
})
|
|
.await?;
|
|
|
|
let result = client
|
|
.request(ClientRequest::ThreadList {
|
|
request_id: RequestId::Integer(1),
|
|
params: ThreadListParams {
|
|
cursor: None,
|
|
limit: Some(10),
|
|
sort_key: None,
|
|
sort_direction: None,
|
|
model_providers: Some(Vec::new()),
|
|
source_kinds: None,
|
|
archived: None,
|
|
cwd: None,
|
|
use_state_db_only: false,
|
|
search_term: None,
|
|
parent_thread_id: None,
|
|
ancestor_thread_id: None,
|
|
},
|
|
})
|
|
.await?
|
|
.expect("thread/list should succeed");
|
|
let ThreadListResponse { data, .. } = serde_json::from_value(result)?;
|
|
|
|
assert_eq!(data.len(), 1);
|
|
let thread = &data[0];
|
|
assert_eq!(thread.id, thread_id.to_string());
|
|
assert_eq!(thread.path, None);
|
|
assert_eq!(thread.preview, "");
|
|
assert_eq!(thread.name.as_deref(), Some("named pathless thread"));
|
|
|
|
client.shutdown().await?;
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_read_can_return_archived_threads_by_id() -> Result<()> {
|
|
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let filename_ts = "2025-01-05T12-00-00";
|
|
let preview = "Archived saved user message";
|
|
let conversation_id = create_fake_rollout_with_text_elements(
|
|
codex_home.path(),
|
|
filename_ts,
|
|
"2025-01-05T12:00:00Z",
|
|
preview,
|
|
vec![],
|
|
Some("mock_provider"),
|
|
/*git_info*/ None,
|
|
)?;
|
|
let active_rollout_path = rollout_path(codex_home.path(), filename_ts, &conversation_id);
|
|
let archived_dir = codex_home.path().join(ARCHIVED_SESSIONS_SUBDIR);
|
|
std::fs::create_dir_all(&archived_dir)?;
|
|
let archived_rollout_path =
|
|
archived_dir.join(active_rollout_path.file_name().expect("rollout file name"));
|
|
std::fs::rename(&active_rollout_path, &archived_rollout_path)?;
|
|
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
|
|
|
let read_id = mcp
|
|
.send_thread_read_request(ThreadReadParams {
|
|
thread_id: conversation_id.clone(),
|
|
include_turns: false,
|
|
})
|
|
.await?;
|
|
let read_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
let ThreadReadResponse { thread } = to_response::<ThreadReadResponse>(read_resp)?;
|
|
|
|
assert_eq!(thread.id, conversation_id);
|
|
assert_eq!(thread.preview, preview);
|
|
let path = thread.path.expect("thread path");
|
|
assert_eq!(path.canonicalize()?, archived_rollout_path.canonicalize()?);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_resume_initial_turns_page_matches_requested_turns_list_page() -> Result<()> {
|
|
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let filename_ts = "2025-01-05T12-00-00";
|
|
let conversation_id = create_fake_rollout_with_text_elements(
|
|
codex_home.path(),
|
|
filename_ts,
|
|
"2025-01-05T12:00:00Z",
|
|
"first",
|
|
vec![],
|
|
Some("mock_provider"),
|
|
/*git_info*/ None,
|
|
)?;
|
|
let rollout_path = rollout_path(codex_home.path(), filename_ts, &conversation_id);
|
|
append_user_message(rollout_path.as_path(), "2025-01-05T12:01:00Z", "second")?;
|
|
append_user_message(rollout_path.as_path(), "2025-01-05T12:02:00Z", "third")?;
|
|
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
|
|
|
let turns_list_id = mcp
|
|
.send_thread_turns_list_request(ThreadTurnsListParams {
|
|
thread_id: conversation_id.clone(),
|
|
cursor: None,
|
|
limit: Some(2),
|
|
sort_direction: Some(SortDirection::Asc),
|
|
items_view: Some(TurnItemsView::NotLoaded),
|
|
})
|
|
.await?;
|
|
let turns_list_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(turns_list_id)),
|
|
)
|
|
.await??;
|
|
let expected_page = to_response::<ThreadTurnsListResponse>(turns_list_resp)?;
|
|
|
|
let resume_id = mcp
|
|
.send_thread_resume_request(ThreadResumeParams {
|
|
thread_id: conversation_id,
|
|
exclude_turns: true,
|
|
initial_turns_page: Some(ThreadResumeInitialTurnsPageParams {
|
|
limit: Some(2),
|
|
sort_direction: Some(SortDirection::Asc),
|
|
items_view: Some(TurnItemsView::NotLoaded),
|
|
}),
|
|
..Default::default()
|
|
})
|
|
.await?;
|
|
let resume_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(resume_id)),
|
|
)
|
|
.await??;
|
|
let ThreadResumeResponse {
|
|
thread,
|
|
initial_turns_page,
|
|
..
|
|
} = to_response::<ThreadResumeResponse>(resume_resp)?;
|
|
|
|
assert!(thread.turns.is_empty());
|
|
assert_eq!(
|
|
initial_turns_page,
|
|
Some(codex_app_server_protocol::TurnsPage::from(expected_page))
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_turns_list_rejects_cursor_when_anchor_turn_is_rolled_back() -> Result<()> {
|
|
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let filename_ts = "2025-01-05T12-00-00";
|
|
let conversation_id = create_fake_rollout_with_text_elements(
|
|
codex_home.path(),
|
|
filename_ts,
|
|
"2025-01-05T12:00:00Z",
|
|
"first",
|
|
vec![],
|
|
Some("mock_provider"),
|
|
/*git_info*/ None,
|
|
)?;
|
|
let rollout_path = rollout_path(codex_home.path(), filename_ts, &conversation_id);
|
|
append_user_message(rollout_path.as_path(), "2025-01-05T12:01:00Z", "second")?;
|
|
append_user_message(rollout_path.as_path(), "2025-01-05T12:02:00Z", "third")?;
|
|
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
|
|
|
let read_id = mcp
|
|
.send_thread_turns_list_request(ThreadTurnsListParams {
|
|
thread_id: conversation_id.clone(),
|
|
cursor: None,
|
|
limit: Some(2),
|
|
sort_direction: Some(SortDirection::Desc),
|
|
items_view: None,
|
|
})
|
|
.await?;
|
|
let read_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
let ThreadTurnsListResponse {
|
|
backwards_cursor, ..
|
|
} = to_response::<ThreadTurnsListResponse>(read_resp)?;
|
|
let backwards_cursor = backwards_cursor.expect("expected backwardsCursor for newest turn");
|
|
|
|
append_thread_rollback(
|
|
rollout_path.as_path(),
|
|
"2025-01-05T12:03:00Z",
|
|
/*num_turns*/ 1,
|
|
)?;
|
|
|
|
let read_id = mcp
|
|
.send_thread_turns_list_request(ThreadTurnsListParams {
|
|
thread_id: conversation_id,
|
|
cursor: Some(backwards_cursor),
|
|
limit: Some(10),
|
|
sort_direction: Some(SortDirection::Asc),
|
|
items_view: None,
|
|
})
|
|
.await?;
|
|
let read_err: JSONRPCError = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_error_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
|
|
assert_eq!(
|
|
read_err.error.message,
|
|
"invalid cursor: anchor turn is no longer present"
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_read_returns_forked_from_id_for_forked_threads() -> Result<()> {
|
|
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let conversation_id = create_fake_rollout_with_text_elements(
|
|
codex_home.path(),
|
|
"2025-01-05T12-00-00",
|
|
"2025-01-05T12:00:00Z",
|
|
"Saved user message",
|
|
vec![],
|
|
Some("mock_provider"),
|
|
/*git_info*/ None,
|
|
)?;
|
|
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
|
|
|
let fork_id = mcp
|
|
.send_thread_fork_request(ThreadForkParams {
|
|
thread_id: conversation_id.clone(),
|
|
..Default::default()
|
|
})
|
|
.await?;
|
|
let fork_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(fork_id)),
|
|
)
|
|
.await??;
|
|
let ThreadForkResponse { thread: forked, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
|
|
|
|
let read_id = mcp
|
|
.send_thread_read_request(ThreadReadParams {
|
|
thread_id: forked.id,
|
|
include_turns: false,
|
|
})
|
|
.await?;
|
|
let read_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
let ThreadReadResponse { thread, .. } = to_response::<ThreadReadResponse>(read_resp)?;
|
|
|
|
assert_eq!(thread.forked_from_id, Some(conversation_id));
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_read_loaded_thread_returns_precomputed_path_before_materialization() -> Result<()> {
|
|
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let mut mcp = TestAppServer::new_with_auto_env(codex_home.path()).await?;
|
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
|
|
|
let start_id = mcp
|
|
.send_thread_start_request_with_auto_env(ThreadStartParams {
|
|
model: Some("mock-model".to_string()),
|
|
..Default::default()
|
|
})
|
|
.await?;
|
|
let start_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(start_id)),
|
|
)
|
|
.await??;
|
|
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
|
|
let thread_path = thread.path.clone().expect("thread path");
|
|
assert!(
|
|
!thread_path.exists(),
|
|
"fresh thread rollout should not be materialized yet"
|
|
);
|
|
|
|
let read_id = mcp
|
|
.send_thread_read_request(ThreadReadParams {
|
|
thread_id: thread.id.clone(),
|
|
include_turns: false,
|
|
})
|
|
.await?;
|
|
let read_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
let ThreadReadResponse { thread: read, .. } = to_response::<ThreadReadResponse>(read_resp)?;
|
|
|
|
assert_eq!(read.id, thread.id);
|
|
assert_eq!(read.path, Some(thread_path));
|
|
assert!(read.preview.is_empty());
|
|
assert_eq!(read.turns.len(), 0);
|
|
assert_eq!(read.status, ThreadStatus::Idle);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> {
|
|
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let preview = "Saved user message";
|
|
let conversation_id = create_fake_rollout_with_text_elements(
|
|
codex_home.path(),
|
|
"2025-01-05T12-00-00",
|
|
"2025-01-05T12:00:00Z",
|
|
preview,
|
|
vec![],
|
|
Some("mock_provider"),
|
|
/*git_info*/ None,
|
|
)?;
|
|
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
|
|
|
// Set a user-facing thread title.
|
|
let new_name = "My renamed thread";
|
|
let set_id = mcp
|
|
.send_thread_set_name_request(ThreadSetNameParams {
|
|
thread_id: conversation_id.clone(),
|
|
name: new_name.to_string(),
|
|
})
|
|
.await?;
|
|
let set_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(set_id)),
|
|
)
|
|
.await??;
|
|
let _: ThreadSetNameResponse = to_response::<ThreadSetNameResponse>(set_resp)?;
|
|
let notification = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_notification_message("thread/name/updated"),
|
|
)
|
|
.await??;
|
|
let notification: ThreadNameUpdatedNotification =
|
|
serde_json::from_value(notification.params.expect("thread/name/updated params"))?;
|
|
assert_eq!(notification.thread_id, conversation_id);
|
|
assert_eq!(notification.thread_name.as_deref(), Some(new_name));
|
|
|
|
// Read should now surface `thread.name`, and the wire payload must include `name`.
|
|
let read_id = mcp
|
|
.send_thread_read_request(ThreadReadParams {
|
|
thread_id: conversation_id.clone(),
|
|
include_turns: false,
|
|
})
|
|
.await?;
|
|
let read_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
let read_result = read_resp.result.clone();
|
|
let ThreadReadResponse { thread, .. } = to_response::<ThreadReadResponse>(read_resp)?;
|
|
assert_eq!(thread.id, conversation_id);
|
|
assert_eq!(thread.name.as_deref(), Some(new_name));
|
|
let thread_json = read_result
|
|
.get("thread")
|
|
.and_then(Value::as_object)
|
|
.expect("thread/read result.thread must be an object");
|
|
assert_eq!(
|
|
thread_json.get("name").and_then(Value::as_str),
|
|
Some(new_name),
|
|
"thread/read must serialize `thread.name` on the wire"
|
|
);
|
|
assert_eq!(
|
|
thread_json.get("ephemeral").and_then(Value::as_bool),
|
|
Some(false),
|
|
"thread/read must serialize `thread.ephemeral` on the wire"
|
|
);
|
|
|
|
// List should also surface the name.
|
|
let list_id = mcp
|
|
.send_thread_list_request(ThreadListParams {
|
|
cursor: None,
|
|
limit: Some(50),
|
|
sort_key: None,
|
|
sort_direction: None,
|
|
model_providers: Some(vec!["mock_provider".to_string()]),
|
|
source_kinds: None,
|
|
archived: None,
|
|
cwd: None,
|
|
use_state_db_only: false,
|
|
search_term: None,
|
|
parent_thread_id: None,
|
|
ancestor_thread_id: None,
|
|
})
|
|
.await?;
|
|
let list_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(list_id)),
|
|
)
|
|
.await??;
|
|
let list_result = list_resp.result.clone();
|
|
let ThreadListResponse { data, .. } = to_response::<ThreadListResponse>(list_resp)?;
|
|
let listed = data
|
|
.iter()
|
|
.find(|t| t.id == conversation_id)
|
|
.expect("thread/list should include the created thread");
|
|
assert_eq!(listed.name.as_deref(), Some(new_name));
|
|
let listed_json = list_result
|
|
.get("data")
|
|
.and_then(Value::as_array)
|
|
.expect("thread/list result.data must be an array")
|
|
.iter()
|
|
.find(|t| t.get("id").and_then(Value::as_str) == Some(&conversation_id))
|
|
.and_then(Value::as_object)
|
|
.expect("thread/list should include the created thread as an object");
|
|
assert_eq!(
|
|
listed_json.get("name").and_then(Value::as_str),
|
|
Some(new_name),
|
|
"thread/list must serialize `thread.name` on the wire"
|
|
);
|
|
assert_eq!(
|
|
listed_json.get("ephemeral").and_then(Value::as_bool),
|
|
Some(false),
|
|
"thread/list must serialize `thread.ephemeral` on the wire"
|
|
);
|
|
|
|
// Resume should also surface the name.
|
|
let resume_id = mcp
|
|
.send_thread_resume_request(ThreadResumeParams {
|
|
thread_id: conversation_id.clone(),
|
|
..Default::default()
|
|
})
|
|
.await?;
|
|
let resume_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(resume_id)),
|
|
)
|
|
.await??;
|
|
let resume_result = resume_resp.result.clone();
|
|
let ThreadResumeResponse {
|
|
thread: resumed, ..
|
|
} = to_response::<ThreadResumeResponse>(resume_resp)?;
|
|
assert_eq!(resumed.id, conversation_id);
|
|
assert_eq!(resumed.name.as_deref(), Some(new_name));
|
|
let resumed_json = resume_result
|
|
.get("thread")
|
|
.and_then(Value::as_object)
|
|
.expect("thread/resume result.thread must be an object");
|
|
assert_eq!(
|
|
resumed_json.get("name").and_then(Value::as_str),
|
|
Some(new_name),
|
|
"thread/resume must serialize `thread.name` on the wire"
|
|
);
|
|
assert_eq!(
|
|
resumed_json.get("ephemeral").and_then(Value::as_bool),
|
|
Some(false),
|
|
"thread/resume must serialize `thread.ephemeral` on the wire"
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_read_include_turns_rejects_unmaterialized_loaded_thread() -> Result<()> {
|
|
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let mut mcp = TestAppServer::new_with_auto_env(codex_home.path()).await?;
|
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
|
|
|
let start_id = mcp
|
|
.send_thread_start_request_with_auto_env(ThreadStartParams {
|
|
model: Some("mock-model".to_string()),
|
|
..Default::default()
|
|
})
|
|
.await?;
|
|
let start_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(start_id)),
|
|
)
|
|
.await??;
|
|
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
|
|
let thread_path = thread.path.clone().expect("thread path");
|
|
assert!(
|
|
!thread_path.exists(),
|
|
"fresh thread rollout should not be materialized yet"
|
|
);
|
|
|
|
let read_id = mcp
|
|
.send_thread_read_request(ThreadReadParams {
|
|
thread_id: thread.id.clone(),
|
|
include_turns: true,
|
|
})
|
|
.await?;
|
|
let read_err: JSONRPCError = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_error_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
|
|
assert!(
|
|
read_err
|
|
.error
|
|
.message
|
|
.contains("includeTurns is unavailable before first user message"),
|
|
"unexpected error: {}",
|
|
read_err.error.message
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_turns_list_rejects_unmaterialized_loaded_thread() -> Result<()> {
|
|
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let mut mcp = TestAppServer::new_with_auto_env(codex_home.path()).await?;
|
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
|
|
|
let start_id = mcp
|
|
.send_thread_start_request_with_auto_env(ThreadStartParams {
|
|
model: Some("mock-model".to_string()),
|
|
..Default::default()
|
|
})
|
|
.await?;
|
|
let start_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(start_id)),
|
|
)
|
|
.await??;
|
|
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
|
|
let thread_path = thread.path.clone().expect("thread path");
|
|
assert!(
|
|
!thread_path.exists(),
|
|
"fresh thread rollout should not be materialized yet"
|
|
);
|
|
|
|
let read_id = mcp
|
|
.send_thread_turns_list_request(ThreadTurnsListParams {
|
|
thread_id: thread.id,
|
|
cursor: None,
|
|
limit: None,
|
|
sort_direction: None,
|
|
items_view: None,
|
|
})
|
|
.await?;
|
|
let read_err: JSONRPCError = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_error_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
|
|
assert!(
|
|
read_err
|
|
.error
|
|
.message
|
|
.contains("thread/turns/list is unavailable before first user message"),
|
|
"unexpected error: {}",
|
|
read_err.error.message
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_items_list_returns_unsupported() -> Result<()> {
|
|
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
|
|
|
let read_id = mcp
|
|
.send_thread_items_list_request(ThreadItemsListParams {
|
|
thread_id: "00000000-0000-4000-8000-000000000123".to_string(),
|
|
turn_id: None,
|
|
cursor: None,
|
|
limit: None,
|
|
sort_direction: None,
|
|
})
|
|
.await?;
|
|
let read_err: JSONRPCError = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_error_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
|
|
assert_eq!(read_err.error.code, -32601);
|
|
assert_eq!(
|
|
read_err.error.message,
|
|
"thread/items/list is not supported yet"
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn thread_read_reports_system_error_idle_flag_after_failed_turn() -> Result<()> {
|
|
let server = responses::start_mock_server().await;
|
|
let _response_mock = responses::mount_sse_once(
|
|
&server,
|
|
responses::sse_failed("resp-1", "server_error", "simulated failure"),
|
|
)
|
|
.await;
|
|
let codex_home = TempDir::new()?;
|
|
create_config_toml(codex_home.path(), &server.uri())?;
|
|
|
|
let mut mcp = TestAppServer::new_with_auto_env(codex_home.path()).await?;
|
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
|
|
|
let start_id = mcp
|
|
.send_thread_start_request_with_auto_env(ThreadStartParams {
|
|
model: Some("mock-model".to_string()),
|
|
..Default::default()
|
|
})
|
|
.await?;
|
|
let start_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(start_id)),
|
|
)
|
|
.await??;
|
|
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
|
|
|
|
let turn_start_id = mcp
|
|
.send_turn_start_request(TurnStartParams {
|
|
thread_id: thread.id.clone(),
|
|
client_user_message_id: None,
|
|
input: vec![UserInput::Text {
|
|
text: "fail this turn".to_string(),
|
|
text_elements: Vec::new(),
|
|
}],
|
|
..Default::default()
|
|
})
|
|
.await?;
|
|
let turn_start_response: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(turn_start_id)),
|
|
)
|
|
.await??;
|
|
let _: TurnStartResponse = to_response::<TurnStartResponse>(turn_start_response)?;
|
|
timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_notification_message("error"),
|
|
)
|
|
.await??;
|
|
|
|
let read_id = mcp
|
|
.send_thread_read_request(ThreadReadParams {
|
|
thread_id: thread.id,
|
|
include_turns: false,
|
|
})
|
|
.await?;
|
|
let read_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
let ThreadReadResponse { thread, .. } = to_response::<ThreadReadResponse>(read_resp)?;
|
|
|
|
assert_eq!(thread.status, ThreadStatus::SystemError,);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
fn append_user_message(path: &Path, timestamp: &str, text: &str) -> std::io::Result<()> {
|
|
let mut file = std::fs::OpenOptions::new().append(true).open(path)?;
|
|
writeln!(
|
|
file,
|
|
"{}",
|
|
json!({
|
|
"timestamp": timestamp,
|
|
"type":"event_msg",
|
|
"payload": {
|
|
"type":"user_message",
|
|
"message": text,
|
|
"text_elements": [],
|
|
"local_images": []
|
|
}
|
|
})
|
|
)
|
|
}
|
|
|
|
fn append_agent_message(path: &Path, timestamp: &str, text: &str) -> anyhow::Result<()> {
|
|
let mut file = std::fs::OpenOptions::new().append(true).open(path)?;
|
|
writeln!(
|
|
file,
|
|
"{}",
|
|
json!({
|
|
"timestamp": timestamp,
|
|
"type": "event_msg",
|
|
"payload": serde_json::to_value(EventMsg::AgentMessage(AgentMessageEvent {
|
|
message: text.to_string(),
|
|
phase: None,
|
|
memory_citation: None,
|
|
}))?,
|
|
})
|
|
)?;
|
|
Ok(())
|
|
}
|
|
|
|
fn append_thread_rollback(path: &Path, timestamp: &str, num_turns: u32) -> std::io::Result<()> {
|
|
let mut file = std::fs::OpenOptions::new().append(true).open(path)?;
|
|
writeln!(
|
|
file,
|
|
"{}",
|
|
json!({
|
|
"timestamp": timestamp,
|
|
"type":"event_msg",
|
|
"payload": {
|
|
"type":"thread_rolled_back",
|
|
"num_turns": num_turns
|
|
}
|
|
})
|
|
)
|
|
}
|
|
|
|
async fn read_single_turn_items_view(
|
|
mcp: &mut TestAppServer,
|
|
thread_id: &str,
|
|
items_view: Option<TurnItemsView>,
|
|
) -> anyhow::Result<codex_app_server_protocol::Turn> {
|
|
let read_id = mcp
|
|
.send_thread_turns_list_request(ThreadTurnsListParams {
|
|
thread_id: thread_id.to_string(),
|
|
cursor: None,
|
|
limit: Some(10),
|
|
sort_direction: Some(SortDirection::Asc),
|
|
items_view,
|
|
})
|
|
.await?;
|
|
let read_resp: JSONRPCResponse = timeout(
|
|
DEFAULT_READ_TIMEOUT,
|
|
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
|
|
)
|
|
.await??;
|
|
let ThreadTurnsListResponse { mut data, .. } =
|
|
to_response::<ThreadTurnsListResponse>(read_resp)?;
|
|
assert_eq!(data.len(), 1);
|
|
Ok(data.remove(0))
|
|
}
|
|
|
|
fn turn_user_texts(turns: &[codex_app_server_protocol::Turn]) -> Vec<&str> {
|
|
turns
|
|
.iter()
|
|
.filter_map(|turn| match turn.items.first()? {
|
|
ThreadItem::UserMessage { content, .. } => match content.first()? {
|
|
UserInput::Text { text, .. } => Some(text.as_str()),
|
|
UserInput::Image { .. }
|
|
| UserInput::LocalImage { .. }
|
|
| UserInput::Skill { .. }
|
|
| UserInput::Mention { .. } => None,
|
|
},
|
|
_ => None,
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
fn turn_agent_texts(turns: &[codex_app_server_protocol::Turn]) -> Vec<&str> {
|
|
turns
|
|
.iter()
|
|
.flat_map(|turn| &turn.items)
|
|
.filter_map(|item| match item {
|
|
ThreadItem::AgentMessage { text, .. } => Some(text.as_str()),
|
|
_ => None,
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
struct InMemoryThreadStoreId {
|
|
store_id: String,
|
|
}
|
|
|
|
impl Drop for InMemoryThreadStoreId {
|
|
fn drop(&mut self) {
|
|
InMemoryThreadStore::remove_id(&self.store_id);
|
|
}
|
|
}
|
|
|
|
async fn seed_pathless_store_thread(
|
|
store: &InMemoryThreadStore,
|
|
thread_id: codex_protocol::ThreadId,
|
|
) -> Result<()> {
|
|
store
|
|
.create_thread(CreateThreadParams {
|
|
session_id: thread_id.into(),
|
|
thread_id,
|
|
extra_config: None,
|
|
forked_from_id: None,
|
|
parent_thread_id: None,
|
|
source: ProtocolSessionSource::Cli,
|
|
thread_source: None,
|
|
originator: "test_originator".to_string(),
|
|
base_instructions: BaseInstructions::default(),
|
|
dynamic_tools: Vec::new(),
|
|
selected_capability_roots: Vec::new(),
|
|
multi_agent_version: None,
|
|
initial_window_id: Uuid::now_v7().to_string(),
|
|
metadata: ThreadPersistenceMetadata {
|
|
cwd: None,
|
|
model_provider: "test-provider".to_string(),
|
|
memory_mode: ThreadMemoryMode::Disabled,
|
|
},
|
|
})
|
|
.await?;
|
|
store
|
|
.append_items(AppendThreadItemsParams {
|
|
thread_id,
|
|
items: store_history_items(),
|
|
})
|
|
.await?;
|
|
store
|
|
.update_thread_metadata(UpdateThreadMetadataParams {
|
|
thread_id,
|
|
patch: ThreadMetadataPatch {
|
|
name: Some(Some("named pathless thread".to_string())),
|
|
..Default::default()
|
|
},
|
|
include_archived: true,
|
|
})
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
fn store_history_items() -> Vec<RolloutItem> {
|
|
vec![RolloutItem::EventMsg(EventMsg::UserMessage(
|
|
UserMessageEvent {
|
|
client_id: None,
|
|
message: "history from store".to_string(),
|
|
images: None,
|
|
local_images: Vec::new(),
|
|
text_elements: Vec::new(),
|
|
..Default::default()
|
|
},
|
|
))]
|
|
}
|
|
|
|
fn create_config_toml_with_thread_store(codex_home: &Path, store_id: &str) -> std::io::Result<()> {
|
|
let config_toml = codex_home.join("config.toml");
|
|
std::fs::write(
|
|
config_toml,
|
|
format!(
|
|
r#"
|
|
model = "mock-model"
|
|
approval_policy = "never"
|
|
sandbox_mode = "read-only"
|
|
experimental_thread_store = {{ type = "in_memory", id = "{store_id}" }}
|
|
|
|
model_provider = "mock_provider"
|
|
|
|
[model_providers.mock_provider]
|
|
name = "Mock provider for test"
|
|
base_url = "http://127.0.0.1:1/v1"
|
|
wire_api = "responses"
|
|
request_max_retries = 0
|
|
stream_max_retries = 0
|
|
"#
|
|
),
|
|
)
|
|
}
|
|
|
|
// Helper to create a config.toml pointing at the mock model server.
|
|
fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> {
|
|
let config_toml = codex_home.join("config.toml");
|
|
std::fs::write(
|
|
config_toml,
|
|
format!(
|
|
r#"
|
|
model = "mock-model"
|
|
approval_policy = "never"
|
|
sandbox_mode = "read-only"
|
|
|
|
model_provider = "mock_provider"
|
|
|
|
[model_providers.mock_provider]
|
|
name = "Mock provider for test"
|
|
base_url = "{server_uri}/v1"
|
|
wire_api = "responses"
|
|
request_max_retries = 0
|
|
stream_max_retries = 0
|
|
"#
|
|
),
|
|
)
|
|
}
|