mirror of
https://github.com/openai/codex.git
synced 2026-10-18 17:18:52 +00:00
Cover instant interruption in steered-input persistence tests (#52048)
## What changed Extend `steered_input_checkpoint_controls_next_request` to cover user input with `InstantInterrupt` enabled under both background and synchronous checkpoint policies. Keep the original response gated until the follow-up request arrives to prove that interruption advances the turn while preserving checkpoint ordering. Explicitly disable `InstantInterrupt` for user-input cases that wait for response completion, and enable it for the tool-output case to verify that tool outputs still wait for completion and synchronous persistence. GitOrigin-RevId: d39544ce6f5ad83489fd8a80b1e731a64c7a876c
This commit is contained in:
@@ -66,6 +66,12 @@ enum InputKind {
|
||||
ToolOutput,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
enum InferenceCompletion {
|
||||
Response,
|
||||
Interrupt,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
enum FirstInputKind {
|
||||
User,
|
||||
@@ -174,13 +180,16 @@ impl ThreadStore for GatedCheckpointStore {
|
||||
}
|
||||
}
|
||||
|
||||
#[test_case(CheckpointPolicy::Background, InputKind::User; "background_user_input")]
|
||||
#[test_case(CheckpointPolicy::Synchronous, InputKind::User; "synchronous_store")]
|
||||
#[test_case(CheckpointPolicy::Background, InputKind::ToolOutput; "tool_output_stays_synchronous")]
|
||||
#[test_case(CheckpointPolicy::Background, InputKind::User, InferenceCompletion::Response; "background_user_input")]
|
||||
#[test_case(CheckpointPolicy::Synchronous, InputKind::User, InferenceCompletion::Response; "synchronous_store")]
|
||||
#[test_case(CheckpointPolicy::Background, InputKind::ToolOutput, InferenceCompletion::Response; "tool_output_stays_synchronous")]
|
||||
#[test_case(CheckpointPolicy::Background, InputKind::User, InferenceCompletion::Interrupt; "background_user_input_interrupts")]
|
||||
#[test_case(CheckpointPolicy::Synchronous, InputKind::User, InferenceCompletion::Interrupt; "synchronous_store_interrupts")]
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn steered_input_checkpoint_controls_next_request(
|
||||
policy: CheckpointPolicy,
|
||||
input_kind: InputKind,
|
||||
inference_completion: InferenceCompletion,
|
||||
) -> anyhow::Result<()> {
|
||||
let (first_completed, first_completion) = oneshot::channel();
|
||||
let (second_completed, second_completion) = oneshot::channel();
|
||||
@@ -225,7 +234,23 @@ async fn steered_input_checkpoint_controls_next_request(
|
||||
let test = test_codex()
|
||||
.with_thread_store(store.clone())
|
||||
.with_history_mode(ThreadHistoryMode::Legacy)
|
||||
.with_config(move |config| config.model_provider.base_url = Some(base_url))
|
||||
.with_config(move |config| {
|
||||
config.model_provider.base_url = Some(base_url);
|
||||
// Tool outputs must wait for response completion even with interruption enabled.
|
||||
if matches!(inference_completion, InferenceCompletion::Response)
|
||||
&& input_kind == InputKind::User
|
||||
{
|
||||
config
|
||||
.features
|
||||
.disable(Feature::InstantInterrupt)
|
||||
.expect("disable interruption for response completion");
|
||||
} else {
|
||||
config
|
||||
.features
|
||||
.enable(Feature::InstantInterrupt)
|
||||
.expect("enable interruption for steered input");
|
||||
}
|
||||
})
|
||||
.build_with_auto_env(&config_server)
|
||||
.await?;
|
||||
let first = test
|
||||
@@ -264,9 +289,17 @@ async fn steered_input_checkpoint_controls_next_request(
|
||||
turn_id
|
||||
}
|
||||
);
|
||||
// Steering still waits for the existing inference stream to finish.
|
||||
assert!(checkpoint_requests.try_recv().is_err());
|
||||
first_completed.send(()).expect("finish original inference");
|
||||
let first_completed = match inference_completion {
|
||||
InferenceCompletion::Response => {
|
||||
// Without preemption, steering waits for the existing inference stream.
|
||||
assert!(checkpoint_requests.try_recv().is_err());
|
||||
first_completed.send(()).expect("finish original inference");
|
||||
None
|
||||
}
|
||||
// Keep the first response gated until the follow-up request proves that
|
||||
// interruption, rather than server completion, advanced the turn.
|
||||
InferenceCompletion::Interrupt => Some(first_completed),
|
||||
};
|
||||
let checkpoint = timeout(Duration::from_secs(10), checkpoint_requests.recv())
|
||||
.await?
|
||||
.expect("Core should checkpoint the accepted input");
|
||||
@@ -298,6 +331,11 @@ async fn steered_input_checkpoint_controls_next_request(
|
||||
assert_eq!(requests.len(), 2);
|
||||
assert!(!String::from_utf8_lossy(&requests[0]).contains("steered input"));
|
||||
assert!(String::from_utf8_lossy(&requests[1]).contains("steered input"));
|
||||
if let Some(first_completed) = first_completed {
|
||||
first_completed
|
||||
.send(())
|
||||
.expect("release interrupted response");
|
||||
}
|
||||
second_completed
|
||||
.send(())
|
||||
.expect("finish follow-up inference");
|
||||
|
||||
Reference in New Issue
Block a user