Skip to content

core: box wrapper futures to reduce stack pressure #13429

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bolinfest merged 1 commit into from
Mar 4, 2026
Merged

core: box wrapper futures to reduce stack pressure #13429

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 46 additions & 48 deletions codex-rs/core/src/thread_manager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -312,8 +312,9 @@ impl ThreadManager {
}

pub async fn start_thread(&self, config: Config) -> CodexResult<NewThread> {
self.start_thread_with_tools(config, Vec::new(), false)
.await
// Box delegated thread-spawn futures so these convenience wrappers do
// not inline the full spawn path into every caller's async state.
Box::pin(self.start_thread_with_tools(config, Vec::new(), false)).await
}

pub async fn start_thread_with_tools(
Expand All @@ -322,12 +323,12 @@ impl ThreadManager {
dynamic_tools: Vec<codex_protocol::dynamic_tools::DynamicToolSpec>,
persist_extended_history: bool,
) -> CodexResult<NewThread> {
self.start_thread_with_tools_and_service_name(
Box::pin(self.start_thread_with_tools_and_service_name(
config,
dynamic_tools,
persist_extended_history,
None,
)
))
.await
}

Expand All @@ -338,17 +339,16 @@ impl ThreadManager {
persist_extended_history: bool,
metrics_service_name: Option<String>,
) -> CodexResult<NewThread> {
self.state
.spawn_thread(
config,
InitialHistory::New,
Arc::clone(&self.state.auth_manager),
self.agent_control(),
dynamic_tools,
persist_extended_history,
metrics_service_name,
)
.await
Box::pin(self.state.spawn_thread(
config,
InitialHistory::New,
Arc::clone(&self.state.auth_manager),
self.agent_control(),
dynamic_tools,
persist_extended_history,
metrics_service_name,
))
.await
}

pub async fn resume_thread_from_rollout(
Expand All @@ -358,7 +358,7 @@ impl ThreadManager {
auth_manager: Arc<AuthManager>,
) -> CodexResult<NewThread> {
let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?;
self.resume_thread_with_history(config, initial_history, auth_manager, false)
Box::pin(self.resume_thread_with_history(config, initial_history, auth_manager, false))
.await
}

Expand All @@ -369,17 +369,16 @@ impl ThreadManager {
auth_manager: Arc<AuthManager>,
persist_extended_history: bool,
) -> CodexResult<NewThread> {
self.state
.spawn_thread(
config,
initial_history,
auth_manager,
self.agent_control(),
Vec::new(),
persist_extended_history,
None,
)
.await
Box::pin(self.state.spawn_thread(
config,
initial_history,
auth_manager,
self.agent_control(),
Vec::new(),
persist_extended_history,
None,
))
.await
}

/// Removes the thread from the manager's internal map, though the thread is stored
Expand Down Expand Up @@ -411,17 +410,16 @@ impl ThreadManager {
) -> CodexResult<NewThread> {
let history = RolloutRecorder::get_rollout_history(&path).await?;
let history = truncate_before_nth_user_message(history, nth_user_message);
self.state
.spawn_thread(
config,
history,
Arc::clone(&self.state.auth_manager),
self.agent_control(),
Vec::new(),
persist_extended_history,
None,
)
.await
Box::pin(self.state.spawn_thread(
config,
history,
Arc::clone(&self.state.auth_manager),
self.agent_control(),
Vec::new(),
persist_extended_history,
None,
))
.await
}

pub(crate) fn agent_control(&self) -> AgentControl {
Expand Down Expand Up @@ -474,14 +472,14 @@ impl ThreadManagerState {
config: Config,
agent_control: AgentControl,
) -> CodexResult<NewThread> {
self.spawn_new_thread_with_source(
Box::pin(self.spawn_new_thread_with_source(
config,
agent_control,
self.session_source.clone(),
false,
None,
None,
)
))
.await
}

Expand All @@ -494,7 +492,7 @@ impl ThreadManagerState {
metrics_service_name: Option<String>,
inherited_shell_snapshot: Option<Arc<ShellSnapshot>>,
) -> CodexResult<NewThread> {
self.spawn_thread_with_source(
Box::pin(self.spawn_thread_with_source(
config,
InitialHistory::New,
Arc::clone(&self.auth_manager),
Expand All @@ -504,7 +502,7 @@ impl ThreadManagerState {
persist_extended_history,
metrics_service_name,
inherited_shell_snapshot,
)
))
.await
}

Expand All @@ -517,7 +515,7 @@ impl ThreadManagerState {
inherited_shell_snapshot: Option<Arc<ShellSnapshot>>,
) -> CodexResult<NewThread> {
let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?;
self.spawn_thread_with_source(
Box::pin(self.spawn_thread_with_source(
config,
initial_history,
Arc::clone(&self.auth_manager),
Expand All @@ -527,7 +525,7 @@ impl ThreadManagerState {
false,
None,
inherited_shell_snapshot,
)
))
.await
}

Expand All @@ -540,7 +538,7 @@ impl ThreadManagerState {
persist_extended_history: bool,
inherited_shell_snapshot: Option<Arc<ShellSnapshot>>,
) -> CodexResult<NewThread> {
self.spawn_thread_with_source(
Box::pin(self.spawn_thread_with_source(
config,
initial_history,
Arc::clone(&self.auth_manager),
Expand All @@ -550,7 +548,7 @@ impl ThreadManagerState {
persist_extended_history,
None,
inherited_shell_snapshot,
)
))
.await
}

Expand All @@ -566,7 +564,7 @@ impl ThreadManagerState {
persist_extended_history: bool,
metrics_service_name: Option<String>,
) -> CodexResult<NewThread> {
self.spawn_thread_with_source(
Box::pin(self.spawn_thread_with_source(
config,
initial_history,
auth_manager,
Expand All @@ -576,7 +574,7 @@ impl ThreadManagerState {
persist_extended_history,
metrics_service_name,
None,
)
))
.await
}

Expand Down
25 changes: 13 additions & 12 deletions codex-rs/core/tests/common/test_codex.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,7 @@ impl TestCodexBuilder {
Some(home) => home,
None => Arc::new(TempDir::new()?),
};
self.build_with_home(server, home, None).await
Box::pin(self.build_with_home(server, home, None)).await
}

pub async fn build_with_streaming_server(
Expand All @@ -117,8 +117,7 @@ impl TestCodexBuilder {
Some(home) => home,
None => Arc::new(TempDir::new()?),
};
self.build_with_home_and_base_url(format!("{base_url}/v1"), home, None)
.await
Box::pin(self.build_with_home_and_base_url(format!("{base_url}/v1"), home, None)).await
}

pub async fn build_with_websocket_server(
Expand All @@ -139,8 +138,7 @@ impl TestCodexBuilder {
.enable(Feature::ResponsesWebsockets)
.expect("test config should allow feature update");
}));
self.build_with_home_and_base_url(base_url, home, None)
.await
Box::pin(self.build_with_home_and_base_url(base_url, home, None)).await
}

pub async fn resume(
Expand All @@ -149,7 +147,7 @@ impl TestCodexBuilder {
home: Arc<TempDir>,
rollout_path: PathBuf,
) -> anyhow::Result<TestCodex> {
self.build_with_home(server, home, Some(rollout_path)).await
Box::pin(self.build_with_home(server, home, Some(rollout_path))).await
}

async fn build_with_home(
Expand All @@ -160,7 +158,7 @@ impl TestCodexBuilder {
) -> anyhow::Result<TestCodex> {
let base_url = format!("{}/v1", server.uri());
let (config, cwd) = self.prepare_config(base_url, &home).await?;
self.build_from_config(config, cwd, home, resume_from).await
Box::pin(self.build_from_config(config, cwd, home, resume_from)).await
}

async fn build_with_home_and_base_url(
Expand All @@ -170,7 +168,7 @@ impl TestCodexBuilder {
resume_from: Option<PathBuf>,
) -> anyhow::Result<TestCodex> {
let (config, cwd) = self.prepare_config(base_url, &home).await?;
self.build_from_config(config, cwd, home, resume_from).await
Box::pin(self.build_from_config(config, cwd, home, resume_from)).await
}

async fn build_from_config(
Expand Down Expand Up @@ -201,11 +199,14 @@ impl TestCodexBuilder {
let new_conversation = match resume_from {
Some(path) => {
let auth_manager = codex_core::test_support::auth_manager_from_auth(auth);
thread_manager
.resume_thread_from_rollout(config.clone(), path, auth_manager)
.await?
Box::pin(thread_manager.resume_thread_from_rollout(
config.clone(),
path,
auth_manager,
))
.await?
}
None => thread_manager.start_thread(config.clone()).await?,
None => Box::pin(thread_manager.start_thread(config.clone())).await?,
};

Ok(TestCodex {
Expand Down
Loading
Loading