openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-28 18:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

permissions: move workspace roots onto thread state

Browse Source
This commit is contained in:
Michael Bolin
2026-05-13 16:50:23 -07:00
parent d666238b40
commit 7f45c2f625
165 changed files with 3171 additions and 3675 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/in_process.rs
+119 -74
View File
@@ -448,12 +448,12 @@ async fn start_uninitialized(args: InProcessStartArgs) -> IoResult<InProcessClie
let was_initialized = session.initialized();
processor
.process_client_request(
IN_PROCESS_CONNECTION_ID,
*request,
Arc::clone(&session),
&outbound_initialized,
)
.await;
IN_PROCESS_CONNECTION_ID,
*request,
Arc::clone(&session),
&outbound_initialized,
)
.await;
let opted_out_notification_methods_snapshot =
session.opted_out_notification_methods();
let experimental_api_enabled =
@@ -797,80 +797,125 @@ mod tests {
start_test_client_with_capacity(session_source, DEFAULT_IN_PROCESS_CHANNEL_CAPACITY).await
}
#[tokio::test]
async fn in_process_start_initializes_and_handles_typed_v2_request() {
let client = start_test_client(SessionSource::Cli).await;
let response = client
.request(ClientRequest::ConfigRequirementsRead {
request_id: RequestId::Integer(1),
params: None,
fn run_current_thread_test_with_stack<F>(name: &str, future: F) -> Result<()>
where
F: Future<Output = Result<()>> + Send + 'static,
{
const TEST_STACK_SIZE_BYTES: usize = 4 * 1024 * 1024;
let handle = std::thread::Builder::new()
.name(name.to_string())
.stack_size(TEST_STACK_SIZE_BYTES)
.spawn(move || -> Result<()> {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("current-thread runtime should build");
runtime.block_on(Box::pin(future));
})
.await
.expect("request transport should work")
.expect("request should succeed");
assert!(response.is_object());
.expect("stack-sized test thread should spawn");
let _parsed: ConfigRequirementsReadResponse =
serde_json::from_value(response).expect("response should match v2 schema");
client
.shutdown()
.await
.expect("in-process runtime should shutdown cleanly");
}
#[tokio::test]
async fn in_process_start_uses_requested_session_source_for_thread_start() {
for (requested_source, expected_source) in [
(SessionSource::Cli, ApiSessionSource::Cli),
(SessionSource::Exec, ApiSessionSource::Exec),
] {
let client = start_test_client(requested_source).await;
let response = client
.request(ClientRequest::ThreadStart {
request_id: RequestId::Integer(2),
params: ThreadStartParams {
ephemeral: Some(true),
..ThreadStartParams::default()
},
})
.await
.expect("request transport should work")
.expect("thread/start should succeed");
let parsed: ThreadStartResponse =
serde_json::from_value(response).expect("thread/start response should parse");
assert_eq!(parsed.thread.source, expected_source);
client
.shutdown()
.await
.expect("in-process runtime should shutdown cleanly");
match handle.join() {
Ok(result) => result,
Err(_) => Err(anyhow::anyhow!("{name} thread panicked")),
}
}
#[tokio::test]
async fn in_process_start_clamps_zero_channel_capacity() {
let client =
start_test_client_with_capacity(SessionSource::Cli, /*channel_capacity*/ 0).await;
let response = loop {
match client
.request(ClientRequest::ConfigRequirementsRead {
request_id: RequestId::Integer(4),
params: None,
})
.await
{
Ok(response) => break response.expect("request should succeed"),
Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
tokio::task::yield_now().await;
#[test]
fn in_process_start_initializes_and_handles_typed_v2_request() -> Result<()> {
run_current_thread_test_with_stack(
"in_process_start_initializes_and_handles_typed_v2_request",
async move {
let client = start_test_client(SessionSource::Cli).await;
let response = client
.request(ClientRequest::ConfigRequirementsRead {
request_id: RequestId::Integer(1),
params: None,
})
.await
.expect("request transport should work")
.expect("request should succeed");
assert!(response.is_object());
let _parsed: ConfigRequirementsReadResponse =
serde_json::from_value(response).expect("response should match v2 schema");
client
.shutdown()
.await
.expect("in-process runtime should shutdown cleanly");
Ok(())
},
)
}
#[test]
fn in_process_start_uses_requested_session_source_for_thread_start() -> Result<()> {
run_current_thread_test_with_stack(
"in_process_start_uses_requested_session_source_for_thread_start",
async move {
for (requested_source, expected_source) in [
(SessionSource::Cli, ApiSessionSource::Cli),
(SessionSource::Exec, ApiSessionSource::Exec),
] {
let client = start_test_client(requested_source).await;
let response = client
.request(ClientRequest::ThreadStart {
request_id: RequestId::Integer(2),
params: ThreadStartParams {
ephemeral: Some(true),
..ThreadStartParams::default()
},
})
.await
.expect("request transport should work")
.expect("thread/start should succeed");
let parsed: ThreadStartResponse = serde_json::from_value(response)
.expect("thread/start response should parse");
assert_eq!(parsed.thread.source, expected_source);
client
.shutdown()
.await
.expect("in-process runtime should shutdown cleanly");
}
Err(err) => panic!("request transport should work: {err}"),
}
};
let _parsed: ConfigRequirementsReadResponse =
serde_json::from_value(response).expect("response should match v2 schema");
client
.shutdown()
.await
.expect("in-process runtime should shutdown cleanly");
Ok(())
},
)
}
#[test]
fn in_process_start_clamps_zero_channel_capacity() -> Result<()> {
run_current_thread_test_with_stack(
"in_process_start_clamps_zero_channel_capacity",
async move {
let client = start_test_client_with_capacity(
SessionSource::Cli,
/*channel_capacity*/ 0,
)
.await;
let response = loop {
match client
.request(ClientRequest::ConfigRequirementsRead {
request_id: RequestId::Integer(4),
params: None,
})
.await
{
Ok(response) => break response.expect("request should succeed"),
Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
tokio::task::yield_now().await;
}
Err(err) => panic!("request transport should work: {err}"),
}
};
let _parsed: ConfigRequirementsReadResponse =
serde_json::from_value(response).expect("response should match v2 schema");
client
.shutdown()
.await
.expect("in-process runtime should shutdown cleanly");
Ok(())
},
)
}
#[test]
codex-rs/app-server/src/lib.rs
+3 -3
View File
@@ -580,9 +580,9 @@ pub async fn run_main_with_transport_options(
range: None,
});
}
if let Some(warning) =
codex_core::config::system_bwrap_warning(config.permissions.permission_profile.get())
{
if let Some(warning) = codex_core::config::system_bwrap_warning(
config.permissions.permission_profile_constraint().get(),
) {
config_warnings.push(ConfigWarningNotification {
summary: warning,
details: None,
codex-rs/app-server/src/message_processor.rs
+71 -67
View File
@@ -1,5 +1,6 @@
use std::collections::HashSet;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::OnceLock;
use std::sync::atomic::AtomicBool;
@@ -501,13 +502,13 @@ impl MessageProcessor {
self.account_processor.clear_external_auth();
}
pub(crate) async fn process_request(
self: &Arc<Self>,
pub(crate) fn process_request<'a>(
self: &'a Arc<Self>,
connection_id: ConnectionId,
request: JSONRPCRequest,
transport: &AppServerTransport,
transport: &'a AppServerTransport,
session: Arc<ConnectionSessionState>,
) {
) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> {
let request_method = request.method.as_str();
tracing::trace!(
?connection_id,
@@ -525,52 +526,55 @@ impl MessageProcessor {
tracestate: trace.tracestate.clone(),
});
let request_context = RequestContext::new(request_id.clone(), request_span, request_trace);
Self::run_request_with_context(
Arc::clone(&self.outgoing),
request_context.clone(),
async {
let codex_request = serde_json::to_value(&request)
.map_err(|err| invalid_request(format!("Invalid request: {err}")))
.and_then(|request_json| {
serde_json::from_value::<ClientRequest>(request_json)
.map_err(|err| invalid_request(format!("Invalid request: {err}")))
});
let result = match codex_request {
Ok(codex_request) => {
// Websocket callers finalize outbound readiness in lib.rs after mirroring
// session state into outbound state and sending initialize notifications to
// this specific connection. Passing `None` avoids marking the connection
// ready too early from inside the shared request handler.
self.handle_client_request(
request_id.clone(),
codex_request,
Arc::clone(&session),
/*outbound_initialized*/ None,
request_context.clone(),
)
.await
Box::pin(async move {
Self::run_request_with_context(
Arc::clone(&self.outgoing),
request_context.clone(),
Box::pin(async move {
let codex_request = serde_json::to_value(&request)
.map_err(|err| invalid_request(format!("Invalid request: {err}")))
.and_then(|request_json| {
serde_json::from_value::<ClientRequest>(request_json)
.map_err(|err| invalid_request(format!("Invalid request: {err}")))
});
let result = match codex_request {
Ok(codex_request) => {
// Websocket callers finalize outbound readiness in lib.rs after
// mirroring session state into outbound state and sending initialize
// notifications to this specific connection. Passing `None` avoids
// marking the connection ready too early from inside the shared
// request handler.
self.handle_client_request(
request_id.clone(),
codex_request,
Arc::clone(&session),
/*outbound_initialized*/ None,
request_context.clone(),
)
.await
}
Err(error) => Err(error),
};
if let Err(error) = result {
self.outgoing.send_error(request_id.clone(), error).await;
}
Err(error) => Err(error),
};
if let Err(error) = result {
self.outgoing.send_error(request_id.clone(), error).await;
}
},
)
.await;
}),
)
.await;
})
}
/// Handles a typed request path used by in-process embedders.
///
/// This bypasses JSON request deserialization but keeps identical request
/// semantics by delegating to `handle_client_request`.
pub(crate) async fn process_client_request(
self: &Arc<Self>,
pub(crate) fn process_client_request<'a>(
self: &'a Arc<Self>,
connection_id: ConnectionId,
request: ClientRequest,
session: Arc<ConnectionSessionState>,
outbound_initialized: &AtomicBool,
) {
outbound_initialized: &'a AtomicBool,
) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> {
let request_id = ConnectionRequestId {
connection_id,
request_id: request.id().clone(),
@@ -584,28 +588,30 @@ impl MessageProcessor {
request_id = ?request_id.request_id,
"app-server typed request"
);
Self::run_request_with_context(
Arc::clone(&self.outgoing),
request_context.clone(),
async {
// In-process clients do not have the websocket transport loop that performs
// post-initialize bookkeeping, so they still finalize outbound readiness in
// the shared request handler.
let result = self
.handle_client_request(
request_id.clone(),
request,
Arc::clone(&session),
Some(outbound_initialized),
request_context.clone(),
)
.await;
if let Err(error) = result {
self.outgoing.send_error(request_id.clone(), error).await;
}
},
)
.await;
Box::pin(async move {
Self::run_request_with_context(
Arc::clone(&self.outgoing),
request_context.clone(),
Box::pin(async move {
// In-process clients do not have the websocket transport loop that performs
// post-initialize bookkeeping, so they still finalize outbound readiness in
// the shared request handler.
let result = self
.handle_client_request(
request_id.clone(),
request,
Arc::clone(&session),
Some(outbound_initialized),
request_context.clone(),
)
.await;
if let Err(error) = result {
self.outgoing.send_error(request_id.clone(), error).await;
}
}),
)
.await;
})
}
pub(crate) async fn process_notification(&self, notification: JSONRPCNotification) {
@@ -621,13 +627,11 @@ impl MessageProcessor {
tracing::info!("<- typed notification: {:?}", notification);
}
async fn run_request_with_context<F>(
async fn run_request_with_context<'a>(
outgoing: Arc<OutgoingMessageSender>,
request_context: RequestContext,
request_fut: F,
) where
F: Future<Output = ()>,
{
request_fut: Pin<Box<dyn Future<Output = ()> + Send + 'a>>,
) {
outgoing
.register_request_context(request_context.clone())
.await;
codex-rs/app-server/src/message_processor_tracing_tests.rs
+70 -67
View File
@@ -632,77 +632,80 @@ fn thread_start_jsonrpc_span_exports_server_span_and_parents_children() -> Resul
)
}
#[tokio::test(flavor = "current_thread")]
#[test]
#[serial(app_server_tracing)]
async fn turn_start_jsonrpc_span_parents_core_turn_spans() -> Result<()> {
let mut harness = TracingHarness::new().await?;
let thread_start_response = harness.start_thread(/*request_id*/ 2, /*trace*/ None).await;
let thread_id = thread_start_response.thread.id.clone();
fn turn_start_jsonrpc_span_parents_core_turn_spans() -> Result<()> {
run_current_thread_test_with_stack("turn_start_jsonrpc_span_parents_core_turn_spans", async {
let mut harness = TracingHarness::new().await?;
let thread_start_response = harness.start_thread(/*request_id*/ 2, /*trace*/ None).await;
let thread_id = thread_start_response.thread.id.clone();
harness.reset_tracing();
harness.reset_tracing();
let RemoteTrace {
trace_id: remote_trace_id,
parent_span_id: remote_parent_span_id,
context: remote_trace,
} = RemoteTrace::new("00000000000000000000000000000077", "0000000000000088");
let turn_start_response: TurnStartResponse = harness
.request(
ClientRequest::TurnStart {
request_id: RequestId::Integer(3),
params: TurnStartParams {
environments: None,
thread_id,
input: vec![UserInput::Text {
text: "hello".to_string(),
text_elements: Vec::new(),
}],
responsesapi_client_metadata: None,
cwd: None,
approval_policy: None,
sandbox_policy: None,
permissions: None,
approvals_reviewer: None,
model: None,
service_tier: None,
effort: None,
summary: None,
personality: None,
output_schema: None,
collaboration_mode: None,
let RemoteTrace {
trace_id: remote_trace_id,
parent_span_id: remote_parent_span_id,
context: remote_trace,
} = RemoteTrace::new("00000000000000000000000000000077", "0000000000000088");
let turn_start_response: TurnStartResponse = harness
.request(
ClientRequest::TurnStart {
request_id: RequestId::Integer(3),
params: TurnStartParams {
environments: None,
thread_id,
input: vec![UserInput::Text {
text: "hello".to_string(),
text_elements: Vec::new(),
}],
responsesapi_client_metadata: None,
cwd: None,
approval_policy: None,
sandbox_policy: None,
permissions: None,
workspace_roots: None,
approvals_reviewer: None,
model: None,
service_tier: None,
effort: None,
summary: None,
personality: None,
output_schema: None,
collaboration_mode: None,
},
},
},
Some(remote_trace),
)
.await;
let spans = wait_for_exported_spans(harness.tracing, |spans| {
spans.iter().any(|span| {
span.span_kind == SpanKind::Server
&& span_attr(span, "rpc.method") == Some("turn/start")
&& span.span_context.trace_id() == remote_trace_id
}) && spans.iter().any(|span| {
span_attr(span, "codex.op") == Some("user_input")
&& span.span_context.trace_id() == remote_trace_id
Some(remote_trace),
)
.await;
let spans = wait_for_exported_spans(harness.tracing, |spans| {
spans.iter().any(|span| {
span.span_kind == SpanKind::Server
&& span_attr(span, "rpc.method") == Some("turn/start")
&& span.span_context.trace_id() == remote_trace_id
}) && spans.iter().any(|span| {
span_attr(span, "codex.op") == Some("user_input")
&& span.span_context.trace_id() == remote_trace_id
})
})
.await;
let server_request_span =
find_rpc_span_with_trace(&spans, SpanKind::Server, "turn/start", remote_trace_id);
let core_turn_span =
find_span_with_trace(&spans, remote_trace_id, "codex.op=user_input", |span| {
span_attr(span, "codex.op") == Some("user_input")
});
assert_eq!(server_request_span.parent_span_id, remote_parent_span_id);
assert!(server_request_span.parent_span_is_remote);
assert_eq!(server_request_span.span_context.trace_id(), remote_trace_id);
assert_eq!(
span_attr(server_request_span, "turn.id"),
Some(turn_start_response.turn.id.as_str())
);
assert_span_descends_from(&spans, core_turn_span, server_request_span);
harness.shutdown().await;
Ok(())
})
.await;
let server_request_span =
find_rpc_span_with_trace(&spans, SpanKind::Server, "turn/start", remote_trace_id);
let core_turn_span =
find_span_with_trace(&spans, remote_trace_id, "codex.op=user_input", |span| {
span_attr(span, "codex.op") == Some("user_input")
});
assert_eq!(server_request_span.parent_span_id, remote_parent_span_id);
assert!(server_request_span.parent_span_is_remote);
assert_eq!(server_request_span.span_context.trace_id(), remote_trace_id);
assert_eq!(
span_attr(server_request_span, "turn.id"),
Some(turn_start_response.turn.id.as_str())
);
assert_span_descends_from(&spans, core_turn_span, server_request_span);
harness.shutdown().await;
Ok(())
}
codex-rs/app-server/src/request_processors.rs
-2
View File
@@ -103,8 +103,6 @@ use codex_app_server_protocol::MockExperimentalMethodParams;
use codex_app_server_protocol::MockExperimentalMethodResponse;
use codex_app_server_protocol::ModelListParams;
use codex_app_server_protocol::ModelListResponse;
use codex_app_server_protocol::PermissionProfileModificationParams;
use codex_app_server_protocol::PermissionProfileSelectionParams;
use codex_app_server_protocol::PluginDetail;
use codex_app_server_protocol::PluginInstallParams;
use codex_app_server_protocol::PluginInstallResponse;
codex-rs/app-server/src/request_processors/command_exec_processor.rs
+7 -3
View File
@@ -164,7 +164,10 @@ impl CommandExecRequestProcessor {
let started_network_proxy = match self.config.permissions.network.as_ref() {
Some(spec) => match spec
.start_proxy(
self.config.permissions.permission_profile.get(),
self.config
.permissions
.permission_profile_constraint()
.get(),
/*policy_decider*/ None,
/*blocked_request_observer*/ None,
managed_network_requirements_enabled,
@@ -243,7 +246,7 @@ impl CommandExecRequestProcessor {
);
self.config
.permissions
.permission_profile
.permission_profile_constraint()
.can_set(&effective_permission_profile)
.map_err(|err| invalid_request(format!("invalid permission profile: {err}")))?;
effective_permission_profile
@@ -264,7 +267,7 @@ impl CommandExecRequestProcessor {
);
self.config
.permissions
.permission_profile
.permission_profile_constraint()
.can_set(&permission_profile)
.map_err(|err| invalid_request(format!("invalid sandbox policy: {err}")))?;
permission_profile
@@ -287,6 +290,7 @@ impl CommandExecRequestProcessor {
exec_params,
&effective_permission_profile,
&sandbox_cwd,
std::slice::from_ref(&sandbox_cwd),
&codex_linux_sandbox_exe,
use_legacy_landlock,
)
codex-rs/app-server/src/request_processors/thread_lifecycle.rs
+10 -5
View File
@@ -604,13 +604,18 @@ pub(super) async fn handle_pending_thread_resume_request(
permission_profile,
active_permission_profile,
cwd,
workspace_roots,
reasoning_effort,
..
} = pending.config_snapshot;
let instruction_sources = pending.instruction_sources;
let sandbox = thread_response_sandbox_policy(&permission_profile, cwd.as_path());
let active_permission_profile =
thread_response_active_permission_profile(active_permission_profile);
let sandbox = thread_response_sandbox_policy(
&permission_profile,
cwd.as_path(),
&workspace_roots,
conversation.config().await.codex_home.as_path(),
);
let permission_profile = thread_response_active_permission_profile(active_permission_profile);
let session_id = conversation.session_configured().session_id.to_string();
thread.session_id = session_id;
@@ -624,8 +629,8 @@ pub(super) async fn handle_pending_thread_resume_request(
approval_policy: approval_policy.into(),
approvals_reviewer: approvals_reviewer.into(),
sandbox,
permission_profile: Some(permission_profile.into()),
active_permission_profile,
permission_profile,
workspace_roots,
reasoning_effort,
};
outgoing.send_response(request_id, response).await;
codex-rs/app-server/src/request_processors/thread_processor.rs
+75 -4
View File
@@ -17,6 +17,25 @@ struct ThreadListFilters {
use_state_db_only: bool,
}
workspace_roots = Some(if meta_line.meta.workspace_roots.is_empty() {
cwd.iter().cloned().collect()
} else {
meta_line.meta.workspace_roots
});
let context_cwd = cwd
.as_ref()
.map(AbsolutePathBuf::as_path)
.unwrap_or(context.cwd.as_path());
permission_profile = Some(context.permission_profile.unwrap_or_else(|| {
PermissionProfile::from_legacy_sandbox_policy_for_cwd(
&context.sandbox_policy,
context_cwd,
)
}));
if context.active_permission_profile.is_some() {
active_permission_profile = context.active_permission_profile;
}
fn collect_resume_override_mismatches(
request: &ThreadResumeParams,
config_snapshot: &ThreadConfigSnapshot,
@@ -1162,6 +1181,10 @@ impl ThreadRequestProcessor {
&config_snapshot.permission_profile,
config_snapshot.cwd.as_path(),
);
let permission_profile = thread_response_permission_profile(
&config_snapshot.permission_profile,
&config_snapshot.workspace_roots,
);
let active_permission_profile =
thread_response_active_permission_profile(config_snapshot.active_permission_profile);
@@ -1175,7 +1198,8 @@ impl ThreadRequestProcessor {
approval_policy: config_snapshot.approval_policy.into(),
approvals_reviewer: config_snapshot.approvals_reviewer.into(),
sandbox,
permission_profile: Some(config_snapshot.permission_profile.into()),
permission_profile,
workspace_roots: config_snapshot.workspace_roots,
active_permission_profile,
reasoning_effort: config_snapshot.reasoning_effort,
};
@@ -1215,7 +1239,8 @@ impl ThreadRequestProcessor {
approval_policy: Option<codex_app_server_protocol::AskForApproval>,
approvals_reviewer: Option<codex_app_server_protocol::ApprovalsReviewer>,
sandbox: Option<SandboxMode>,
permissions: Option<PermissionProfileSelectionParams>,
permissions: Option<String>,
workspace_roots: Option<Vec<PathBuf>>,
base_instructions: Option<String>,
developer_instructions: Option<String>,
personality: Option<Personality>,
@@ -2379,6 +2404,8 @@ impl ThreadRequestProcessor {
};
let history_cwd = thread_history.session_cwd();
let permissions_were_explicit = permissions.is_some();
let sandbox_was_explicit = sandbox.is_some();
let mut typesafe_overrides = self.build_thread_config_overrides(
model,
model_provider,
@@ -2392,6 +2419,14 @@ impl ThreadRequestProcessor {
developer_instructions,
personality,
);
if !permissions_were_explicit
&& !sandbox_was_explicit
&& let Some(persisted_permission_state) = persisted_permission_state.as_ref()
{
typesafe_overrides.permission_profile =
Some(persisted_permission_state.permission_profile.clone());
}
.map(AbsolutePathBuf::to_path_buf)
self.load_and_apply_persisted_resume_metadata(
&thread_history,
&mut request_overrides,
@@ -2412,6 +2447,17 @@ impl ThreadRequestProcessor {
return Ok(());
}
};
if !permissions_were_explicit
&& !sandbox_was_explicit
&& let Some(persisted_permission_state) = persisted_permission_state
{
config
.permissions
.set_active_permission_profile_for_current_profile(
persisted_permission_state.active_permission_profile,
Some(&persisted_permission_state.permission_profile),
);
}
let instruction_sources = Self::instruction_sources_from_config(&config).await;
let response_history = thread_history.clone();
@@ -2507,6 +2553,10 @@ impl ThreadRequestProcessor {
&config_snapshot.permission_profile,
config_snapshot.cwd.as_path(),
);
let permission_profile = thread_response_permission_profile(
&config_snapshot.permission_profile,
&config_snapshot.workspace_roots,
);
let active_permission_profile = thread_response_active_permission_profile(
config_snapshot.active_permission_profile,
);
@@ -2525,7 +2575,8 @@ impl ThreadRequestProcessor {
approval_policy: session_configured.approval_policy.into(),
approvals_reviewer: session_configured.approvals_reviewer.into(),
sandbox,
permission_profile: Some(config_snapshot.permission_profile.into()),
permission_profile,
workspace_roots: config_snapshot.workspace_roots,
active_permission_profile,
reasoning_effort: session_configured.reasoning_effort,
};
@@ -3045,6 +3096,8 @@ impl ThreadRequestProcessor {
} else {
Some(cli_overrides)
};
let permissions_were_explicit = permissions.is_some();
let sandbox_was_explicit = sandbox.is_some();
let mut typesafe_overrides = self.build_thread_config_overrides(
model,
model_provider,
@@ -3058,6 +3111,11 @@ impl ThreadRequestProcessor {
developer_instructions,
/*personality*/ None,
);
if !permissions_were_explicit && !sandbox_was_explicit {
typesafe_overrides.permission_profile =
Some(persisted_permission_state.permission_profile.clone());
}
.map(AbsolutePathBuf::to_path_buf)
typesafe_overrides.ephemeral = ephemeral.then_some(true);
// Derive a Config using the same logic as new conversation, honoring overrides if provided.
let config = self
@@ -3065,6 +3123,14 @@ impl ThreadRequestProcessor {
.load_for_cwd(request_overrides, typesafe_overrides, history_cwd)
.await
.map_err(|err| config_load_error(&err))?;
if !permissions_were_explicit && !sandbox_was_explicit {
config
.permissions
.set_active_permission_profile_for_current_profile(
persisted_permission_state.active_permission_profile,
Some(&persisted_permission_state.permission_profile),
);
}
let fallback_model_provider = config.model_provider_id.clone();
let instruction_sources = Self::instruction_sources_from_config(&config).await;
@@ -3170,6 +3236,10 @@ impl ThreadRequestProcessor {
&config_snapshot.permission_profile,
config_snapshot.cwd.as_path(),
);
let permission_profile = thread_response_permission_profile(
&config_snapshot.permission_profile,
&config_snapshot.workspace_roots,
);
let active_permission_profile =
thread_response_active_permission_profile(config_snapshot.active_permission_profile);
@@ -3183,7 +3253,8 @@ impl ThreadRequestProcessor {
approval_policy: session_configured.approval_policy.into(),
approvals_reviewer: session_configured.approvals_reviewer.into(),
sandbox,
permission_profile: Some(config_snapshot.permission_profile.into()),
permission_profile,
workspace_roots: config_snapshot.workspace_roots,
active_permission_profile,
reasoning_effort: session_configured.reasoning_effort,
};
codex-rs/app-server/src/request_processors/thread_processor_tests.rs
+1
View File
@@ -651,6 +651,7 @@ mod thread_processor_behavior_tests {
permission_profile: codex_protocol::models::PermissionProfile::Disabled,
active_permission_profile: None,
cwd,
workspace_roots: vec![test_path_buf("/tmp").abs()],
ephemeral: false,
reasoning_effort: None,
personality: None,
codex-rs/app-server/src/request_processors/thread_summary.rs
+29 -16
View File
@@ -177,35 +177,48 @@ pub(super) fn thread_response_active_permission_profile(
pub(super) fn apply_permission_profile_selection_to_config_overrides(
overrides: &mut ConfigOverrides,
permissions: Option<PermissionProfileSelectionParams>,
permissions: Option<String>,
workspace_roots: Option<Vec<PathBuf>>,
) {
let Some(PermissionProfileSelectionParams::Profile { id, modifications }) = permissions else {
return;
};
overrides.default_permissions = Some(id);
overrides
.additional_writable_roots
.extend(modifications.unwrap_or_default().into_iter().map(
|modification| match modification {
PermissionProfileModificationParams::AdditionalWritableRoot { path } => {
path.to_path_buf()
}
},
));
if let Some(permissions) = permissions {
overrides.default_permissions = Some(permissions);
}
if let Some(workspace_roots) = workspace_roots {
overrides.workspace_roots = Some(workspace_roots);
}
}
pub(super) fn thread_response_sandbox_policy(
permission_profile: &codex_protocol::models::PermissionProfile,
cwd: &Path,
workspace_roots: &[AbsolutePathBuf],
codex_home: &Path,
) -> codex_app_server_protocol::SandboxPolicy {
let permission_profile = permission_profile
.clone()
.materialize_project_roots_with_workspace_roots(workspace_roots);
let file_system_policy = permission_profile.file_system_sandbox_policy();
let sandbox_policy = codex_sandboxing::compatibility_sandbox_policy_for_permission_profile(
permission_profile,
&permission_profile,
&file_system_policy,
permission_profile.network_sandbox_policy(),
cwd,
);
sandbox_policy.into()
let mut sandbox: codex_app_server_protocol::SandboxPolicy = sandbox_policy.into();
if let codex_app_server_protocol::SandboxPolicy::WorkspaceWrite {
legacy_writable_roots,
..
} = &mut sandbox
{
let memories_root = codex_home.join("memories");
legacy_writable_roots.extend(
workspace_roots
.iter()
.filter(|root| root.as_path() != cwd && root.as_path() != memories_root)
.cloned(),
);
}
sandbox
}
#[cfg(test)]
codex-rs/app-server/src/request_processors/turn_processor.rs
+4
View File
@@ -357,6 +357,7 @@ impl TurnRequestProcessor {
let turn_has_input = !mapped_items.is_empty();
let has_any_overrides = params.cwd.is_some()
|| params.workspace_roots.is_some()
|| params.approval_policy.is_some()
|| params.approvals_reviewer.is_some()
|| params.sandbox_policy.is_some()
@@ -375,6 +376,7 @@ impl TurnRequestProcessor {
}
let cwd = params.cwd;
let workspace_roots = params.workspace_roots;
let approval_policy = params.approval_policy.map(AskForApproval::to_core);
let approvals_reviewer = params
.approvals_reviewer
@@ -432,6 +434,7 @@ impl TurnRequestProcessor {
thread
.validate_turn_context_overrides(CodexThreadTurnContextOverrides {
cwd: cwd.clone(),
workspace_roots: workspace_roots.clone(),
approval_policy,
approvals_reviewer,
sandbox_policy: sandbox_policy.clone(),
@@ -457,6 +460,7 @@ impl TurnRequestProcessor {
final_output_json_schema: params.output_schema,
responsesapi_client_metadata: params.responsesapi_client_metadata,
cwd,
workspace_roots,
approval_policy,
approvals_reviewer,
sandbox_policy,
codex-rs/app-server/tests/common/lib.rs
+25
View File
@@ -1,3 +1,6 @@
use std::future::Future;
use anyhow::Result;
mod analytics_server;
mod auth_fixtures;
mod config;
@@ -45,6 +48,28 @@ pub use rollout::create_fake_rollout_with_token_usage;
pub use rollout::rollout_path;
use serde::de::DeserializeOwned;
pub fn run_current_thread_test_with_stack<F>(name: &str, future: F) -> Result<()>
where
F: Future<Output = Result<()>> + Send + 'static,
{
const TEST_STACK_SIZE_BYTES: usize = 4 * 1024 * 1024;
let handle = std::thread::Builder::new()
.name(name.to_string())
.stack_size(TEST_STACK_SIZE_BYTES)
.spawn(move || -> Result<()> {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
runtime.block_on(Box::pin(future))
})?;
match handle.join() {
Ok(result) => result,
Err(_) => Err(anyhow::anyhow!("{name} thread panicked")),
}
}
pub fn to_response<T: DeserializeOwned>(response: JSONRPCResponse) -> anyhow::Result<T> {
let value = serde_json::to_value(response.result)?;
let codex_response = serde_json::from_value(value)?;
codex-rs/app-server/tests/common/rollout.rs
+2
View File
@@ -142,6 +142,7 @@ pub fn create_fake_rollout_with_source(
agent_path: None,
agent_nickname: None,
agent_role: None,
workspace_roots: vec![PathBuf::from("/").try_into()?],
model_provider: model_provider.map(str::to_string),
base_instructions: None,
dynamic_tools: None,
@@ -226,6 +227,7 @@ pub fn create_fake_rollout_with_text_elements(
agent_path: None,
agent_nickname: None,
agent_role: None,
workspace_roots: vec![PathBuf::from("/").try_into()?],
model_provider: model_provider.map(str::to_string),
base_instructions: None,
dynamic_tools: None,
codex-rs/app-server/tests/suite/conversation_summary.rs
+85 -78
View File
@@ -2,6 +2,7 @@ use anyhow::Result;
use app_test_support::McpProcess;
use app_test_support::create_fake_rollout;
use app_test_support::rollout_path;
use app_test_support::run_current_thread_test_with_stack;
use app_test_support::to_response;
use codex_app_server::in_process;
use codex_app_server::in_process::InProcessStartArgs;
@@ -112,86 +113,92 @@ async fn get_conversation_summary_by_thread_id_reads_rollout() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn get_conversation_summary_by_thread_id_reads_pathless_store_thread() -> Result<()> {
let codex_home = TempDir::new()?;
let store_id = Uuid::new_v4().to_string();
create_config_toml_with_in_memory_thread_store(codex_home.path(), &store_id)?;
let store = InMemoryThreadStore::for_id(store_id.clone());
let _in_memory_store = InMemoryThreadStoreId { store_id };
let thread_id = ThreadId::from_string("00000000-0000-4000-8000-000000000125")?;
store
.create_thread(CreateThreadParams {
thread_id,
forked_from_id: None,
source: SessionSource::Cli,
thread_source: None,
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
metadata: ThreadPersistenceMetadata {
cwd: None,
model_provider: "test-provider".to_string(),
memory_mode: ThreadMemoryMode::Disabled,
},
event_persistence_mode: ThreadEventPersistenceMode::default(),
})
.await?;
#[test]
fn get_conversation_summary_by_thread_id_reads_pathless_store_thread() -> Result<()> {
run_current_thread_test_with_stack(
"get_conversation_summary_by_thread_id_reads_pathless_store_thread",
async move {
let codex_home = TempDir::new()?;
let store_id = Uuid::new_v4().to_string();
create_config_toml_with_in_memory_thread_store(codex_home.path(), &store_id)?;
let store = InMemoryThreadStore::for_id(store_id.clone());
let _in_memory_store = InMemoryThreadStoreId { store_id };
let thread_id = ThreadId::from_string("00000000-0000-4000-8000-000000000125")?;
store
.create_thread(CreateThreadParams {
thread_id,
forked_from_id: None,
source: SessionSource::Cli,
thread_source: None,
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
metadata: ThreadPersistenceMetadata {
cwd: None,
workspace_roots: Vec::new(),
model_provider: "test-provider".to_string(),
memory_mode: ThreadMemoryMode::Disabled,
},
event_persistence_mode: ThreadEventPersistenceMode::default(),
})
.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_requirements: CloudRequirementsLoader::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,
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()
}),
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_requirements: CloudRequirementsLoader::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,
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::GetConversationSummary {
request_id: RequestId::Integer(1),
params: GetConversationSummaryParams::ThreadId {
conversation_id: thread_id,
},
})
.await?
.expect("getConversationSummary should succeed");
let GetConversationSummaryResponse { summary } = serde_json::from_value(result)?;
assert_eq!(summary.conversation_id, thread_id);
assert_eq!(summary.path, PathBuf::new());
assert_eq!(summary.cwd, PathBuf::new());
assert_eq!(summary.model_provider, "test");
client.shutdown().await?;
Ok(())
},
channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY,
})
.await?;
let result = client
.request(ClientRequest::GetConversationSummary {
request_id: RequestId::Integer(1),
params: GetConversationSummaryParams::ThreadId {
conversation_id: thread_id,
},
})
.await?
.expect("getConversationSummary should succeed");
let GetConversationSummaryResponse { summary } = serde_json::from_value(result)?;
assert_eq!(summary.conversation_id, thread_id);
assert_eq!(summary.path, PathBuf::new());
assert_eq!(summary.cwd, PathBuf::new());
assert_eq!(summary.model_provider, "test");
client.shutdown().await?;
Ok(())
)
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
codex-rs/app-server/tests/suite/v2/mcp_resource.rs
+68 -60
View File
@@ -4,6 +4,7 @@ use std::time::Duration;
use anyhow::Result;
use app_test_support::ChatGptAuthFixture;
use app_test_support::McpProcess;
use app_test_support::run_current_thread_test_with_stack;
use app_test_support::to_response;
use app_test_support::write_chatgpt_auth;
use axum::Router;
@@ -183,67 +184,74 @@ apps = true
Ok(())
}
#[tokio::test]
async fn mcp_resource_read_returns_error_for_unknown_thread() -> Result<()> {
let codex_home = TempDir::new()?;
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?;
// This negative-path test does not need the stdio subprocess; keeping it
// in-process avoids child-process teardown timing in nextest leak detection.
let client = in_process::start(InProcessStartArgs {
arg0_paths: Arg0DispatchPaths::default(),
config: Arc::new(config),
cli_overrides: Vec::new(),
loader_overrides,
strict_config: false,
cloud_requirements: CloudRequirementsLoader::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,
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: None,
#[test]
fn mcp_resource_read_returns_error_for_unknown_thread() -> Result<()> {
run_current_thread_test_with_stack(
"mcp_resource_read_returns_error_for_unknown_thread",
async {
let codex_home = TempDir::new()?;
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?;
// This negative-path test does not need the stdio subprocess; keeping it
// in-process avoids child-process teardown timing in nextest leak detection.
let client = in_process::start(InProcessStartArgs {
arg0_paths: Arg0DispatchPaths::default(),
config: Arc::new(config),
cli_overrides: Vec::new(),
loader_overrides,
strict_config: false,
cloud_requirements: CloudRequirementsLoader::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,
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: None,
},
channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY,
})
.await?;
let response = client
.request(ClientRequest::McpResourceRead {
request_id: RequestId::Integer(1),
params: McpResourceReadParams {
thread_id: Some("00000000-0000-4000-8000-000000000000".to_string()),
server: "codex_apps".to_string(),
uri: TEST_RESOURCE_URI.to_string(),
},
})
.await;
client.shutdown().await?;
let error = match response? {
Ok(result) => {
anyhow::bail!("expected thread-not-found error, got response: {result:?}")
}
Err(error) => error,
};
assert!(
error.message.contains("thread not found"),
"expected thread-not-found error, got: {error:?}"
);
Ok(())
},
channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY,
})
.await?;
let response = client
.request(ClientRequest::McpResourceRead {
request_id: RequestId::Integer(1),
params: McpResourceReadParams {
thread_id: Some("00000000-0000-4000-8000-000000000000".to_string()),
server: "codex_apps".to_string(),
uri: TEST_RESOURCE_URI.to_string(),
},
})
.await;
client.shutdown().await?;
let error = match response? {
Ok(result) => anyhow::bail!("expected thread-not-found error, got response: {result:?}"),
Err(error) => error,
};
assert!(
error.message.contains("thread not found"),
"expected thread-not-found error, got: {error:?}"
);
Ok(())
)
}
async fn start_resource_apps_mcp_server() -> Result<(String, JoinHandle<()>)> {
codex-rs/app-server/tests/suite/v2/process_exec.rs
+3 -4
View File
@@ -83,8 +83,7 @@ async fn process_spawn_returns_before_exit_and_emits_exit_notification() -> Resu
.await?;
assert_eq!(response.result, serde_json::json!({}));
wait_for_file(&probe_file).await?;
assert_eq!(std::fs::read_to_string(&probe_file)?, "process");
wait_for_file_contents(&probe_file, "process").await?;
std::fs::write(&release_file, "release")?;
let exited = read_process_exited(&mut mcp).await?;
@@ -239,9 +238,9 @@ async fn read_process_exited(mcp: &mut McpProcess) -> Result<ProcessExitedNotifi
serde_json::from_value(params).context("deserialize process/exited notification")
}
async fn wait_for_file(path: &Path) -> Result<()> {
async fn wait_for_file_contents(path: &Path, expected: &str) -> Result<()> {
timeout(DEFAULT_READ_TIMEOUT, async {
while !path.exists() {
while std::fs::read_to_string(path).ok().as_deref() != Some(expected) {
sleep(Duration::from_millis(20)).await;
}
})
codex-rs/app-server/tests/suite/v2/remote_thread_store.rs
+132 -127
View File
@@ -19,6 +19,7 @@ use std::sync::Arc;
use anyhow::Result;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::run_current_thread_test_with_stack;
use codex_app_server::in_process;
use codex_app_server::in_process::InProcessServerEvent;
use codex_app_server::in_process::InProcessStartArgs;
@@ -49,137 +50,141 @@ use uuid::Uuid;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[tokio::test]
async fn thread_start_with_non_local_thread_store_does_not_create_local_persistence() -> Result<()>
{
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
let store_id = Uuid::new_v4().to_string();
// Plugin startup warmups may create `.tmp` under codex_home. Disable them
// here so this regression stays focused on thread persistence artifacts.
create_config_toml_with_thread_store(codex_home.path(), &server.uri(), &store_id)?;
#[test]
fn thread_start_with_non_local_thread_store_does_not_create_local_persistence() -> Result<()> {
run_current_thread_test_with_stack(
"thread_start_with_non_local_thread_store_does_not_create_local_persistence",
async move {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
let store_id = Uuid::new_v4().to_string();
// Plugin startup warmups may create `.tmp` under codex_home. Disable them
// here so this regression stays focused on thread persistence artifacts.
create_config_toml_with_thread_store(codex_home.path(), &server.uri(), &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 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 thread_store = InMemoryThreadStore::for_id(store_id.clone());
let _in_memory_store = InMemoryThreadStoreId { store_id };
let thread_store = InMemoryThreadStore::for_id(store_id.clone());
let _in_memory_store = InMemoryThreadStoreId { store_id };
let mut client = in_process::start(InProcessStartArgs {
arg0_paths: Arg0DispatchPaths::default(),
config: Arc::new(config),
cli_overrides: Vec::new(),
loader_overrides,
strict_config: false,
cloud_requirements: CloudRequirementsLoader::default(),
thread_config_loader: Arc::new(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,
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: None,
let mut client = in_process::start(InProcessStartArgs {
arg0_paths: Arg0DispatchPaths::default(),
config: Arc::new(config),
cli_overrides: Vec::new(),
loader_overrides,
strict_config: false,
cloud_requirements: CloudRequirementsLoader::default(),
thread_config_loader: Arc::new(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,
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: None,
},
channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY,
})
.await?;
let response = client
.request(ClientRequest::ThreadStart {
request_id: RequestId::Integer(1),
params: ThreadStartParams::default(),
})
.await?
.expect("thread/start should succeed");
let ThreadStartResponse { thread, .. } =
serde_json::from_value(response).expect("thread/start response should parse");
assert_eq!(thread.path, None);
client
.request(ClientRequest::TurnStart {
request_id: RequestId::Integer(2),
params: TurnStartParams {
thread_id: thread.id.clone(),
input: vec![V2UserInput::Text {
text: "Hello".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
},
})
.await?
.expect("turn/start should succeed");
timeout(DEFAULT_READ_TIMEOUT, async {
loop {
let Some(event) = client.next_event().await else {
anyhow::bail!("in-process app-server stopped before turn/completed");
};
if let InProcessServerEvent::ServerNotification(
ServerNotification::TurnCompleted(completed),
) = event
&& completed.thread_id == thread.id
{
return Ok::<(), anyhow::Error>(());
}
}
})
.await??;
let response = client
.request(ClientRequest::ThreadList {
request_id: RequestId::Integer(3),
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,
},
})
.await?
.expect("thread/list should succeed");
let ThreadListResponse { data, .. } =
serde_json::from_value(response).expect("thread/list response should parse");
assert_eq!(data.len(), 1);
assert_eq!(data[0].id, thread.id);
assert_eq!(data[0].path, None);
client.shutdown().await?;
let calls = thread_store.calls().await;
assert_eq!(calls.create_thread, 1);
assert_eq!(calls.list_threads, 1);
assert!(
calls.append_items > 0,
"turn/start should append rollout items through the injected store"
);
assert!(
calls.flush_thread > 0,
"turn completion should flush through the injected store"
);
assert_no_local_persistence_artifacts(codex_home.path())?;
Ok(())
},
channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY,
})
.await?;
let response = client
.request(ClientRequest::ThreadStart {
request_id: RequestId::Integer(1),
params: ThreadStartParams::default(),
})
.await?
.expect("thread/start should succeed");
let ThreadStartResponse { thread, .. } =
serde_json::from_value(response).expect("thread/start response should parse");
assert_eq!(thread.path, None);
client
.request(ClientRequest::TurnStart {
request_id: RequestId::Integer(2),
params: TurnStartParams {
thread_id: thread.id.clone(),
input: vec![V2UserInput::Text {
text: "Hello".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
},
})
.await?
.expect("turn/start should succeed");
timeout(DEFAULT_READ_TIMEOUT, async {
loop {
let Some(event) = client.next_event().await else {
anyhow::bail!("in-process app-server stopped before turn/completed");
};
if let InProcessServerEvent::ServerNotification(ServerNotification::TurnCompleted(
completed,
)) = event
&& completed.thread_id == thread.id
{
return Ok::<(), anyhow::Error>(());
}
}
})
.await??;
let response = client
.request(ClientRequest::ThreadList {
request_id: RequestId::Integer(3),
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,
},
})
.await?
.expect("thread/list should succeed");
let ThreadListResponse { data, .. } =
serde_json::from_value(response).expect("thread/list response should parse");
assert_eq!(data.len(), 1);
assert_eq!(data[0].id, thread.id);
assert_eq!(data[0].path, None);
client.shutdown().await?;
let calls = thread_store.calls().await;
assert_eq!(calls.create_thread, 1);
assert_eq!(calls.list_threads, 1);
assert!(
calls.append_items > 0,
"turn/start should append rollout items through the injected store"
);
assert!(
calls.flush_thread > 0,
"turn completion should flush through the injected store"
);
assert_no_local_persistence_artifacts(codex_home.path())?;
Ok(())
)
}
fn assert_no_local_persistence_artifacts(codex_home: &Path) -> Result<()> {
codex-rs/app-server/tests/suite/v2/skills_list.rs
+1
View File
@@ -618,6 +618,7 @@ async fn skills_changed_notification_is_emitted_after_skill_change() -> Result<(
approvals_reviewer: None,
sandbox: None,
permissions: None,
workspace_roots: None,
config: None,
service_name: None,
base_instructions: None,
codex-rs/app-server/tests/suite/v2/thread_fork.rs
+108
View File
@@ -29,6 +29,9 @@ use codex_app_server_protocol::TurnStatus;
use codex_app_server_protocol::UserInput;
use codex_config::types::AuthCredentialsStoreMode;
use codex_login::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR;
use codex_protocol::protocol::RolloutItem;
use codex_rollout::RolloutRecorder;
use codex_utils_absolute_path::AbsolutePathBuf;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
@@ -216,6 +219,111 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_fork_preserves_persisted_workspace_roots_when_request_omits_them() -> 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 repo = TempDir::new()?;
let shared = TempDir::new()?;
let workspace_roots = vec![
AbsolutePathBuf::try_from(repo.path())?,
AbsolutePathBuf::try_from(shared.path())?,
];
let request_workspace_roots = workspace_roots
.iter()
.map(AbsolutePathBuf::to_path_buf)
.collect::<Vec<_>>();
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let start_id = mcp
.send_thread_start_request(ThreadStartParams {
cwd: Some(repo.path().display().to_string()),
permissions: Some(":workspace".to_string()),
workspace_roots: Some(request_workspace_roots),
..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 rollout_path = thread.path.clone().expect("thread path");
let turn_id = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
input: vec![UserInput::Text {
text: "seed history".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(turn_id)),
)
.await??;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
for _ in 0..250 {
if let Ok((items, _, _)) = RolloutRecorder::load_rollout_items(rollout_path.as_path()).await
&& items
.iter()
.any(|item| matches!(item, RolloutItem::TurnContext(_)))
{
break;
}
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
let (items, _, _) = RolloutRecorder::load_rollout_items(rollout_path.as_path()).await?;
if !items
.iter()
.any(|item| matches!(item, RolloutItem::TurnContext(_)))
{
anyhow::bail!(
"turn context was not persisted to rollout {}",
rollout_path.display()
);
}
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: thread.id,
..Default::default()
})
.await?;
let fork_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(fork_id)),
)
.await??;
let ThreadForkResponse {
workspace_roots: forked_roots,
permission_profile,
..
} = to_response::<ThreadForkResponse>(fork_resp)?;
assert_eq!(forked_roots, workspace_roots);
assert!(
permission_profile
.as_ref()
.map(|profile| profile.id.as_str())
== Some(":workspace")
);
Ok(())
}
#[tokio::test]
async fn thread_fork_can_load_source_by_path() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
codex-rs/app-server/tests/suite/v2/thread_list.rs
+1
View File
@@ -610,6 +610,7 @@ sqlite = true
codex_home: codex_home.path().to_path_buf(),
sqlite_home: codex_home.path().to_path_buf(),
cwd: codex_home.path().to_path_buf(),
workspace_roots: vec![codex_home.path().to_path_buf().try_into()?],
model_provider_id: "mock_provider".to_string(),
generate_memories: false,
};
codex-rs/app-server/tests/suite/v2/thread_read.rs
+237 -218
View File
@@ -3,6 +3,7 @@ use app_test_support::McpProcess;
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::run_current_thread_test_with_stack;
use app_test_support::test_absolute_path;
use app_test_support::to_response;
use codex_app_server::in_process;
@@ -353,234 +354,251 @@ async fn thread_turns_list_supports_requested_items_view() -> Result<()> {
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?;
#[test]
fn thread_turns_list_reads_store_history_without_rollout_path() -> Result<()> {
run_current_thread_test_with_stack(
"thread_turns_list_reads_store_history_without_rollout_path",
async move {
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_requirements: CloudRequirementsLoader::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()
}),
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_requirements: CloudRequirementsLoader::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(())
},
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 };
#[test]
fn thread_read_loaded_include_turns_reads_store_history_without_rollout_path() -> Result<()> {
run_current_thread_test_with_stack(
"thread_read_loaded_include_turns_reads_store_history_without_rollout_path",
async move {
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_requirements: CloudRequirementsLoader::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()
}),
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_requirements: CloudRequirementsLoader::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(())
},
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?;
#[test]
fn thread_list_includes_store_thread_without_rollout_path() -> Result<()> {
run_current_thread_test_with_stack(
"thread_list_includes_store_thread_without_rollout_path",
async move {
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_requirements: CloudRequirementsLoader::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()
}),
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_requirements: CloudRequirementsLoader::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,
},
})
.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(())
},
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,
},
})
.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]
@@ -1291,6 +1309,7 @@ async fn seed_pathless_store_thread(
dynamic_tools: Vec::new(),
metadata: ThreadPersistenceMetadata {
cwd: None,
workspace_roots: Vec::new(),
model_provider: "test-provider".to_string(),
memory_mode: ThreadMemoryMode::Disabled,
},
codex-rs/app-server/tests/suite/v2/thread_unarchive.rs
+96 -89
View File
@@ -1,6 +1,7 @@
use anyhow::Result;
use app_test_support::McpProcess;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::run_current_thread_test_with_stack;
use app_test_support::to_response;
use codex_app_server::in_process;
use codex_app_server::in_process::InProcessStartArgs;
@@ -197,97 +198,103 @@ async fn thread_unarchive_moves_rollout_back_into_sessions_directory() -> Result
Ok(())
}
#[tokio::test]
async fn thread_unarchive_preserves_pathless_store_metadata() -> Result<()> {
let codex_home = TempDir::new()?;
let store_id = Uuid::new_v4().to_string();
create_config_toml_with_in_memory_thread_store(codex_home.path(), &store_id)?;
let store = InMemoryThreadStore::for_id(store_id.clone());
let _in_memory_store = InMemoryThreadStoreId { store_id };
let thread_id = ThreadId::from_string("00000000-0000-4000-8000-000000000126")?;
let parent_thread_id = ThreadId::from_string("00000000-0000-4000-8000-000000000127")?;
store
.create_thread(CreateThreadParams {
thread_id,
forked_from_id: Some(parent_thread_id),
source: SessionSource::Cli,
thread_source: None,
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
metadata: ThreadPersistenceMetadata {
cwd: None,
model_provider: "test-provider".to_string(),
memory_mode: ThreadMemoryMode::Disabled,
},
event_persistence_mode: ThreadEventPersistenceMode::default(),
})
.await?;
store
.update_thread_metadata(UpdateThreadMetadataParams {
thread_id,
patch: ThreadMetadataPatch {
name: Some(Some("named pathless thread".to_string())),
..Default::default()
},
include_archived: true,
})
.await?;
#[test]
fn thread_unarchive_preserves_pathless_store_metadata() -> Result<()> {
run_current_thread_test_with_stack(
"thread_unarchive_preserves_pathless_store_metadata",
async {
let codex_home = TempDir::new()?;
let store_id = Uuid::new_v4().to_string();
create_config_toml_with_in_memory_thread_store(codex_home.path(), &store_id)?;
let store = InMemoryThreadStore::for_id(store_id.clone());
let _in_memory_store = InMemoryThreadStoreId { store_id };
let thread_id = ThreadId::from_string("00000000-0000-4000-8000-000000000126")?;
let parent_thread_id = ThreadId::from_string("00000000-0000-4000-8000-000000000127")?;
store
.create_thread(CreateThreadParams {
thread_id,
forked_from_id: Some(parent_thread_id),
source: SessionSource::Cli,
thread_source: None,
base_instructions: BaseInstructions::default(),
dynamic_tools: Vec::new(),
metadata: ThreadPersistenceMetadata {
cwd: None,
workspace_roots: Vec::new(),
model_provider: "test-provider".to_string(),
memory_mode: ThreadMemoryMode::Disabled,
},
event_persistence_mode: ThreadEventPersistenceMode::default(),
})
.await?;
store
.update_thread_metadata(UpdateThreadMetadataParams {
thread_id,
patch: ThreadMetadataPatch {
name: Some(Some("named pathless thread".to_string())),
..Default::default()
},
include_archived: true,
})
.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_requirements: CloudRequirementsLoader::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,
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()
}),
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_requirements: CloudRequirementsLoader::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,
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::ThreadUnarchive {
request_id: RequestId::Integer(1),
params: ThreadUnarchiveParams {
thread_id: thread_id.to_string(),
},
})
.await?
.expect("thread/unarchive should succeed");
let ThreadUnarchiveResponse { thread } = serde_json::from_value(result)?;
assert_eq!(thread.id, thread_id.to_string());
assert_eq!(thread.path, None);
assert_eq!(thread.forked_from_id, Some(parent_thread_id.to_string()));
assert_eq!(thread.name, Some("named pathless thread".to_string()));
client.shutdown().await?;
Ok(())
},
channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY,
})
.await?;
let result = client
.request(ClientRequest::ThreadUnarchive {
request_id: RequestId::Integer(1),
params: ThreadUnarchiveParams {
thread_id: thread_id.to_string(),
},
})
.await?
.expect("thread/unarchive should succeed");
let ThreadUnarchiveResponse { thread } = serde_json::from_value(result)?;
assert_eq!(thread.id, thread_id.to_string());
assert_eq!(thread.path, None);
assert_eq!(thread.forked_from_id, Some(parent_thread_id.to_string()));
assert_eq!(thread.name, Some("named pathless thread".to_string()));
client.shutdown().await?;
Ok(())
)
}
fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> {
codex-rs/app-server/tests/suite/v2/turn_start.rs
+23 -9
View File
@@ -34,7 +34,6 @@ use codex_app_server_protocol::JSONRPCNotification;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::PatchApplyStatus;
use codex_app_server_protocol::PatchChangeKind;
use codex_app_server_protocol::PermissionProfileSelectionParams;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::ServerRequestResolvedNotification;
@@ -329,9 +328,26 @@ async fn turn_start_emits_thread_scoped_warning_notification_for_trimmed_skills(
let warning: WarningNotification =
serde_json::from_value(params).expect("deserialize warning notification");
assert_eq!(warning.thread_id.as_deref(), Some(thread.id.as_str()));
assert_eq!(
warning.message,
"Exceeded skills context budget of 2%. All skill descriptions were removed and 7 additional skills were not included in the model-visible skills list."
let warning_prefix =
"Exceeded skills context budget of 2%. All skill descriptions were removed and ";
let warning_suffix = " were not included in the model-visible skills list.";
assert!(
warning.message.starts_with(warning_prefix),
"unexpected warning prefix: {}",
warning.message
);
assert!(
warning.message.ends_with(warning_suffix),
"unexpected warning suffix: {}",
warning.message
);
let trimmed_count = warning.message
[WARNING_PREFIX.len()..warning.message.len() - WARNING_SUFFIX.len()]
.parse::<usize>()
.expect("warning should include a numeric trimmed skills count");
assert!(
trimmed_count > 0,
"warning should report at least one trimmed skill"
);
timeout(
@@ -779,10 +795,7 @@ async fn turn_start_rejects_invalid_permission_selection_before_starting_turn()
text: "Hello".to_string(),
text_elements: Vec::new(),
}],
permissions: Some(PermissionProfileSelectionParams::Profile {
id: ":danger-no-sandbox".to_string(),
modifications: None,
}),
permissions: Some(":danger-no-sandbox".to_string()),
..Default::default()
})
.await?;
@@ -1890,13 +1903,14 @@ async fn turn_start_updates_sandbox_and_cwd_between_turns_v2() -> Result<()> {
}],
responsesapi_client_metadata: None,
cwd: Some(first_cwd.clone()),
workspace_roots: Some(vec![first_cwd.try_into()?]),
approval_policy: Some(codex_app_server_protocol::AskForApproval::Never),
approvals_reviewer: None,
sandbox_policy: Some(codex_app_server_protocol::SandboxPolicy::WorkspaceWrite {
writable_roots: vec![first_cwd.try_into()?],
network_access: false,
exclude_tmpdir_env_var: false,
exclude_slash_tmp: false,
legacy_writable_roots: Vec::new(),
}),
permissions: None,
model: Some("mock-model".to_string()),
codex-rs/app-server/tests/suite/v2/turn_start_zsh_fork.rs
+2 -1
View File
@@ -533,12 +533,13 @@ async fn turn_start_shell_zsh_fork_subcommand_decline_marks_parent_declined_v2()
text_elements: Vec::new(),
}],
cwd: Some(workspace.clone()),
workspace_roots: Some(vec![workspace.clone()]),
approval_policy: Some(codex_app_server_protocol::AskForApproval::UnlessTrusted),
sandbox_policy: Some(codex_app_server_protocol::SandboxPolicy::WorkspaceWrite {
writable_roots: vec![workspace.clone().try_into()?],
network_access: false,
exclude_tmpdir_env_var: false,
exclude_slash_tmp: false,
legacy_writable_roots: Vec::new(),
}),
model: Some("mock-model".to_string()),
effort: Some(codex_protocol::openai_models::ReasoningEffort::Medium),