openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-07 20:30:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

shift towards app-server stream consumption

Browse Source
This commit is contained in:
Roy Han
2026-03-27 10:25:52 -07:00
parent 9ae940f704
commit b8615d471f
10 changed files with 910 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/bespoke_event_handling.rs
+54 -5
View File
@@ -12,6 +12,7 @@ use crate::thread_state::ThreadState;
use crate::thread_state::TurnSummary;
use crate::thread_status::ThreadWatchActiveGuard;
use crate::thread_status::ThreadWatchManager;
use codex_analytics::AnalyticsEventsClient;
use codex_app_server_protocol::AccountRateLimitsUpdatedNotification;
use codex_app_server_protocol::AdditionalPermissionProfile as V2AdditionalPermissionProfile;
use codex_app_server_protocol::AgentMessageDeltaNotification;
@@ -257,6 +258,7 @@ pub(crate) async fn apply_bespoke_event_handling(
conversation_id: ThreadId,
conversation: Arc<CodexThread>,
thread_manager: Arc<ThreadManager>,
analytics_events_client: AnalyticsEventsClient,
outgoing: ThreadScopedOutgoingMessageSender,
thread_state: Arc<tokio::sync::Mutex<ThreadState>>,
thread_watch_manager: ThreadWatchManager,
@@ -289,6 +291,8 @@ pub(crate) async fn apply_bespoke_event_handling(
thread_id: conversation_id.to_string(),
turn,
};
analytics_events_client
.track_notification(ServerNotification::TurnStarted(notification.clone()));
outgoing
.send_server_notification(ServerNotification::TurnStarted(notification))
.await;
@@ -301,7 +305,14 @@ pub(crate) async fn apply_bespoke_event_handling(
thread_watch_manager
.note_turn_completed(&conversation_id.to_string(), turn_failed)
.await;
handle_turn_complete(conversation_id, event_turn_id, &outgoing, &thread_state).await;
handle_turn_complete(
conversation_id,
event_turn_id,
Some(&analytics_events_client),
&outgoing,
&thread_state,
)
.await;
}
EventMsg::SkillsUpdateAvailable => {
if let ApiVersion::V2 = api_version {
@@ -1946,6 +1957,7 @@ async fn emit_turn_completed_with_status(
event_turn_id: String,
status: TurnStatus,
error: Option<TurnError>,
analytics_events_client: Option<&AnalyticsEventsClient>,
outgoing: &ThreadScopedOutgoingMessageSender,
) {
let notification = TurnCompletedNotification {
@@ -1957,6 +1969,10 @@ async fn emit_turn_completed_with_status(
status,
},
};
if let Some(analytics_events_client) = analytics_events_client {
analytics_events_client
.track_notification(ServerNotification::TurnCompleted(notification.clone()));
}
outgoing
.send_server_notification(ServerNotification::TurnCompleted(notification))
.await;
@@ -2100,6 +2116,7 @@ async fn find_and_remove_turn_summary(
async fn handle_turn_complete(
conversation_id: ThreadId,
event_turn_id: String,
analytics_events_client: Option<&AnalyticsEventsClient>,
outgoing: &ThreadScopedOutgoingMessageSender,
thread_state: &Arc<Mutex<ThreadState>>,
) {
@@ -2110,7 +2127,15 @@ async fn handle_turn_complete(
None => (TurnStatus::Completed, None),
};
emit_turn_completed_with_status(conversation_id, event_turn_id, status, error, outgoing).await;
emit_turn_completed_with_status(
conversation_id,
event_turn_id,
status,
error,
analytics_events_client,
outgoing,
)
.await;
}
async fn handle_turn_interrupted(
@@ -2126,6 +2151,7 @@ async fn handle_turn_interrupted(
event_turn_id,
TurnStatus::Interrupted,
/*error*/ None,
None,
outgoing,
)
.await;
@@ -3280,6 +3306,7 @@ mod tests {
handle_turn_complete(
conversation_id,
event_turn_id.clone(),
None,
&outgoing,
&thread_state,
)
@@ -3368,6 +3395,7 @@ mod tests {
handle_turn_complete(
conversation_id,
event_turn_id.clone(),
None,
&outgoing,
&thread_state,
)
@@ -3628,7 +3656,14 @@ mod tests {
&thread_state,
)
.await;
handle_turn_complete(conversation_a, a_turn1.clone(), &outgoing, &thread_state).await;
handle_turn_complete(
conversation_a,
a_turn1.clone(),
None,
&outgoing,
&thread_state,
)
.await;
// Turn 1 on conversation B
let b_turn1 = "b_turn1".to_string();
@@ -3642,11 +3677,25 @@ mod tests {
&thread_state,
)
.await;
handle_turn_complete(conversation_b, b_turn1.clone(), &outgoing, &thread_state).await;
handle_turn_complete(
conversation_b,
b_turn1.clone(),
None,
&outgoing,
&thread_state,
)
.await;
// Turn 2 on conversation A
let a_turn2 = "a_turn2".to_string();
handle_turn_complete(conversation_a, a_turn2.clone(), &outgoing, &thread_state).await;
handle_turn_complete(
conversation_a,
a_turn2.clone(),
None,
&outgoing,
&thread_state,
)
.await;
// Verify: A turn 1
let msg = recv_broadcast_message(&mut rx).await?;
codex-rs/app-server/src/codex_message_processor.rs
+19 -1
View File
@@ -6179,6 +6179,16 @@ impl CodexMessageProcessor {
return;
}
let analytics_turn_start_params = params.clone();
self.analytics_events_client.track_request(
request_id.connection_id.0,
request_id.request_id.clone(),
ClientRequest::TurnStart {
request_id: request_id.request_id.clone(),
params: analytics_turn_start_params,
},
);
let collaboration_modes_config = CollaborationModesConfig {
default_mode_request_user_input: thread.enabled(Feature::DefaultModeRequestUserInput),
};
@@ -6254,6 +6264,13 @@ impl CodexMessageProcessor {
};
let response = TurnStartResponse { turn };
self.analytics_events_client.track_response(
request_id.connection_id.0,
ClientResponse::TurnStart {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
self.outgoing.send_response(request_id, response).await;
}
Err(err) => {
@@ -6976,7 +6993,7 @@ impl CodexMessageProcessor {
outgoing,
thread_manager,
thread_state_manager,
analytics_events_client: _,
analytics_events_client,
thread_watch_manager,
fallback_model_provider,
codex_home,
@@ -7023,6 +7040,7 @@ impl CodexMessageProcessor {
conversation_id,
conversation.clone(),
thread_manager.clone(),
analytics_events_client.clone(),
thread_outgoing,
thread_state.clone(),
thread_watch_manager.clone(),