diff --git a/codex-rs/core/src/compact_remote_v2_attempt.rs b/codex-rs/core/src/compact_remote_v2_attempt.rs index 2c77c045af..37bbc15afb 100644 --- a/codex-rs/core/src/compact_remote_v2_attempt.rs +++ b/codex-rs/core/src/compact_remote_v2_attempt.rs @@ -118,9 +118,7 @@ pub(super) async fn run_remote_compact_v2_attempt( ); } if estimated_deleted_tokens > 0 { - let max_local_deleted_tokens = sess - .estimated_tokens_after_last_model_generated_item() - .await; + let max_local_deleted_tokens = sess.estimated_tokens_since_last_usage().await; analytics_details.active_context_tokens_before = analytics_details .active_context_tokens_before .map(|active_context_tokens_before| { diff --git a/codex-rs/core/src/context_manager/history.rs b/codex-rs/core/src/context_manager/history.rs index 12fff896a8..d8be070f41 100644 --- a/codex-rs/core/src/context_manager/history.rs +++ b/codex-rs/core/src/context_manager/history.rs @@ -6,6 +6,7 @@ //! thread-owned reviewers still use only the parent model window. //! Old text checkpoints can seed that backup from their surviving plaintext instructions. //! Token estimates charge item content rather than transport metadata. +//! Live usage boundaries advance only when a response supplies usable usage. //! Oversized instructions keep an incomplete excerpt for bounded root review, including //! sources recovered from legacy Guardian checkpoints before their raw history is dropped. //! Prompt projections replace unavailable file images without changing retained history. @@ -112,6 +113,11 @@ pub(crate) struct ContextManager { /// Process-unique so resumed roots cannot match a worker's cached assistant evidence. guardian_review_context_revision: u64, token_info: Option, + /// History items at this index and later are estimated on top of the saved token count. + /// Failed responses preserve this index. It is not saved in rollouts. + last_server_reported_usage_item_count: usize, + /// Index immediately after the latest model output, including failed responses. + last_model_output_item_count: Option, /// Reference context snapshot used for diffing and producing model-visible /// settings update items. /// @@ -247,6 +253,8 @@ impl ContextManager { user_message_revision: 0, guardian_review_context_revision: next_guardian_review_context_revision(), token_info: None, + last_server_reported_usage_item_count: 0, + last_model_output_item_count: None, reference_context_item: None, world_state_baseline: None, } @@ -440,6 +448,16 @@ impl ContextManager { self.token_info = info; } + /// Install a context count together with the history it covers. + pub(crate) fn set_token_info_with_boundary( + &mut self, + info: Option, + covered_item_count: usize, + ) { + self.last_server_reported_usage_item_count = covered_item_count; + self.token_info = info; + } + pub(crate) fn set_reference_context_item(&mut self, item: Option) { self.reference_context_item = item; } @@ -593,6 +611,9 @@ impl ContextManager { if let Some(source) = &source { processed.metadata.get_or_insert_default().retained_source = Some(source.clone()); } + if is_model_generated_item(&processed.item) { + self.last_model_output_item_count = Some(self.items.len() + 1); + } Arc::make_mut(&mut self.items).push(processed); source } @@ -706,6 +727,7 @@ impl ContextManager { // its corresponding counterpart to keep the invariants intact without // running a full normalization pass. normalize::remove_corresponding_for(items, &removed.item); + self.reset_last_usage_boundary(); self.world_state_baseline = None; } } @@ -726,6 +748,7 @@ impl ContextManager { ); } self.items = Arc::new(items); + self.reset_last_usage_boundary(); self.history_version = self.history_version.saturating_add(1); self.reset_version = self.history_version; self.world_state_baseline = None; @@ -764,6 +787,7 @@ impl ContextManager { self.review_history = Some(retained); } self.items = Arc::new(items); + self.reset_last_usage_boundary(); self.history_version = self.history_version.saturating_add(1); if promoted { self.reset_version = self.history_version; @@ -902,13 +926,39 @@ impl ContextManager { usage: &TokenUsage, model_context_window: Option, includes_reasoning: bool, + input_item_count: usize, ) { - self.token_info = Some(TokenUsageInfo::new_or_append( + let info = TokenUsageInfo::new_or_append( &self.token_info, usage, includes_reasoning, model_context_window, - )); + ); + let covered_item_count = self + .last_model_output_item_count + .unwrap_or(0) + .max(input_item_count); + tracing::debug!( + target: "codex_core::history_item_count", + history_items = self.items.len(), + input_items = input_item_count, + covered_items = covered_item_count, + unreported_items = self.items.len().saturating_sub(covered_item_count), + "context history item counts" + ); + self.set_token_info_with_boundary(Some(info), covered_item_count); + } + + /// Restored or rewritten history keeps the existing last-model-output approximation. + fn reset_last_usage_boundary(&mut self) { + self.last_model_output_item_count = self + .items + .iter() + .rposition(|envelope| is_model_generated_item(&envelope.item)) + .map(|index| index + 1); + self.last_server_reported_usage_item_count = self + .last_model_output_item_count + .unwrap_or(self.items.len()); } fn get_non_last_reasoning_items_tokens(&self) -> i64 { @@ -923,7 +973,8 @@ impl ContextManager { self.items .iter() - .take(last_user_index) + // Reasoning after the usage boundary is already included in the tail estimate. + .take(last_user_index.min(self.last_server_reported_usage_item_count)) .filter(|envelope| { matches!( &envelope.item, @@ -937,17 +988,13 @@ impl ContextManager { .fold(0i64, i64::saturating_add) } - // These are local items added after the most recent model-emitted item. - // They are not reflected in `last_token_usage.total_tokens`. - fn items_after_last_model_generated_item( + // Keep failed or usage-less output in the estimate until a usable count covers it. + fn items_since_last_usage( &self, ) -> impl Clone + ExactSizeIterator + DoubleEndedIterator { - let start = self - .items + self.items[self.last_server_reported_usage_item_count..] .iter() - .rposition(|envelope| is_model_generated_item(&envelope.item)) - .map_or(self.items.len(), |index| index.saturating_add(1)); - self.items[start..].iter().map(|envelope| &envelope.item) + .map(|envelope| &envelope.item) } pub(crate) fn get_total_token_usage(&self) -> i64 { @@ -956,25 +1003,22 @@ impl ContextManager { .as_ref() .map(|info| info.last_token_usage.total_tokens) .unwrap_or(0); - let items_after_last_model_generated_tokens = self - .items_after_last_model_generated_item() - .map(estimate_item_token_count) - .fold(0i64, i64::saturating_add); + let unreported_tokens = self.estimated_tokens_since_last_usage(); if self .token_info .as_ref() .is_some_and(|info| info.last_usage_includes_reasoning == Some(true)) { - last_tokens.saturating_add(items_after_last_model_generated_tokens) + last_tokens.saturating_add(unreported_tokens) } else { last_tokens .saturating_add(self.get_non_last_reasoning_items_tokens()) - .saturating_add(items_after_last_model_generated_tokens) + .saturating_add(unreported_tokens) } } - pub(crate) fn estimated_tokens_after_last_model_generated_item(&self) -> i64 { - self.items_after_last_model_generated_item() + pub(crate) fn estimated_tokens_since_last_usage(&self) -> i64 { + self.items_since_last_usage() .map(estimate_item_token_count) .fold(0i64, i64::saturating_add) } @@ -997,6 +1041,11 @@ impl ContextManager { // strip audio when model does not support it normalize::strip_audio_when_unsupported(input_modalities, items); + + self.last_model_output_item_count = items + .iter() + .rposition(|envelope| is_model_generated_item(&envelope.item)) + .map(|index| index + 1); } /// Walk backward from a rollback cut and trim contiguous pre-turn context-update items. diff --git a/codex-rs/core/src/context_manager/history_tests.rs b/codex-rs/core/src/context_manager/history_tests.rs index 7e80afe42b..3aea92f140 100644 --- a/codex-rs/core/src/context_manager/history_tests.rs +++ b/codex-rs/core/src/context_manager/history_tests.rs @@ -924,13 +924,19 @@ fn non_last_reasoning_tokens_return_zero_when_no_user_messages() { #[test] fn non_last_reasoning_tokens_ignore_entries_after_last_user() { - let history = create_history_with_items(vec![ + let mut history = create_history_with_items(vec![ reasoning_with_encrypted_content(/*len*/ 900), user_msg("first"), reasoning_with_encrypted_content(/*len*/ 1_000), user_msg("second"), reasoning_with_encrypted_content(/*len*/ 2_000), ]); + history.update_token_info( + &TokenUsage::default(), + /*model_context_window*/ None, + /*includes_reasoning*/ false, + /*input_item_count*/ 0, + ); // first: (900 * 0.75 - 650) / 4 = 6.25 tokens // second: (1000 * 0.75 - 650) / 4 = 25 tokens // first + second = 62.5 @@ -944,7 +950,8 @@ fn total_token_usage_respects_saved_reasoning_accounting( includes_reasoning: Option, expected: i64, ) { - let mut history = create_history_with_items(vec![ + let mut history = ContextManager::new(); + history.replace(vec![ reasoning_with_encrypted_content(/*len*/ 1_000), user_msg("second"), assistant_msg("counted reply"), @@ -965,19 +972,28 @@ fn total_token_usage_respects_saved_reasoning_accounting( } #[test] -fn items_after_last_model_generated_tokens_include_user_and_tool_output() { - let history = create_history_with_items(vec![ - assistant_msg("already counted by API"), - user_msg("new user message"), - custom_tool_call_output("call-tail", "new tool output"), - ]); +fn items_since_last_usage_include_user_and_tool_output() { + let mut history = create_history_with_items(vec![assistant_msg("already counted by API")]); + history.update_token_info( + &TokenUsage::default(), + /*model_context_window*/ None, + /*includes_reasoning*/ true, + /*input_item_count*/ 0, + ); + history.record_items( + [ + &user_msg("new user message"), + &custom_tool_call_output("call-tail", "new tool output"), + ], + TruncationPolicy::Tokens(10_000), + ); let expected_tokens = estimate_item_token_count(&user_msg("new user message")).saturating_add( estimate_item_token_count(&custom_tool_call_output("call-tail", "new tool output")), ); assert_eq!( history - .items_after_last_model_generated_item() + .items_since_last_usage() .map(estimate_item_token_count) .fold(0i64, i64::saturating_add), expected_tokens @@ -985,15 +1001,15 @@ fn items_after_last_model_generated_tokens_include_user_and_tool_output() { } #[test] -fn items_after_last_model_generated_tokens_are_zero_without_model_generated_items() { +fn items_since_last_usage_include_initial_input() { let history = create_history_with_items(vec![user_msg("no model output yet")]); assert_eq!( history - .items_after_last_model_generated_item() + .items_since_last_usage() .map(estimate_item_token_count) .fold(0i64, i64::saturating_add), - 0 + estimate_item_token_count(&user_msg("no model output yet")) ); } @@ -1178,8 +1194,9 @@ fn legacy_inter_agent_assistant_messages_are_not_turn_boundaries() { assert!(!is_user_turn_boundary(&item)); } -#[test] -fn total_token_usage_includes_all_items_after_last_model_generated_item() { +#[test_case(false; "exclusive count")] +#[test_case(true; "inclusive count")] +fn total_token_usage_includes_all_items_since_last_usage(includes_reasoning: bool) { let mut history = create_history_with_items(vec![assistant_msg("already counted by API")]); history.update_token_info( &TokenUsage { @@ -1187,7 +1204,8 @@ fn total_token_usage_includes_all_items_after_last_model_generated_item() { ..Default::default() }, /*model_context_window*/ None, - /*includes_reasoning*/ true, + includes_reasoning, + /*input_item_count*/ 0, ); let added_user = user_msg("new user message"); let added_tool_output = custom_tool_call_output("tool-tail", "new tool output"); @@ -1201,6 +1219,54 @@ fn total_token_usage_includes_all_items_after_last_model_generated_item() { 100 + estimate_item_token_count(&added_user) + estimate_item_token_count(&added_tool_output) ); + + // An empty successful response still covers the input sent with that request. + let input_item_count = history.raw_items().len(); + history.update_token_info( + &TokenUsage { + total_tokens: 100, + ..Default::default() + }, + /*model_context_window*/ None, + includes_reasoning, + input_item_count, + ); + assert_eq!(history.get_total_token_usage(), 100); + + // A failed response can report spending without covering its retained output. + let partial = reasoning_with_encrypted_content(/*len*/ 2_000); + history.record_items([&partial], TruncationPolicy::Tokens(10_000)); + let mut spending = history.token_info().expect("saved usage"); + spending.total_token_usage.total_tokens += 15; + history.set_token_info(Some(spending)); + history.record_items([&added_user], TruncationPolicy::Tokens(10_000)); + let expected = + 100 + estimate_item_token_count(&partial) + estimate_item_token_count(&added_user); + assert_eq!(history.get_total_token_usage(), expected); + + // Local output after the final model item is not covered by server usage. + history.record_items( + [&assistant_msg("next response"), &added_tool_output], + TruncationPolicy::Tokens(10_000), + ); + history.update_token_info( + &TokenUsage { + total_tokens: 100, + ..Default::default() + }, + /*model_context_window*/ None, + includes_reasoning, + input_item_count, + ); + let prior_reasoning = if includes_reasoning { + 0 + } else { + estimate_item_token_count(&partial) + }; + assert_eq!( + history.get_total_token_usage(), + 100 + prior_reasoning + estimate_item_token_count(&added_tool_output), + ); } /// Unavailable images become harness-attributed placeholders without changing saved history or diff --git a/codex-rs/core/src/session/guardian_checkpoint_tests.rs b/codex-rs/core/src/session/guardian_checkpoint_tests.rs index c64bf40429..edac477801 100644 --- a/codex-rs/core/src/session/guardian_checkpoint_tests.rs +++ b/codex-rs/core/src/session/guardian_checkpoint_tests.rs @@ -79,6 +79,7 @@ async fn guardian_checkpoint_preserves_live_context_without_storage( }, Some(32_000), /*includes_reasoning*/ true, + /*input_item_count*/ 0, ); } let expected = session.clone_history().await; diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index f3913aea6c..c4a0dee278 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -418,7 +418,11 @@ use codex_utils_stream_parser::ProposedPlanSegment; /// Whether reported usage updates the installed context baseline or only spending. #[derive(Clone, Copy)] pub(crate) enum TokenUsageContext { - Sampling { includes_reasoning: bool }, + Sampling { + includes_reasoning: bool, + /// Input history length before the response, including requests with no output items. + input_item_count: usize, + }, SpendingOnly, } @@ -1592,11 +1596,9 @@ impl Session { state.auto_compact_window_snapshot() } - pub(crate) async fn estimated_tokens_after_last_model_generated_item(&self) -> i64 { + pub(crate) async fn estimated_tokens_since_last_usage(&self) -> i64 { let state = self.state.lock().await; - state - .history - .estimated_tokens_after_last_model_generated_item() + state.history.estimated_tokens_since_last_usage() } pub(crate) async fn total_token_usage(&self) -> Option { @@ -5326,7 +5328,10 @@ impl Session { turn_context, &turn_context.initial_settings, token_usage, - TokenUsageContext::Sampling { includes_reasoning }, + TokenUsageContext::Sampling { + includes_reasoning, + input_item_count: 0, + }, ) .await; self.send_token_count_event(turn_context).await; @@ -5388,11 +5393,15 @@ impl Session { let token_info = { let mut state = self.state.lock().await; match context { - TokenUsageContext::Sampling { includes_reasoning } => { + TokenUsageContext::Sampling { + includes_reasoning, + input_item_count, + } => { state.update_token_info_from_usage( token_usage, turn_context.model_context_window(), includes_reasoning, + input_item_count, ); if matches!( turn_context.config.model_auto_compact_token_limit_scope, @@ -5484,7 +5493,9 @@ impl Session { // The local estimate already includes every retained reasoning item. info.last_usage_includes_reasoning = Some(true); - state.set_token_info(Some(info)); + state + .history + .set_token_info_with_boundary(Some(info), history.raw_items().len()); } self.set_auto_compact_window_estimated_prefill_for_scope( turn_context, diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index d2d62152a0..9bb8aa2db3 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -3618,7 +3618,7 @@ async fn recompute_token_usage_updates_model_context_window() { assert_eq!(actual.model_context_window, Some(128_000)); } -#[test_case(crate::session::TokenUsageContext::Sampling { includes_reasoning: false }; "sampling")] +#[test_case(crate::session::TokenUsageContext::Sampling { includes_reasoning: false, input_item_count: 0 }; "sampling")] #[test_case(crate::session::TokenUsageContext::SpendingOnly; "spending_only")] #[tokio::test] async fn record_token_usage_info_notifies_extension_contributors( @@ -3707,6 +3707,7 @@ async fn record_token_usage_info_notifies_extension_contributors( Some(&first_usage), crate::session::TokenUsageContext::Sampling { includes_reasoning: true, + input_item_count: 0, }, ) .await @@ -3756,9 +3757,9 @@ async fn record_token_usage_info_notifies_extension_contributors( token_usage: TokenUsageInfo { total_token_usage: expected_total_usage, last_usage_includes_reasoning: match context { - crate::session::TokenUsageContext::Sampling { includes_reasoning } => { - Some(includes_reasoning) - } + crate::session::TokenUsageContext::Sampling { + includes_reasoning, .. + } => Some(includes_reasoning), crate::session::TokenUsageContext::SpendingOnly => None, }, last_token_usage: second_usage, diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index c575824c46..7227e13784 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -571,17 +571,10 @@ pub(crate) async fn run_turn( .await; // Construct the input that we will send to the model. - let sampling_request_input: Vec = async { - sess.clone_history() - .await - .for_prompt( - &step_context.settings.model_info.input_modalities, - sess.services.image_store.as_ref(), - ) - .await - } - .instrument(trace_span!("run_turn.prepare_sampling_request_input")) - .await; + let sampling_request_input = sess + .clone_history() + .instrument(trace_span!("run_turn.prepare_sampling_request_input")) + .await; run_sampling_request( Arc::clone(&sess), @@ -1744,7 +1737,7 @@ async fn run_sampling_request( turn_store: Arc, turn_diff_tracker: SharedTurnDiffTracker, client_session: &mut ModelClientSession, - input: Vec, + input: crate::context_manager::ContextManager, cancellation_token: CancellationToken, ) -> CodexResult<(SamplingRequestResult, Vec)> { let turn_context = Arc::clone(&step_context.turn); @@ -1785,18 +1778,18 @@ async fn run_sampling_request( // Running code-mode cells can request review while this response is in flight. // Keep the latest received ID until response.created replaces it. let first_attempt = initial_input.is_some(); - let prompt_input = if let Some(input) = initial_input.take() { + let history = if let Some(input) = initial_input.take() { input } else { - sess.clone_history() - .await - .for_prompt( - &step_context.settings.model_info.input_modalities, - sess.services.image_store.as_ref(), - ) - .await + sess.clone_history().await }; - let mut prompt_input = prompt_input; + let input_item_count = history.raw_items().len(); + let mut prompt_input = history + .for_prompt( + &step_context.settings.model_info.input_modalities, + sess.services.image_store.as_ref(), + ) + .await; sess.services .executed_tool_calls .attach_to_prompt(&mut prompt_input, &mut executed_tool_calls_by_output); @@ -1831,6 +1824,7 @@ async fn run_sampling_request( &responses_metadata, Arc::clone(&turn_diff_tracker), prompt, + input_item_count, cancellation_token.child_token(), &mut draining_for_mailbox, ) @@ -2683,6 +2677,7 @@ async fn try_run_sampling_request( responses_metadata: &CodexResponsesMetadata, turn_diff_tracker: SharedTurnDiffTracker, prompt: Prompt, + input_item_count: usize, cancellation_token: CancellationToken, draining_for_mailbox: &mut bool, ) -> CodexResult { @@ -3304,7 +3299,10 @@ async fn try_run_sampling_request( &turn_context, &step_context.settings, token_usage.as_ref(), - super::TokenUsageContext::Sampling { includes_reasoning }, + super::TokenUsageContext::Sampling { + includes_reasoning, + input_item_count, + }, ) .await; should_emit_token_count = true; diff --git a/codex-rs/core/src/state/session.rs b/codex-rs/core/src/state/session.rs index 968f3425fb..94c8874687 100644 --- a/codex-rs/core/src/state/session.rs +++ b/codex-rs/core/src/state/session.rs @@ -295,9 +295,14 @@ impl SessionState { usage: &TokenUsage, model_context_window: Option, includes_reasoning: bool, + input_item_count: usize, ) { - self.history - .update_token_info(usage, model_context_window, includes_reasoning); + self.history.update_token_info( + usage, + model_context_window, + includes_reasoning, + input_item_count, + ); } pub(crate) fn ensure_auto_compact_window_server_prefill_from_usage( diff --git a/codex-rs/core/src/tasks/mod_tests.rs b/codex-rs/core/src/tasks/mod_tests.rs index eb4bc4b5e2..32e80ee38e 100644 --- a/codex-rs/core/src/tasks/mod_tests.rs +++ b/codex-rs/core/src/tasks/mod_tests.rs @@ -232,6 +232,7 @@ async fn turn_completion_metrics_follow_model_switch(scenario: UsageScenario) { Some(&response_usage), crate::session::TokenUsageContext::Sampling { includes_reasoning: false, + input_item_count: 0, }, ) .await diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index c480acee0c..b522e2dedb 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -5591,6 +5591,76 @@ async fn auto_compact_accounts_for_encrypted_reasoning(continuation: ContinueThr ); } +#[test_case::test_case("response.incomplete", true, true; "incomplete_with_usage")] +#[test_case::test_case("response.incomplete", false, true; "incomplete_without_usage")] +#[test_case::test_case("response.completed", false, true; "completed_without_usage")] +#[test_case::test_case("response.incomplete", false, false; "non_lite_incomplete_without_usage")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn auto_compact_accounts_for_unreported_reasoning( + event_type: &str, + with_usage: bool, + use_responses_lite: bool, +) { + skip_if_no_network!(); + + let server = start_mock_server().await; + let mut responses = vec![sse(vec![ + ev_assistant_message("initial", "Initial reply."), + ev_completed_with_tokens("initial", /*total_tokens*/ 80), + ])]; + for id in ["partial-1", "partial-2"] { + let usage = with_usage.then(|| { + ev_completed_with_tokens(id, /*total_tokens*/ 15)["response"]["usage"].clone() + }); + responses.push(sse(vec![ + ev_reasoning_item(id, &["partial"], &[&"r".repeat(12_000)]), + json!({ + "type": event_type, + "response": { + "id": id, + "incomplete_details": {"reason": "max_output_tokens"}, + "usage": usage + } + }), + ])); + } + responses.push(remote_v2_compaction_response()); + responses.push(sse(vec![ + ev_assistant_message("final", FINAL_REPLY), + ev_completed_with_tokens("final", /*total_tokens*/ 1), + ])); + let request_log = mount_sse_sequence(&server, responses).await; + let test = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model_info_override("gpt-5.5", move |model| { + model.use_responses_lite = use_responses_lite; + }) + .with_config(|config| { + set_test_compact_prompt(config); + config.model_auto_compact_token_limit = Some(5_000); + config.model_provider.stream_max_retries = Some(0); + }) + .build_with_auto_env(&server) + .await + .expect("build codex"); + + for user in ["INITIAL", "PARTIAL_ONE", "PARTIAL_TWO", "CONTINUE"] { + test.submit_text_turn(user).await.expect("submit user turn"); + } + + // The saved count is 80. Each partial response retains about 3,000 tokens: + // one stays below the 5,000-token limit, but two must compact before CONTINUE. + assert_eq!( + request_log + .requests() + .iter() + .map(|request| request.inputs_of_type("compaction_trigger").len()) + .collect::>(), + vec![0, 0, 0, 1, 0], + "retained reasoning must trigger compaction before the continuation request", + ); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] // TODO(ccunningham): Update once pre-turn compaction includes incoming user input. async fn snapshot_request_shape_pre_turn_compaction_including_incoming_user_message() {