openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-15 12:08:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: address leak in thread_resume_rejects_history_when_thread_is_running test

Browse Source
This commit is contained in:
Michael Bolin
2026-02-19 12:24:09 -08:00
parent 2f3d0b186b
commit edfd2ae7df
4 changed files with 133 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
codex-rs/app-server/tests/suite/v2/thread_resume.rs
View File

@@ -15,7 +15,10 @@ use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::TurnInterruptParams;
use codex_app_server_protocol::TurnInterruptResponse;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::TurnStatus;
use codex_app_server_protocol::UserInput;
use codex_protocol::config_types::Personality;
@@ -358,9 +361,10 @@ async fn thread_resume_rejects_history_when_thread_is_running() -> Result<()> {
.await??;
primary.clear_message_buffer();
let running_turn_id = primary
let thread_id = thread.id.clone();
let running_turn_request_id = primary
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
thread_id: thread_id.clone(),
input: vec![UserInput::Text {
text: "keep running".to_string(),
text_elements: Vec::new(),
@@ -368,11 +372,13 @@ async fn thread_resume_rejects_history_when_thread_is_running() -> Result<()> {
..Default::default()
})
.await?;
timeout(
let running_turn_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
primary.read_stream_until_response_message(RequestId::Integer(running_turn_id)),
primary.read_stream_until_response_message(RequestId::Integer(running_turn_request_id)),
)
.await??;
let TurnStartResponse { turn: running_turn } =
to_response::<TurnStartResponse>(running_turn_resp)?;
timeout(
DEFAULT_READ_TIMEOUT,
primary.read_stream_until_notification_message("turn/started"),
@@ -381,7 +387,7 @@ async fn thread_resume_rejects_history_when_thread_is_running() -> Result<()> {
let resume_id = primary
.send_thread_resume_request(ThreadResumeParams {
thread_id: thread.id,
thread_id: thread_id.clone(),
history: Some(vec![ResponseItem::Message {
id: None,
role: "user".to_string(),
@@ -407,6 +413,26 @@ async fn thread_resume_rejects_history_when_thread_is_running() -> Result<()> {
resume_err.error.message
);
let interrupt_id = primary
.send_turn_interrupt_request(TurnInterruptParams {
thread_id,
turn_id: running_turn.id,
})
.await?;
let interrupt_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
primary.read_stream_until_response_message(RequestId::Integer(interrupt_id)),
)
.await??;
let _turn_interrupt_response: TurnInterruptResponse =
to_response::<TurnInterruptResponse>(interrupt_resp)?;
timeout(
DEFAULT_READ_TIMEOUT,
primary.read_stream_until_notification_message("codex/event/turn_aborted"),
)
.await??;
Ok(())
}
@@ -462,9 +488,10 @@ async fn thread_resume_rejects_mismatched_path_when_thread_is_running() -> Resul
.await??;
primary.clear_message_buffer();
let running_turn_id = primary
let thread_id = thread.id.clone();
let running_turn_request_id = primary
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
thread_id: thread_id.clone(),
input: vec![UserInput::Text {
text: "keep running".to_string(),
text_elements: Vec::new(),
@@ -472,11 +499,13 @@ async fn thread_resume_rejects_mismatched_path_when_thread_is_running() -> Resul
..Default::default()
})
.await?;
timeout(
let running_turn_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
primary.read_stream_until_response_message(RequestId::Integer(running_turn_id)),
primary.read_stream_until_response_message(RequestId::Integer(running_turn_request_id)),
)
.await??;
let TurnStartResponse { turn: running_turn } =
to_response::<TurnStartResponse>(running_turn_resp)?;
timeout(
DEFAULT_READ_TIMEOUT,
primary.read_stream_until_notification_message("turn/started"),
@@ -485,7 +514,7 @@ async fn thread_resume_rejects_mismatched_path_when_thread_is_running() -> Resul
let resume_id = primary
.send_thread_resume_request(ThreadResumeParams {
thread_id: thread.id,
thread_id: thread_id.clone(),
path: Some(PathBuf::from("/tmp/does-not-match-running-rollout.jsonl")),
..Default::default()
})
@@ -501,6 +530,26 @@ async fn thread_resume_rejects_mismatched_path_when_thread_is_running() -> Resul
resume_err.error.message
);
let interrupt_id = primary
.send_turn_interrupt_request(TurnInterruptParams {
thread_id,
turn_id: running_turn.id,
})
.await?;
let interrupt_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
primary.read_stream_until_response_message(RequestId::Integer(interrupt_id)),
)
.await??;
let _turn_interrupt_response: TurnInterruptResponse =
to_response::<TurnInterruptResponse>(interrupt_resp)?;
timeout(
DEFAULT_READ_TIMEOUT,
primary.read_stream_until_notification_message("codex/event/turn_aborted"),
)
.await??;
Ok(())
}

7
codex-rs/app-server/tests/suite/v2/turn_start.rs
View File

@@ -747,7 +747,12 @@ async fn turn_start_accepts_local_image_input() -> Result<()> {
let TurnStartResponse { turn } = to_response::<TurnStartResponse>(turn_resp)?;
assert!(!turn.id.is_empty());
// This test only validates that turn/start responds and returns a turn.
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
Ok(())
}

49
codex-rs/app-server/tests/suite/v2/turn_start_zsh_fork.rs
View File

@@ -28,7 +28,10 @@ use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnInterruptParams;
use codex_app_server_protocol::TurnInterruptResponse;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::TurnStatus;
use codex_app_server_protocol::UserInput as V2UserInput;
use codex_core::features::FEATURES;
@@ -100,7 +103,7 @@ async fn turn_start_shell_zsh_fork_executes_command_v2() -> Result<()> {
let turn_id = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id,
thread_id: thread.id.clone(),
input: vec![V2UserInput::Text {
text: "run echo hi".to_string(),
text_elements: Vec::new(),
@@ -114,11 +117,12 @@ async fn turn_start_shell_zsh_fork_executes_command_v2() -> Result<()> {
..Default::default()
})
.await?;
timeout(
let turn_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(turn_id)),
)
.await??;
let TurnStartResponse { turn } = to_response::<TurnStartResponse>(turn_resp)?;
let started_command_execution = timeout(DEFAULT_READ_TIMEOUT, async {
loop {
@@ -149,6 +153,25 @@ async fn turn_start_shell_zsh_fork_executes_command_v2() -> Result<()> {
assert!(command.contains(" -lc 'echo hi'"));
assert_eq!(cwd, workspace);
let turn_interrupt_request_id = mcp
.send_turn_interrupt_request(TurnInterruptParams {
thread_id: thread.id,
turn_id: turn.id,
})
.await?;
let _: TurnInterruptResponse = to_response(
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(turn_interrupt_request_id)),
)
.await??,
)?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("codex/event/turn_aborted"),
)
.await??;
Ok(())
}
@@ -504,11 +527,12 @@ async fn turn_start_shell_zsh_fork_subcommand_decline_marks_parent_declined_v2()
..Default::default()
})
.await?;
timeout(
let turn_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(turn_id)),
)
.await??;
let TurnStartResponse { turn } = to_response::<TurnStartResponse>(turn_resp)?;
let mut approval_ids = Vec::new();
for decision in [
@@ -577,6 +601,25 @@ async fn turn_start_shell_zsh_fork_subcommand_decline_marks_parent_declined_v2()
assert_eq!(approval_ids.len(), 2);
assert_ne!(approval_ids[0], approval_ids[1]);
let turn_interrupt_request_id = mcp
.send_turn_interrupt_request(TurnInterruptParams {
thread_id: thread.id,
turn_id: turn.id,
})
.await?;
let _: TurnInterruptResponse = to_response(
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(turn_interrupt_request_id)),
)
.await??,
)?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("codex/event/turn_aborted"),
)
.await??;
Ok(())
}

23
codex-rs/app-server/tests/suite/v2/turn_steer.rs
View File

@@ -12,6 +12,8 @@ use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::TurnInterruptParams;
use codex_app_server_protocol::TurnInterruptResponse;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::TurnSteerParams;
@@ -136,7 +138,7 @@ async fn turn_steer_returns_active_turn_id() -> Result<()> {
let steer_req = mcp
.send_turn_steer_request(TurnSteerParams {
thread_id: thread.id,
thread_id: thread.id.clone(),
input: vec![V2UserInput::Text {
text: "steer".to_string(),
text_elements: Vec::new(),
@@ -152,6 +154,25 @@ async fn turn_steer_returns_active_turn_id() -> Result<()> {
let steer: TurnSteerResponse = to_response::<TurnSteerResponse>(steer_resp)?;
assert_eq!(steer.turn_id, turn.id);
let interrupt_req = mcp
.send_turn_interrupt_request(TurnInterruptParams {
thread_id: thread.id,
turn_id: steer.turn_id,
})
.await?;
let _: TurnInterruptResponse = to_response::<TurnInterruptResponse>(
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(interrupt_req)),
)
.await??,
)?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("codex/event/turn_aborted"),
)
.await??;
Ok(())
}