mirror of
https://github.com/openai/codex.git
synced 2026-10-18 17:18:52 +00:00
Enable parallel execution for read-only tools (#52245)
## Why Tools that do not opt into parallel execution acquire an exclusive scheduler lock, serializing independent reads. ## What changed Allow skill listing and reading, memory listing, reading and searching, conversation history search and reading, plugin listing, agent listing, current time, remaining context, and `get_goal` to run in parallel. Keep goal creation and updates serialized. ## Testing Add `cloud_skill_reads_run_in_parallel`, which uses a barrier to require two cloud skill reads to overlap and verifies that each returns the expected resource contents. GitOrigin-RevId: 97b5634f1f84be2ba092be89e6e9c634ca75c07b
This commit is contained in:
committed by
copyberry
parent
845345b1dd
commit
c9e0fffbbc
@@ -94,6 +94,10 @@ impl<'call> ToolExecutor<ToolCall<'call>> for HistoryTool {
|
||||
ToolName::namespaced("user_message", &self.function.name)
|
||||
}
|
||||
|
||||
fn supports_parallel_tool_calls(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn spec(&self) -> ToolSpec {
|
||||
ToolSpec::Namespace(ResponsesApiNamespace {
|
||||
name: "user_message".to_owned(),
|
||||
|
||||
@@ -53,6 +53,10 @@ impl ToolExecutor<ToolInvocation> for CurrentTimeHandler {
|
||||
ToolName::namespaced(NAMESPACE, TOOL_NAME)
|
||||
}
|
||||
|
||||
fn supports_parallel_tool_calls(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn spec(&self) -> ToolSpec {
|
||||
ToolSpec::Namespace(ResponsesApiNamespace {
|
||||
name: NAMESPACE.to_string(),
|
||||
|
||||
@@ -63,6 +63,10 @@ impl ToolExecutor<ToolInvocation> for GetContextRemainingHandler {
|
||||
ToolName::plain(GET_CONTEXT_REMAINING_TOOL_NAME)
|
||||
}
|
||||
|
||||
fn supports_parallel_tool_calls(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn spec(&self) -> ToolSpec {
|
||||
create_get_context_remaining_tool()
|
||||
}
|
||||
|
||||
@@ -64,7 +64,7 @@ impl ToolExecutor<ToolInvocation> for ListAvailablePluginsToInstallHandler {
|
||||
}
|
||||
|
||||
fn supports_parallel_tool_calls(&self) -> bool {
|
||||
false
|
||||
true
|
||||
}
|
||||
|
||||
fn handle<'a>(&'a self, invocation: ToolInvocation) -> codex_tools::ToolExecutorFuture<'a>
|
||||
@@ -118,13 +118,6 @@ mod tests {
|
||||
use codex_tools::DiscoverableToolType;
|
||||
use pretty_assertions::assert_eq;
|
||||
|
||||
#[test]
|
||||
fn list_tool_does_not_support_parallel_calls() {
|
||||
assert!(
|
||||
!ListAvailablePluginsToInstallHandler::new(Vec::new()).supports_parallel_tool_calls()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn result_truncates_candidate_descriptions() {
|
||||
let handler = ListAvailablePluginsToInstallHandler::new(vec![
|
||||
|
||||
@@ -10,6 +10,10 @@ impl ToolExecutor<ToolInvocation> for Handler {
|
||||
ToolName::plain("list_agents")
|
||||
}
|
||||
|
||||
fn supports_parallel_tool_calls(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn spec(&self) -> ToolSpec {
|
||||
create_list_agents_tool()
|
||||
}
|
||||
|
||||
@@ -11,6 +11,7 @@ struct FakeCloudSkillProvider {
|
||||
catalog: SkillCatalog,
|
||||
resources: std::collections::HashMap<String, String>,
|
||||
reads: Mutex<Vec<String>>,
|
||||
read_barrier: Option<tokio::sync::Barrier>,
|
||||
}
|
||||
|
||||
impl SkillProvider for FakeCloudSkillProvider {
|
||||
@@ -42,6 +43,11 @@ impl SkillProvider for FakeCloudSkillProvider {
|
||||
.get(request.resource.as_str())
|
||||
.cloned()
|
||||
.ok_or_else(|| SkillProviderError::new("unknown fake skill resource"))?;
|
||||
if let Some(barrier) = &self.read_barrier {
|
||||
tokio::time::timeout(Duration::from_secs(/*secs*/ 10), barrier.wait())
|
||||
.await
|
||||
.map_err(|_| SkillProviderError::new("skill reads did not overlap"))?;
|
||||
}
|
||||
Ok(SkillReadResult {
|
||||
resource: request.resource,
|
||||
contents,
|
||||
@@ -145,6 +151,7 @@ text({ names: result.skills.map(skill => skill.name), warnings: result.warnings,
|
||||
warnings: Vec::new(),
|
||||
},
|
||||
reads: Mutex::default(),
|
||||
read_barrier: None,
|
||||
resources: std::collections::HashMap::from([
|
||||
(
|
||||
MAIN_RESOURCE.to_string(),
|
||||
@@ -493,6 +500,7 @@ async fn production_turn_aliases_discovered_singleton_cloud_root() -> Result<()>
|
||||
SKILL_BODY.to_string(),
|
||||
)]),
|
||||
reads: Mutex::default(),
|
||||
read_barrier: None,
|
||||
})),
|
||||
|config: &Config| SkillsExtensionConfig {
|
||||
include_instructions: config.include_skill_instructions,
|
||||
@@ -549,6 +557,99 @@ async fn production_turn_aliases_discovered_singleton_cloud_root() -> Result<()>
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn cloud_skill_reads_run_in_parallel() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
|
||||
const PACKAGE: &str = "skill://demo/parallel";
|
||||
const READS: [(&str, &str, &str); 2] = [
|
||||
(
|
||||
"read-main",
|
||||
"skill://demo/parallel/SKILL.md",
|
||||
"Main instructions.",
|
||||
),
|
||||
(
|
||||
"read-reference",
|
||||
"skill://demo/parallel/references/guide.md",
|
||||
"Reference instructions.",
|
||||
),
|
||||
];
|
||||
let server = responses::start_mock_server().await;
|
||||
let mut extensions = ExtensionRegistryBuilder::new();
|
||||
install_with_providers(
|
||||
&mut extensions,
|
||||
SkillProviders::new().with_cloud_provider(Arc::new(FakeCloudSkillProvider {
|
||||
catalog: SkillCatalog {
|
||||
entries: vec![SkillCatalogEntry::new(
|
||||
SkillPackageId(PACKAGE.to_string()),
|
||||
SkillAuthority::new(SkillSourceKind::Cloud, CODEX_APPS_MCP_SERVER_NAME),
|
||||
"demo:parallel",
|
||||
"Instructions and a reference.",
|
||||
SkillResourceId::new(READS[0].1),
|
||||
)],
|
||||
warnings: Vec::new(),
|
||||
},
|
||||
resources: READS
|
||||
.iter()
|
||||
.map(|(_, resource, contents)| (resource.to_string(), contents.to_string()))
|
||||
.collect(),
|
||||
reads: Mutex::default(),
|
||||
// An exclusive scheduler lock prevents the second read from reaching this barrier.
|
||||
read_barrier: Some(tokio::sync::Barrier::new(/*n*/ 2)),
|
||||
})),
|
||||
|config: &Config| SkillsExtensionConfig {
|
||||
include_instructions: config.include_skill_instructions,
|
||||
max_context_tokens: config.skill_max_context_tokens,
|
||||
bundled_skills_enabled: false,
|
||||
cloud_skill_enabled: true,
|
||||
shadow_selection_enabled: false,
|
||||
},
|
||||
);
|
||||
let chatgpt_base_url = server.uri();
|
||||
let mut builder = test_codex()
|
||||
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
|
||||
.with_exec_server_url("none")
|
||||
.with_extensions(Arc::new(extensions.build()))
|
||||
.with_config(move |config| {
|
||||
config.chatgpt_base_url = chatgpt_base_url;
|
||||
config.include_skill_instructions = true;
|
||||
config.cloud_skill_enabled = true;
|
||||
});
|
||||
let test = builder.build_with_auto_env(&server).await?;
|
||||
let mut calls = vec![ev_response_created("read-both")];
|
||||
calls.extend(READS.map(|(call_id, resource, _)| {
|
||||
responses::ev_function_call_with_namespace(
|
||||
call_id,
|
||||
"skills",
|
||||
"read",
|
||||
&json!({"package": PACKAGE, "resource": resource}).to_string(),
|
||||
)
|
||||
}));
|
||||
calls.push(ev_completed("read-both"));
|
||||
let response = responses::mount_sse_sequence(
|
||||
&server,
|
||||
vec![
|
||||
sse(calls),
|
||||
sse(vec![ev_response_created("done"), ev_completed("done")]),
|
||||
],
|
||||
)
|
||||
.await;
|
||||
test.submit_turn("Read both resources.").await?;
|
||||
|
||||
let requests = response.requests();
|
||||
assert_eq!(requests.len(), 2);
|
||||
for (call_id, resource, contents) in READS {
|
||||
let output = requests[1]
|
||||
.function_call_output_text(call_id)
|
||||
.expect("each concurrent skill read should return its resource");
|
||||
assert_eq!(
|
||||
serde_json::from_str::<Value>(&output)?,
|
||||
json!({"resource": resource, "contents": contents, "next_cursor": null})
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn cloud_skill_can_read_referenced_resource_without_an_executor() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
@@ -612,6 +713,7 @@ async fn cloud_skill_can_read_referenced_resource_without_an_executor() -> Resul
|
||||
),
|
||||
]),
|
||||
reads: Mutex::default(),
|
||||
read_barrier: None,
|
||||
});
|
||||
let mut extensions = ExtensionRegistryBuilder::new();
|
||||
install_with_providers(
|
||||
|
||||
@@ -31,6 +31,7 @@ async fn yielded_skill_read_keeps_originating_turn_metadata() -> Result<()> {
|
||||
"Yielded skill instructions.".to_string(),
|
||||
)]),
|
||||
reads: Mutex::default(),
|
||||
read_barrier: None,
|
||||
})),
|
||||
|config: &Config| SkillsExtensionConfig {
|
||||
include_instructions: config.include_skill_instructions,
|
||||
|
||||
@@ -149,6 +149,10 @@ impl<'call> ToolExecutor<ToolCall<'call>> for GoalToolExecutor {
|
||||
})
|
||||
}
|
||||
|
||||
fn supports_parallel_tool_calls(&self) -> bool {
|
||||
matches!(self.kind, GoalToolKind::Get)
|
||||
}
|
||||
|
||||
fn spec(&self) -> ToolSpec {
|
||||
match self.kind {
|
||||
GoalToolKind::Get => create_get_goal_tool(),
|
||||
|
||||
@@ -47,6 +47,10 @@ where
|
||||
memory_tool_name(LIST_TOOL_NAME)
|
||||
}
|
||||
|
||||
fn supports_parallel_tool_calls(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn spec(&self) -> ToolSpec {
|
||||
memory_function_tool::<ListArgs, ListMemoriesResponse>(
|
||||
LIST_TOOL_NAME,
|
||||
|
||||
@@ -46,6 +46,10 @@ where
|
||||
memory_tool_name(READ_TOOL_NAME)
|
||||
}
|
||||
|
||||
fn supports_parallel_tool_calls(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn spec(&self) -> ToolSpec {
|
||||
memory_function_tool::<ReadArgs, ReadMemoryResponse>(
|
||||
READ_TOOL_NAME,
|
||||
|
||||
@@ -55,6 +55,10 @@ where
|
||||
memory_tool_name(SEARCH_TOOL_NAME)
|
||||
}
|
||||
|
||||
fn supports_parallel_tool_calls(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn spec(&self) -> ToolSpec {
|
||||
memory_function_tool::<SearchArgs, SearchMemoriesResponse>(
|
||||
SEARCH_TOOL_NAME,
|
||||
|
||||
@@ -68,6 +68,10 @@ impl<'call> ToolExecutor<ToolCall<'call>> for ListTool {
|
||||
skill_tool_name(TOOL_NAME)
|
||||
}
|
||||
|
||||
fn supports_parallel_tool_calls(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn spec(&self) -> ToolSpec {
|
||||
skill_function_tool::<ListArgs, ListResponse>(
|
||||
TOOL_NAME,
|
||||
|
||||
@@ -61,6 +61,10 @@ impl<'call> ToolExecutor<ToolCall<'call>> for ReadTool {
|
||||
skill_tool_name(TOOL_NAME)
|
||||
}
|
||||
|
||||
fn supports_parallel_tool_calls(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn spec(&self) -> ToolSpec {
|
||||
skill_function_tool::<ReadArgs, ReadResponse>(
|
||||
TOOL_NAME,
|
||||
|
||||
Reference in New Issue
Block a user