mirror of
https://github.com/pchuan98/codex.git
synced 2026-07-01 00:31:56 +08:00
## Description This PR fixes a bad first-turn failure mode in app-server when the startup websocket prewarm hangs. Before this change, `initialize -> thread/start -> turn/start` could sit behind the prewarm for up to five minutes, so the client would not see `turn/started`, and even `turn/interrupt` would block because the turn had not actually started yet. Now, we: - set a (configurable) timeout of 15s for websocket startup time, exposed as `websocket_startup_timeout_ms` in config.toml - `turn/started` is sent immediately on `turn/start` even if the websocket is still connecting - `turn/interrupt` can be used to cancel a turn that is still waiting on the websocket warmup - the turn task will wait for the full 15s websocket warming timeout before falling back ## Why The old behavior made app-server feel stuck at exactly the moment the client expects turn lifecycle events to start flowing. That was especially painful for external clients, because from their point of view the server had accepted the request but then went silent for minutes. ## Configuring the websocket startup timeout Can set it in config.toml like this: ``` [model_providers.openai] supports_websockets = true websocket_connect_timeout_ms = 15000 ```
241 lines
8.0 KiB
Rust
241 lines
8.0 KiB
Rust
//! Session-wide mutable state.
|
|
|
|
use codex_protocol::models::PermissionProfile;
|
|
use codex_protocol::models::ResponseItem;
|
|
use std::collections::HashMap;
|
|
use std::collections::HashSet;
|
|
|
|
use crate::codex::PreviousTurnSettings;
|
|
use crate::codex::SessionConfiguration;
|
|
use crate::context_manager::ContextManager;
|
|
use crate::protocol::RateLimitSnapshot;
|
|
use crate::protocol::TokenUsage;
|
|
use crate::protocol::TokenUsageInfo;
|
|
use crate::sandboxing::merge_permission_profiles;
|
|
use crate::session_startup_prewarm::SessionStartupPrewarmHandle;
|
|
use crate::truncate::TruncationPolicy;
|
|
use codex_protocol::protocol::TurnContextItem;
|
|
|
|
/// Persistent, session-scoped state previously stored directly on `Session`.
|
|
pub(crate) struct SessionState {
|
|
pub(crate) session_configuration: SessionConfiguration,
|
|
pub(crate) history: ContextManager,
|
|
pub(crate) latest_rate_limits: Option<RateLimitSnapshot>,
|
|
pub(crate) server_reasoning_included: bool,
|
|
pub(crate) dependency_env: HashMap<String, String>,
|
|
pub(crate) mcp_dependency_prompted: HashSet<String>,
|
|
/// Settings used by the latest regular user turn, used for turn-to-turn
|
|
/// model/realtime handling on subsequent regular turns (including full-context
|
|
/// reinjection after resume or `/compact`).
|
|
previous_turn_settings: Option<PreviousTurnSettings>,
|
|
/// Startup prewarmed session prepared during session initialization.
|
|
pub(crate) startup_prewarm: Option<SessionStartupPrewarmHandle>,
|
|
pub(crate) active_connector_selection: HashSet<String>,
|
|
pub(crate) pending_session_start_source: Option<codex_hooks::SessionStartSource>,
|
|
granted_permissions: Option<PermissionProfile>,
|
|
}
|
|
|
|
impl SessionState {
|
|
/// Create a new session state mirroring previous `State::default()` semantics.
|
|
pub(crate) fn new(session_configuration: SessionConfiguration) -> Self {
|
|
let history = ContextManager::new();
|
|
Self {
|
|
session_configuration,
|
|
history,
|
|
latest_rate_limits: None,
|
|
server_reasoning_included: false,
|
|
dependency_env: HashMap::new(),
|
|
mcp_dependency_prompted: HashSet::new(),
|
|
previous_turn_settings: None,
|
|
startup_prewarm: None,
|
|
active_connector_selection: HashSet::new(),
|
|
pending_session_start_source: None,
|
|
granted_permissions: None,
|
|
}
|
|
}
|
|
|
|
// History helpers
|
|
pub(crate) fn record_items<I>(&mut self, items: I, policy: TruncationPolicy)
|
|
where
|
|
I: IntoIterator,
|
|
I::Item: std::ops::Deref<Target = ResponseItem>,
|
|
{
|
|
self.history.record_items(items, policy);
|
|
}
|
|
|
|
pub(crate) fn previous_turn_settings(&self) -> Option<PreviousTurnSettings> {
|
|
self.previous_turn_settings.clone()
|
|
}
|
|
pub(crate) fn set_previous_turn_settings(
|
|
&mut self,
|
|
previous_turn_settings: Option<PreviousTurnSettings>,
|
|
) {
|
|
self.previous_turn_settings = previous_turn_settings;
|
|
}
|
|
|
|
pub(crate) fn clone_history(&self) -> ContextManager {
|
|
self.history.clone()
|
|
}
|
|
|
|
pub(crate) fn replace_history(
|
|
&mut self,
|
|
items: Vec<ResponseItem>,
|
|
reference_context_item: Option<TurnContextItem>,
|
|
) {
|
|
self.history.replace(items);
|
|
self.history
|
|
.set_reference_context_item(reference_context_item);
|
|
}
|
|
|
|
pub(crate) fn set_token_info(&mut self, info: Option<TokenUsageInfo>) {
|
|
self.history.set_token_info(info);
|
|
}
|
|
|
|
pub(crate) fn set_reference_context_item(&mut self, item: Option<TurnContextItem>) {
|
|
self.history.set_reference_context_item(item);
|
|
}
|
|
|
|
pub(crate) fn reference_context_item(&self) -> Option<TurnContextItem> {
|
|
self.history.reference_context_item()
|
|
}
|
|
|
|
// Token/rate limit helpers
|
|
pub(crate) fn update_token_info_from_usage(
|
|
&mut self,
|
|
usage: &TokenUsage,
|
|
model_context_window: Option<i64>,
|
|
) {
|
|
self.history.update_token_info(usage, model_context_window);
|
|
}
|
|
|
|
pub(crate) fn token_info(&self) -> Option<TokenUsageInfo> {
|
|
self.history.token_info()
|
|
}
|
|
|
|
pub(crate) fn set_rate_limits(&mut self, snapshot: RateLimitSnapshot) {
|
|
self.latest_rate_limits = Some(merge_rate_limit_fields(
|
|
self.latest_rate_limits.as_ref(),
|
|
snapshot,
|
|
));
|
|
}
|
|
|
|
pub(crate) fn token_info_and_rate_limits(
|
|
&self,
|
|
) -> (Option<TokenUsageInfo>, Option<RateLimitSnapshot>) {
|
|
(self.token_info(), self.latest_rate_limits.clone())
|
|
}
|
|
|
|
pub(crate) fn set_token_usage_full(&mut self, context_window: i64) {
|
|
self.history.set_token_usage_full(context_window);
|
|
}
|
|
|
|
pub(crate) fn get_total_token_usage(&self, server_reasoning_included: bool) -> i64 {
|
|
self.history
|
|
.get_total_token_usage(server_reasoning_included)
|
|
}
|
|
|
|
pub(crate) fn set_server_reasoning_included(&mut self, included: bool) {
|
|
self.server_reasoning_included = included;
|
|
}
|
|
|
|
pub(crate) fn server_reasoning_included(&self) -> bool {
|
|
self.server_reasoning_included
|
|
}
|
|
|
|
pub(crate) fn record_mcp_dependency_prompted<I>(&mut self, names: I)
|
|
where
|
|
I: IntoIterator<Item = String>,
|
|
{
|
|
self.mcp_dependency_prompted.extend(names);
|
|
}
|
|
|
|
pub(crate) fn mcp_dependency_prompted(&self) -> HashSet<String> {
|
|
self.mcp_dependency_prompted.clone()
|
|
}
|
|
|
|
pub(crate) fn set_dependency_env(&mut self, values: HashMap<String, String>) {
|
|
for (key, value) in values {
|
|
self.dependency_env.insert(key, value);
|
|
}
|
|
}
|
|
|
|
pub(crate) fn dependency_env(&self) -> HashMap<String, String> {
|
|
self.dependency_env.clone()
|
|
}
|
|
|
|
pub(crate) fn set_session_startup_prewarm(
|
|
&mut self,
|
|
startup_prewarm: SessionStartupPrewarmHandle,
|
|
) {
|
|
self.startup_prewarm = Some(startup_prewarm);
|
|
}
|
|
|
|
pub(crate) fn take_session_startup_prewarm(&mut self) -> Option<SessionStartupPrewarmHandle> {
|
|
self.startup_prewarm.take()
|
|
}
|
|
|
|
// Adds connector IDs to the active set and returns the merged selection.
|
|
pub(crate) fn merge_connector_selection<I>(&mut self, connector_ids: I) -> HashSet<String>
|
|
where
|
|
I: IntoIterator<Item = String>,
|
|
{
|
|
self.active_connector_selection.extend(connector_ids);
|
|
self.active_connector_selection.clone()
|
|
}
|
|
|
|
// Returns the current connector selection tracked on session state.
|
|
pub(crate) fn get_connector_selection(&self) -> HashSet<String> {
|
|
self.active_connector_selection.clone()
|
|
}
|
|
|
|
// Removes all currently tracked connector selections.
|
|
pub(crate) fn clear_connector_selection(&mut self) {
|
|
self.active_connector_selection.clear();
|
|
}
|
|
|
|
pub(crate) fn set_pending_session_start_source(
|
|
&mut self,
|
|
value: Option<codex_hooks::SessionStartSource>,
|
|
) {
|
|
self.pending_session_start_source = value;
|
|
}
|
|
|
|
pub(crate) fn take_pending_session_start_source(
|
|
&mut self,
|
|
) -> Option<codex_hooks::SessionStartSource> {
|
|
self.pending_session_start_source.take()
|
|
}
|
|
|
|
pub(crate) fn record_granted_permissions(&mut self, permissions: PermissionProfile) {
|
|
self.granted_permissions =
|
|
merge_permission_profiles(self.granted_permissions.as_ref(), Some(&permissions));
|
|
}
|
|
|
|
pub(crate) fn granted_permissions(&self) -> Option<PermissionProfile> {
|
|
self.granted_permissions.clone()
|
|
}
|
|
}
|
|
|
|
// Sometimes new snapshots don't include credits or plan information.
|
|
// Preserve those from the previous snapshot when missing. For `limit_id`, treat
|
|
// missing values as the default `"codex"` bucket.
|
|
fn merge_rate_limit_fields(
|
|
previous: Option<&RateLimitSnapshot>,
|
|
mut snapshot: RateLimitSnapshot,
|
|
) -> RateLimitSnapshot {
|
|
if snapshot.limit_id.is_none() {
|
|
snapshot.limit_id = Some("codex".to_string());
|
|
}
|
|
if snapshot.credits.is_none() {
|
|
snapshot.credits = previous.and_then(|prior| prior.credits.clone());
|
|
}
|
|
if snapshot.plan_type.is_none() {
|
|
snapshot.plan_type = previous.and_then(|prior| prior.plan_type);
|
|
}
|
|
snapshot
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[path = "session_tests.rs"]
|
|
mod tests;
|