diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 316751f827..81474d6833 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -1,10 +1,12 @@ use std::sync::Arc; +use std::sync::OnceLock; use crate::compact_retained_tool_outputs::is_retained_tool_output; use crate::Prompt; use crate::ResponseStream; use crate::client::ModelClientSession; +use crate::client::ResponsesHttpExecutor; use crate::client_common::ResponseEvent; use crate::compact::CompactedHistoryMetadata; use crate::compact::CompactionAnalyticsAttempt; @@ -64,7 +66,11 @@ use tokio_util::sync::CancellationToken; #[path = "compact_remote_v2_attempt.rs"] mod attempt; +pub(crate) use attempt::BackgroundCompactionCandidate; use attempt::RemoteCompactV2Attempt; +use attempt::RemoteCompactV2AttemptInput; +pub(crate) use attempt::capture_background_input; +pub(crate) use attempt::run_background_compaction; use attempt::run_remote_compact_v2_attempt; #[path = "compact_remote_v2_images.rs"] @@ -85,10 +91,14 @@ const MAX_REMOTE_COMPACTION_V2_STREAM_RETRIES: u64 = 2; #[derive(Clone, Copy)] enum CompactionRequestKind<'a> { Foreground, - #[allow(dead_code)] Background(&'a CancellationToken), } +enum CompactionRequestTransport<'a> { + Foreground(&'a mut ModelClientSession), + Background(Option, Option), +} + pub(crate) async fn run_inline_remote_auto_compact_task( sess: Arc, step_context: Arc, @@ -250,10 +260,10 @@ async fn run_remote_compact_task_inner_impl( let attempt = run_remote_compact_v2_attempt( sess, step_context, - client_session.as_deref_mut(), &compaction_trace, compaction_metadata, analytics_details, + RemoteCompactV2AttemptInput::Foreground(client_session.as_deref_mut()), ) .await; let (attempt, compaction_turn_context) = match attempt { @@ -275,10 +285,10 @@ async fn run_remote_compact_task_inner_impl( let fallback_result = run_remote_compact_v2_attempt( sess, replacement_step_context, - client_session, &fallback_compaction_trace, compaction_metadata, analytics_details, + RemoteCompactV2AttemptInput::Foreground(client_session), ) .await; record_model_fallback( @@ -389,11 +399,42 @@ struct RemoteCompactionV2Output { async fn run_remote_compaction_request_v2( sess: &Session, step_context: &StepContext, - client_session: &mut ModelClientSession, prompt: &Prompt, responses_metadata: &CodexResponsesMetadata, + transport: CompactionRequestTransport<'_>, ) -> CodexResult { let turn_context = &step_context.turn; + let client_session = match transport { + CompactionRequestTransport::Background(turn_state, auth_owner_generation) => { + let settings = &step_context.settings; + let cancellation = crate::session::async_compaction::shutdown_cancellation(sess).await; + let effort = sess + .reasoning_effort_for_request(settings, RequestEffortUsage::Compaction) + .await; + let trace = InferenceTraceContext::disabled(); + // Replay the routing token without letting this request change foreground routing. + let turn_state = Arc::new(turn_state.map(OnceLock::from).unwrap_or_default()); + let http_executor = ResponsesHttpExecutor::new(&sess.services.model_client); + let stream = tokio::select! { + biased; + _ = cancellation.cancelled() => return Err(CodexErr::TurnAborted), + stream = http_executor.stream_responses_http( + &turn_state, auth_owner_generation, /*event_sender*/ None, prompt, &settings.model_info, &step_context.session_telemetry, effort, + settings.reasoning_summary, settings.service_tier.clone(), responses_metadata, + &trace, + ) => stream?, + }; + return collect_compaction_output( + sess, + turn_context, + stream, + CompactionRequestKind::Background(&cancellation), + ) + .await; + } + CompactionRequestTransport::Foreground(client_session) => client_session, + }; + let settings = &turn_context.initial_settings; let max_retries = turn_context .provider .info() @@ -404,14 +445,11 @@ async fn run_remote_compaction_request_v2( let result = match client_session .stream( prompt, - turn_context.model_info(), + &settings.model_info, &turn_context.session_telemetry, - sess.reasoning_effort_for_request( - &turn_context.initial_settings, - RequestEffortUsage::Compaction, - ) - .await, - turn_context.reasoning_summary(), + sess.reasoning_effort_for_request(settings, RequestEffortUsage::Compaction) + .await, + settings.reasoning_summary, step_context.settings.service_tier.clone(), responses_metadata, &InferenceTraceContext::disabled(), @@ -429,7 +467,6 @@ async fn run_remote_compaction_request_v2( } Err(err) => Err(err), }; - match result { Ok(compaction_output) => return Ok(compaction_output), Err(err) => { diff --git a/codex-rs/core/src/compact_remote_v2_attempt.rs b/codex-rs/core/src/compact_remote_v2_attempt.rs index 97a70670bc..0194f6526b 100644 --- a/codex-rs/core/src/compact_remote_v2_attempt.rs +++ b/codex-rs/core/src/compact_remote_v2_attempt.rs @@ -1,21 +1,32 @@ +//! Builds and sends foreground and speculative remote V2 compaction attempts. + use crate::context::UserGoalUpdate; use codex_protocol::ResponseItemId; use std::collections::HashSet; use std::sync::Arc; +use super::CompactionRequestTransport; use super::RemoteCompactionV2Output; use super::run_remote_compaction_request_v2; use crate::Prompt; use crate::client::ModelClientSession; use crate::compact::CompactionAnalyticsDetails; use crate::compact_remote_history::trim_function_call_history_to_fit_context_window; +use crate::context::world_state::WorldState; +use crate::context_manager::ContextManager; use crate::responses_metadata::CompactionTurnMetadata; +use crate::session::CompactionPrefix; use crate::session::session::Session; use crate::session::step_context::StepContext; +use codex_analytics::CompactionImplementation; +use codex_analytics::CompactionPhase; +use codex_analytics::CompactionReason; +use codex_analytics::CompactionTrigger; use codex_history::CodexHarnessMetadata; use codex_protocol::error::Result as CodexResult; use codex_protocol::models::BaseInstructions; use codex_protocol::models::ResponseItem; +use codex_protocol::openai_models::ModelInfo; use codex_protocol::protocol::TokenUsage; use codex_rollout_trace::CompactionTraceContext; use tracing::info; @@ -32,66 +43,132 @@ pub(super) struct RemoteCompactV2Attempt { pub(super) owned_client_session: Option, } +pub(super) enum RemoteCompactV2AttemptInput<'a> { + Foreground(Option<&'a mut ModelClientSession>), + Background(BackgroundCompactionInput), +} + +#[allow(dead_code)] +pub(crate) struct BackgroundCompactionCandidate { + pub(super) prefix: CompactionPrefix, + pub(super) model_info: Arc, + pub(super) cutoff_world_state: Arc, + pub(super) attempt: RemoteCompactV2Attempt, +} + +pub(crate) struct BackgroundCompactionInput { + prefix: CompactionPrefix, + turn_state: Option, + window: (String, u64, uuid::Uuid), + base_instructions: BaseInstructions, + input: Vec, + metadata: Vec>, +} + pub(super) async fn run_remote_compact_v2_attempt( sess: &Arc, step_context: &Arc, - client_session: Option<&mut ModelClientSession>, compaction_trace: &CompactionTraceContext, compaction_metadata: CompactionTurnMetadata, analytics_details: &mut CompactionAnalyticsDetails, + input: RemoteCompactV2AttemptInput<'_>, ) -> CodexResult { let turn_context = &step_context.turn; - let mut history = sess.clone_history().await; - let input_goal_ids = UserGoalUpdate::message_ids(history.raw_items()); - let tools_in_history = step_context.settings.model_info.use_responses_lite - && history - .raw_items() - .any(|item| matches!(item, ResponseItem::AdditionalTools { .. })); - let base_instructions = if tools_in_history { - BaseInstructions { - text: String::new(), - provenance: None, - } - } else { - sess.get_prompt_base_instructions().await - }; - let (rewritten_outputs, estimated_deleted_tokens) = - trim_function_call_history_to_fit_context_window( - &mut history, - turn_context, - turn_context.model_context_window(), - &base_instructions, - ); - if rewritten_outputs > 0 { - info!( - turn_id = %turn_context.sub_id, - rewritten_outputs, - "rewrote history outputs before remote compaction v2" - ); - } - if estimated_deleted_tokens > 0 { - let max_local_deleted_tokens = sess - .estimated_tokens_after_last_model_generated_item() - .await; - analytics_details.active_context_tokens_before = analytics_details - .active_context_tokens_before - .map(|active_context_tokens_before| { - active_context_tokens_before - .saturating_sub(estimated_deleted_tokens.min(max_local_deleted_tokens)) + let mut owned_client_session = None; + let ( + model_info, + input_goal_ids, + trace_input_history, + base_instructions, + mut input, + prompt_input_metadata, + transport, + captured_window, + ) = match input { + RemoteCompactV2AttemptInput::Foreground(client_session) => { + let model_info = turn_context.model_info(); + let mut history = sess.clone_history().await; + let input_goal_ids = UserGoalUpdate::message_ids(history.raw_items()); + let tools_in_history = step_context.settings.model_info.use_responses_lite + && history + .raw_items() + .any(|item| matches!(item, ResponseItem::AdditionalTools { .. })); + let base_instructions = if tools_in_history { + BaseInstructions { + text: String::new(), + provenance: None, + } + } else { + sess.get_prompt_base_instructions().await + }; + let (rewritten_outputs, estimated_deleted_tokens) = + trim_function_call_history_to_fit_context_window( + &mut history, + turn_context, + turn_context.model_context_window(), + &base_instructions, + ); + if rewritten_outputs > 0 { + info!( + turn_id = %turn_context.sub_id, + rewritten_outputs, + "rewrote history outputs before remote compaction v2" + ); + } + if estimated_deleted_tokens > 0 { + let max_local_deleted_tokens = sess + .estimated_tokens_after_last_model_generated_item() + .await; + analytics_details.active_context_tokens_before = analytics_details + .active_context_tokens_before + .map(|active_context_tokens_before| { + active_context_tokens_before + .saturating_sub(estimated_deleted_tokens.min(max_local_deleted_tokens)) + }); + } + let trace_input_history = compaction_trace + .is_enabled() + .then(|| history.raw_items().cloned().collect()); + let (mut input, metadata): (Vec<_>, Vec<_>) = history + .for_prompt_annotated(&model_info.input_modalities) + .into_iter() + .map(|envelope| (envelope.item, envelope.metadata)) + .unzip(); + sess.services + .executed_tool_calls + .attach_to_compaction_prompt(&mut input); + let client_session = client_session.unwrap_or_else(|| { + owned_client_session.insert(sess.services.model_client.new_session()) }); - } - - let trace_input_history = compaction_trace - .is_enabled() - .then(|| history.raw_items().cloned().collect()); - let (mut input, prompt_input_metadata): (Vec<_>, Vec<_>) = history - .for_prompt_annotated(&turn_context.model_info().input_modalities) - .into_iter() - .map(|envelope| (envelope.item, envelope.metadata)) - .unzip(); - sess.services - .executed_tool_calls - .attach_to_compaction_prompt(&mut input); + ( + model_info, + input_goal_ids, + trace_input_history, + base_instructions, + input, + metadata, + CompactionRequestTransport::Foreground(client_session), + None, + ) + } + RemoteCompactV2AttemptInput::Background(input) => ( + &step_context.settings.model_info, + UserGoalUpdate::message_ids(input.input.iter()), + None, + input.base_instructions, + input.input, + input.metadata, + CompactionRequestTransport::Background( + input.turn_state, + input.prefix.auth_owner_generation, + ), + Some(input.window), + ), + }; + let tools_in_history = model_info.use_responses_lite + && input + .iter() + .any(|item| matches!(item, ResponseItem::AdditionalTools { .. })); let tool_router = &step_context.tool_router; input.push(ResponseItem::CompactionTrigger {}); let prompt = Prompt { @@ -108,27 +185,28 @@ pub(super) async fn run_remote_compact_v2_attempt( cyber_access_program: turn_context.cyber_access_program, }; - let mut responses_metadata = sess - .compaction_responses_metadata(turn_context.as_ref(), compaction_metadata) - .await; + let mut responses_metadata = match captured_window { + Some(window) => { + sess.compaction_responses_metadata_for_window(turn_context, compaction_metadata, window) + } + None => { + sess.compaction_responses_metadata(turn_context, compaction_metadata) + .await + } + }; responses_metadata.tool_namespaces_info = tool_router.tool_namespaces_info().cloned(); let trace_attempt = compaction_trace.start_attempt(&serde_json::json!({ - "model": turn_context.model_info().slug.as_str(), + "model": model_info.slug.as_str(), "instructions": prompt.base_instructions.text.as_str(), "input": &prompt.input, "parallel_tool_calls": prompt.parallel_tool_calls, })); - let mut owned_client_session = None; - let client_session = match client_session { - Some(client_session) => client_session, - None => owned_client_session.insert(sess.services.model_client.new_session()), - }; let compaction_output_result = run_remote_compaction_request_v2( sess, step_context, - client_session, &prompt, &responses_metadata, + transport, ) .await; trace_attempt.record_result( @@ -154,3 +232,83 @@ pub(super) async fn run_remote_compact_v2_attempt( owned_client_session, }) } + +pub(crate) fn capture_background_input( + sess: &Session, + step_context: &StepContext, + mut snapshot: ContextManager, + base_instructions: BaseInstructions, + turn_state: Option, + auth_owner_generation: Option, + window: (String, u64, uuid::Uuid), +) -> BackgroundCompactionInput { + let base_instructions = if step_context.settings.model_info.use_responses_lite + && snapshot + .raw_items() + .any(|item| matches!(item, ResponseItem::AdditionalTools { .. })) + { + BaseInstructions { + text: String::new(), + provenance: None, + } + } else { + base_instructions + }; + let prefix = CompactionPrefix { + len: snapshot.annotated_items().len(), + auth_owner_generation, + }; + trim_function_call_history_to_fit_context_window( + &mut snapshot, + &step_context.turn, + step_context.settings.model_info.usable_context_window(), + &base_instructions, + ); + let (mut input, metadata): (Vec<_>, Vec<_>) = snapshot + .for_prompt_annotated(&step_context.settings.model_info.input_modalities) + .into_iter() + .map(|envelope| (envelope.item, envelope.metadata)) + .unzip(); + sess.services + .executed_tool_calls + .attach_to_compaction_prompt(&mut input); + BackgroundCompactionInput { + prefix, + turn_state, + window, + base_instructions, + input, + metadata, + } +} + +pub(crate) async fn run_background_compaction( + sess: &Arc, + step_context: Arc, + cutoff_world_state: Arc, + phase: CompactionPhase, + prepared_input: BackgroundCompactionInput, +) -> CodexResult { + let prefix = prepared_input.prefix; + let metadata = CompactionTurnMetadata::new( + CompactionTrigger::Auto, + CompactionReason::ContextLimit, + CompactionImplementation::ResponsesCompactionV2, + phase, + ); + let attempt = run_remote_compact_v2_attempt( + sess, + &step_context, + &CompactionTraceContext::disabled(), + metadata, + &mut CompactionAnalyticsDetails::default(), + RemoteCompactV2AttemptInput::Background(prepared_input), + ) + .await?; + Ok(BackgroundCompactionCandidate { + prefix, + model_info: Arc::clone(&step_context.settings.model_info), + cutoff_world_state, + attempt, + }) +} diff --git a/codex-rs/core/src/session/async_compaction.rs b/codex-rs/core/src/session/async_compaction.rs new file mode 100644 index 0000000000..1dedc11b90 --- /dev/null +++ b/codex-rs/core/src/session/async_compaction.rs @@ -0,0 +1,160 @@ +//! Holds the latest speculative request in memory until the foreground can use it. +//! Discarded requests still charge completed usage; session shutdown cancels remaining network work. + +use std::sync::Arc; + +use crate::compact_remote_v2::BackgroundCompactionCandidate; +use crate::compact_remote_v2::capture_background_input; +use crate::compact_remote_v2::run_background_compaction; +use crate::context::world_state::WorldState; +use crate::session::context_window::ContextWindowTokenStatus; +use crate::session::session::Session; +use crate::session::step_context::StepContext; +use crate::session::turn_context::TurnContext; +use codex_analytics::CompactionPhase; +use codex_features::Feature; +use codex_model_provider::RemoteCompactionSupport; +use codex_otel::auth_storage::AuthStorageOriginator; +use tokio::sync::oneshot; +use tokio_util::sync::CancellationToken; +use tokio_util::task::TaskTracker; +use tracing::debug; + +#[derive(PartialEq, Eq)] +struct CompactionAttempt { + history_version: u64, + model_slug: String, + compaction_hash: Option, + auth_owner_generation: Option, +} + +#[derive(Default)] +pub(crate) struct AsyncCompactionState { + last_attempt: Option, + result: Option>, + tasks: TaskTracker, + shutdown_cancellation: CancellationToken, +} + +pub(super) fn enabled(turn_context: &TurnContext) -> bool { + turn_context + .config + .features + .enabled(Feature::AsyncCompaction) + && !turn_context.config.features.enabled(Feature::TokenBudget) + && matches!( + turn_context.provider.capabilities().remote_compaction, + RemoteCompactionSupport::V2 + ) +} + +pub(super) async fn maybe_start( + sess: &Arc, + step_context: &Arc, + world_state: &Arc, + phase: CompactionPhase, + token_status: &ContextWindowTokenStatus, + (turn_state, auth_owner_generation): (Option, Option), +) { + let turn_context = &step_context.turn; + // Let the foreground establish the backend's model pin before starting a parallel request. + if matches!(phase, CompactionPhase::PreTurn) && turn_state.is_none() { + return; + } + if !enabled(turn_context) + || token_status.token_limit_reached + || (matches!(phase, CompactionPhase::PostTurn) + && token_status.turn_end_compaction_threshold_reached) + || step_context + .settings + .model_info + .resolved_context_window() + .is_none_or(|limit| { + i128::from(token_status.active_context_tokens) * 100 + < i128::from(limit) + * i128::from(turn_context.config.async_compaction_threshold_percent) + }) + { + return; + } + + let base_instructions = sess.get_prompt_base_instructions().await; + let mut state = sess.state.lock().await; + let model_info = &step_context.settings.model_info; + let attempt = CompactionAttempt { + history_version: state.history.history_version(), + model_slug: model_info.slug.clone(), + compaction_hash: model_info.comp_hash.clone(), + auth_owner_generation, + }; + if state.shutting_down + || state.async_compaction.last_attempt.as_ref() == Some(&attempt) + || state.history.annotated_items().is_empty() + || sess.services.model_client.auth_owner_generation() != auth_owner_generation + { + return; + } + + // Capture and register without yielding, so shutdown or history replacement cannot + // overtake this request. Tool attribution still belongs to the foreground's current step. + let prepared_input = capture_background_input( + sess, + step_context, + state.history.clone(), + base_instructions, + turn_state, + auth_owner_generation, + sess.current_window_from_state(&state), + ); + let task_session = Arc::clone(sess); + let task_step = Arc::clone(step_context); + let cutoff_world_state = Arc::clone(world_state); + let storage_originator = AuthStorageOriginator::from_client_name(&turn_context.originator); + state.async_compaction.last_attempt = Some(attempt); + let (send, result) = oneshot::channel(); + state.async_compaction.result = Some(result); + let task = async move { + match run_background_compaction( + &task_session, + task_step, + cutoff_world_state, + phase, + prepared_input, + ) + .await + { + Ok(candidate) => { + let _ = send.send(candidate); + } + Err(error) => { + debug!(%error, "discarding unsuccessful background compaction"); + } + } + }; + state + .async_compaction + .tasks + .spawn(storage_originator.scope(task)); +} + +pub(crate) async fn shutdown_cancellation(sess: &Session) -> CancellationToken { + sess.state + .lock() + .await + .async_compaction + .shutdown_cancellation + .clone() +} + +pub(super) async fn shutdown(sess: &Session) { + let tasks = { + let mut state = sess.state.lock().await; + let async_compaction = &mut state.async_compaction; + async_compaction.result = None; + async_compaction.tasks.close(); + async_compaction.shutdown_cancellation.cancel(); + async_compaction.tasks.clone() + }; + // Drain completed usage charging before shutdown finishes. + tasks.wait().await; +} diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 72a6b55f6e..91bf8de9bf 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -299,6 +299,7 @@ pub(super) async fn shutdown_session_runtime(sess: &Arc) { } let _ = sess.conversation.shutdown().await; sess.abort_all_tasks(TurnAbortReason::Interrupted).await; + super::async_compaction::shutdown(sess).await; let shell_snapshot_prewarm = sess.state.lock().await.shell_snapshot_prewarm.take(); if let Some(shell_snapshot_prewarm) = shell_snapshot_prewarm { shell_snapshot_prewarm.abort(); diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 1a55f1cfc9..b5f12424b3 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -237,6 +237,8 @@ use codex_protocol::error::Result as CodexResult; #[cfg(test)] use codex_protocol::exec_output::StreamOutput; +#[allow(dead_code)] +pub(crate) mod async_compaction; mod code_mode_warning; mod config_refresh; pub(crate) mod context_window; @@ -445,6 +447,7 @@ pub(crate) enum ForkPersistence { #[derive(Clone, Copy)] pub(crate) struct CompactionPrefix { pub(crate) len: usize, + pub(crate) auth_owner_generation: Option, } pub(crate) enum CompactedHistoryInstallation<'a> { @@ -3745,7 +3748,7 @@ impl Session { pub(crate) async fn record_step_world_state_if_changed( &self, step_context: &step_context::StepContext, - ) -> CodexResult<()> { + ) -> CodexResult> { let turn_context = step_context.turn.as_ref(); // Render model-visible state from the same step used to build and run tools. let world_state = self @@ -3773,7 +3776,7 @@ impl Session { self.persist_rollout_items(&[RolloutItem::WorldState(world_state_item)]) .await; } - Ok(()) + Ok(Arc::new(world_state)) } /// Retains the step captured for execution. @@ -4822,7 +4825,7 @@ impl Session { pub(crate) async fn record_context_updates_and_set_reference_context_item( &self, step_context: &StepContext, - ) -> CodexResult<()> { + ) -> CodexResult> { let turn_context = step_context.turn.as_ref(); let reference_context_item = { let state = self.state.lock().await; @@ -4870,7 +4873,7 @@ impl Session { // A snapshot can change without producing model-visible or TurnContext updates. let only_world_state_changed = !turn_context_changed && context_items.is_empty(); if only_world_state_changed && world_state_item.is_none() { - return Ok(()); + return Ok(Arc::new(world_state)); } if !context_items.is_empty() { self.record_conversation_items( @@ -4887,7 +4890,7 @@ impl Session { } // A snapshot-only change does not require a duplicate TurnContext record. if only_world_state_changed { - return Ok(()); + return Ok(Arc::new(world_state)); } // Persist one `TurnContextItem` per real user turn so resume/lazy replay can recover the // latest durable baseline even when this turn emitted no model-visible context diffs. @@ -4911,7 +4914,7 @@ impl Session { // context items. let mut state = self.state.lock().await; state.set_reference_context_item(Some(turn_context_item)); - Ok(()) + Ok(Arc::new(world_state)) } pub(crate) async fn update_token_usage_info( diff --git a/codex-rs/core/src/state/session.rs b/codex-rs/core/src/state/session.rs index 0f63dad3ee..fff1bf667f 100644 --- a/codex-rs/core/src/state/session.rs +++ b/codex-rs/core/src/state/session.rs @@ -91,6 +91,7 @@ pub(crate) struct SessionState { pub(crate) turn_attribution: Option, /// Runtime accounting state for the active auto-compaction window. auto_compact_window: AutoCompactWindow, + pub(crate) async_compaction: crate::session::async_compaction::AsyncCompactionState, /// Original request effort for the current model while configuration updates remain active. pub(crate) reasoning_effort_pin: ReasoningEffortPin, /// Set under the state lock before shutdown takes the last warmup handle. @@ -136,6 +137,7 @@ impl SessionState { last_started_turn_id: None, turn_attribution: None, auto_compact_window: AutoCompactWindow::new_with_ids(auto_compact_window_ids), + async_compaction: crate::session::async_compaction::AsyncCompactionState::default(), reasoning_effort_pin: ReasoningEffortPin::Unset, shutting_down: false, startup_prewarm: None,