diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 634b324fa..4972ac2a8 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -63,7 +63,7 @@ Use the thread APIs to create, list, or archive conversations. Drive a conversat - Initialize once per connection: Immediately after opening a transport connection, send an `initialize` request with your client metadata, then emit an `initialized` notification. Any other request on that connection before this handshake gets rejected. - Start (or resume) a thread: Call `thread/start` to open a fresh conversation. The response returns the thread object and you’ll also get a `thread/started` notification. If you’re continuing an existing conversation, call `thread/resume` with its ID instead. If you want to branch from an existing conversation, call `thread/fork` to create a new thread id with copied history. The returned `thread.ephemeral` flag tells you whether the session is intentionally in-memory only; when it is `true`, `thread.path` is `null`. -- Begin a turn: To send user input, call `turn/start` with the target `threadId` and the user's input. Optional fields let you override model, cwd, sandbox policy, etc. This immediately returns the new turn object and triggers a `turn/started` notification. +- Begin a turn: To send user input, call `turn/start` with the target `threadId` and the user's input. Optional fields let you override model, cwd, sandbox policy, etc. This immediately returns the new turn object. The app-server emits `turn/started` when that turn actually begins running. - Stream events: After `turn/start`, keep reading JSON-RPC notifications on stdout. You’ll see `item/started`, `item/completed`, deltas like `item/agentMessage/delta`, tool progress, etc. These represent streaming model output plus any side effects (commands, tool calls, reasoning notes). - Finish the turn: When the model is done (or the turn is interrupted via making the `turn/interrupt` call), the server sends `turn/completed` with the final turn state and token usage. @@ -620,7 +620,7 @@ Because audio is intentionally separate from `ThreadItem`, clients can opt out o ### Turn events -The app-server streams JSON-RPC notifications while a turn is running. Each turn starts with `turn/started` (initial `turn`) and ends with `turn/completed` (final `turn` status). Token usage events stream separately via `thread/tokenUsage/updated`. Clients subscribe to the events they care about, rendering each item incrementally as updates arrive. The per-item lifecycle is always: `item/started` → zero or more item-specific deltas → `item/completed`. +The app-server streams JSON-RPC notifications while a turn is running. Each turn emits `turn/started` when it begins running and ends with `turn/completed` (final `turn` status). Token usage events stream separately via `thread/tokenUsage/updated`. Clients subscribe to the events they care about, rendering each item incrementally as updates arrive. The per-item lifecycle is always: `item/started` → zero or more item-specific deltas → `item/completed`. - `turn/started` — `{ turn }` with the turn id, empty `items`, and `status: "inProgress"`. - `turn/completed` — `{ turn }` where `turn.status` is `completed`, `interrupted`, or `failed`; failures carry `{ error: { message, codexErrorInfo?, additionalDetails? } }`. diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index f23a1ec09..2f293c934 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -83,6 +83,7 @@ use codex_app_server_protocol::TurnError; use codex_app_server_protocol::TurnInterruptResponse; use codex_app_server_protocol::TurnPlanStep; use codex_app_server_protocol::TurnPlanUpdatedNotification; +use codex_app_server_protocol::TurnStartedNotification; use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::build_turns_from_rollout_items; use codex_app_server_protocol::convert_patch_changes; @@ -185,12 +186,30 @@ pub(crate) async fn apply_bespoke_event_handling( msg, } = event; match msg { - EventMsg::TurnStarted(_) => { + EventMsg::TurnStarted(payload) => { // While not technically necessary as it was already done on TurnComplete, be extra cautios and abort any pending server requests. outgoing.abort_pending_server_requests().await; thread_watch_manager .note_turn_started(&conversation_id.to_string()) .await; + if let ApiVersion::V2 = api_version { + let turn = { + let state = thread_state.lock().await; + state.active_turn_snapshot().unwrap_or_else(|| Turn { + id: payload.turn_id.clone(), + items: Vec::new(), + error: None, + status: TurnStatus::InProgress, + }) + }; + let notification = TurnStartedNotification { + thread_id: conversation_id.to_string(), + turn, + }; + outgoing + .send_server_notification(ServerNotification::TurnStarted(notification)) + .await; + } } EventMsg::TurnComplete(_ev) => { // All per-thread requests are bound to a turn, so abort them. diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 79513a5be..42f2ad4cb 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -173,7 +173,6 @@ use codex_app_server_protocol::Turn; use codex_app_server_protocol::TurnInterruptParams; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; -use codex_app_server_protocol::TurnStartedNotification; use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::TurnSteerParams; use codex_app_server_protocol::TurnSteerResponse; @@ -375,7 +374,6 @@ pub(crate) struct CodexMessageProcessor { outgoing: Arc, arg0_paths: Arg0DispatchPaths, config: Arc, - single_client_mode: bool, cli_overrides: Vec<(String, TomlValue)>, cloud_requirements: Arc>, active_login: Arc>>, @@ -402,7 +400,6 @@ struct ListenerTaskContext { thread_watch_manager: ThreadWatchManager, fallback_model_provider: String, codex_home: PathBuf, - single_client_mode: bool, } #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -419,7 +416,6 @@ pub(crate) struct CodexMessageProcessorArgs { pub(crate) config: Arc, pub(crate) cli_overrides: Vec<(String, TomlValue)>, pub(crate) cloud_requirements: Arc>, - pub(crate) single_client_mode: bool, pub(crate) feedback: CodexFeedback, } @@ -464,7 +460,6 @@ impl CodexMessageProcessor { config, cli_overrides, cloud_requirements, - single_client_mode, feedback, } = args; Self { @@ -473,7 +468,6 @@ impl CodexMessageProcessor { outgoing: outgoing.clone(), arg0_paths, config, - single_client_mode, cli_overrides, cloud_requirements, active_login: Arc::new(Mutex::new(None)), @@ -2082,7 +2076,6 @@ impl CodexMessageProcessor { thread_watch_manager: self.thread_watch_manager.clone(), fallback_model_provider: self.config.model_provider_id.clone(), codex_home: self.config.codex_home.clone(), - single_client_mode: self.single_client_mode, }; tokio::spawn(async move { @@ -5910,17 +5903,8 @@ impl CodexMessageProcessor { status: TurnStatus::InProgress, }; - let response = TurnStartResponse { turn: turn.clone() }; + let response = TurnStartResponse { turn }; self.outgoing.send_response(request_id, response).await; - - // Emit v2 turn/started notification. - let notif = TurnStartedNotification { - thread_id: params.thread_id, - turn, - }; - self.outgoing - .send_server_notification(ServerNotification::TurnStarted(notif)) - .await; } Err(err) => { let error = JSONRPCErrorError { @@ -6211,24 +6195,15 @@ impl CodexMessageProcessor { &self, request_id: &ConnectionRequestId, turn: Turn, - parent_thread_id: String, review_thread_id: String, ) { let response = ReviewStartResponse { - turn: turn.clone(), + turn, review_thread_id, }; self.outgoing .send_response(request_id.clone(), response) .await; - - let notif = TurnStartedNotification { - thread_id: parent_thread_id, - turn, - }; - self.outgoing - .send_server_notification(ServerNotification::TurnStarted(notif)) - .await; } async fn start_inline_review( @@ -6244,13 +6219,8 @@ impl CodexMessageProcessor { match turn_id { Ok(turn_id) => { let turn = Self::build_review_turn(turn_id, display_text); - self.emit_review_started( - request_id, - turn, - parent_thread_id.clone(), - parent_thread_id, - ) - .await; + self.emit_review_started(request_id, turn, parent_thread_id) + .await; Ok(()) } Err(err) => Err(JSONRPCErrorError { @@ -6364,7 +6334,7 @@ impl CodexMessageProcessor { let turn = Self::build_review_turn(turn_id, display_text); let review_thread_id = thread_id.to_string(); - self.emit_review_started(request_id, turn, review_thread_id.clone(), review_thread_id) + self.emit_review_started(request_id, turn, review_thread_id) .await; Ok(()) @@ -6540,7 +6510,6 @@ impl CodexMessageProcessor { thread_watch_manager: self.thread_watch_manager.clone(), fallback_model_provider: self.config.model_provider_id.clone(), codex_home: self.config.codex_home.clone(), - single_client_mode: self.single_client_mode, }, conversation_id, connection_id, @@ -6628,7 +6597,6 @@ impl CodexMessageProcessor { thread_watch_manager: self.thread_watch_manager.clone(), fallback_model_provider: self.config.model_provider_id.clone(), codex_home: self.config.codex_home.clone(), - single_client_mode: self.single_client_mode, }, conversation_id, conversation, @@ -6660,7 +6628,6 @@ impl CodexMessageProcessor { thread_watch_manager, fallback_model_provider, codex_home, - single_client_mode, } = listener_task_context; let outgoing_for_task = Arc::clone(&outgoing); tokio::spawn(async move { @@ -6710,9 +6677,7 @@ impl CodexMessageProcessor { ); let raw_events_enabled = { let mut thread_state = thread_state.lock().await; - if !single_client_mode { - thread_state.track_current_turn_event(&event.msg); - } + thread_state.track_current_turn_event(&event.msg); thread_state.experimental_raw_events }; let subscribed_connection_ids = thread_state_manager diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index fec663b48..a84712ea0 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -558,7 +558,6 @@ pub async fn run_main_with_transport( outgoing: outgoing_message_sender, arg0_paths, config: Arc::new(config), - single_client_mode, cli_overrides, loader_overrides, cloud_requirements: cloud_requirements.clone(), diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 9c22e56ed..48f2f69e8 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -150,7 +150,6 @@ pub(crate) struct MessageProcessorArgs { pub(crate) outgoing: Arc, pub(crate) arg0_paths: Arg0DispatchPaths, pub(crate) config: Arc, - pub(crate) single_client_mode: bool, pub(crate) cli_overrides: Vec<(String, TomlValue)>, pub(crate) loader_overrides: LoaderOverrides, pub(crate) cloud_requirements: CloudRequirementsLoader, @@ -166,7 +165,6 @@ impl MessageProcessor { outgoing, arg0_paths, config, - single_client_mode, cli_overrides, loader_overrides, cloud_requirements, @@ -202,7 +200,6 @@ impl MessageProcessor { config: Arc::clone(&config), cli_overrides: cli_overrides.clone(), cloud_requirements: cloud_requirements.clone(), - single_client_mode, feedback, }); let config_api = ConfigApi::new( diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 7ae56761c..9628a7816 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -1,4 +1,3 @@ -use anyhow::Context; use anyhow::Result; use app_test_support::McpProcess; use app_test_support::create_apply_patch_sse_response; @@ -17,7 +16,6 @@ use codex_app_server_protocol::FileChangeApprovalDecision; use codex_app_server_protocol::FileChangeRequestApprovalResponse; use codex_app_server_protocol::ItemStartedNotification; use codex_app_server_protocol::JSONRPCError; -use codex_app_server_protocol::JSONRPCNotification; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::PatchApplyStatus; use codex_app_server_protocol::PatchChangeKind; @@ -30,7 +28,6 @@ use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus; -use codex_app_server_protocol::ThreadStatusChangedNotification; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::TurnStatus; @@ -293,7 +290,7 @@ async fn thread_resume_keeps_in_flight_turn_streaming() -> Result<()> { .await??; timeout( DEFAULT_READ_TIMEOUT, - wait_for_thread_status_active(&mut primary, &thread.id), + primary.read_stream_until_notification_message("turn/started"), ) .await??; @@ -400,7 +397,7 @@ async fn thread_resume_rejects_history_when_thread_is_running() -> Result<()> { to_response::(running_turn_resp)?; timeout( DEFAULT_READ_TIMEOUT, - wait_for_thread_status_active(&mut primary, &thread_id), + primary.read_stream_until_notification_message("turn/started"), ) .await??; @@ -516,7 +513,7 @@ async fn thread_resume_rejects_mismatched_path_when_thread_is_running() -> Resul to_response::(running_turn_resp)?; timeout( DEFAULT_READ_TIMEOUT, - wait_for_thread_status_active(&mut primary, &thread_id), + primary.read_stream_until_notification_message("turn/started"), ) .await??; @@ -619,7 +616,7 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R .await??; timeout( DEFAULT_READ_TIMEOUT, - wait_for_thread_status_active(&mut primary, &thread.id), + primary.read_stream_until_notification_message("turn/started"), ) .await??; @@ -1419,30 +1416,6 @@ required = true ) } -async fn wait_for_thread_status_active( - mcp: &mut McpProcess, - thread_id: &str, -) -> Result { - loop { - let status_changed_notif: JSONRPCNotification = mcp - .read_stream_until_notification_message("thread/status/changed") - .await?; - let status_changed_params = status_changed_notif - .params - .context("thread/status/changed params must be present")?; - let status_changed: ThreadStatusChangedNotification = - serde_json::from_value(status_changed_params)?; - if status_changed.thread_id == thread_id - && status_changed.status - == (ThreadStatus::Active { - active_flags: Vec::new(), - }) - { - return Ok(status_changed); - } - } -} - #[allow(dead_code)] fn set_rollout_mtime(path: &Path, updated_at_rfc3339: &str) -> Result<()> { let parsed = chrono::DateTime::parse_from_rfc3339(updated_at_rfc3339)?.with_timezone(&Utc); diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index 44063a3ca..472c1a96b 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -434,6 +434,21 @@ async fn turn_start_emits_notifications_and_accepts_model_override() -> Result<( started.turn.status, codex_app_server_protocol::TurnStatus::InProgress ); + assert_eq!(started.turn.id, turn.id); + + let completed_notif: JSONRPCNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + let completed: TurnCompletedNotification = serde_json::from_value( + completed_notif + .params + .expect("turn/completed params must be present"), + )?; + assert_eq!(completed.thread_id, thread.id); + assert_eq!(completed.turn.id, turn.id); + assert_eq!(completed.turn.status, TurnStatus::Completed); // Send a second turn that exercises the overrides path: change the model. let turn_req2 = mcp @@ -457,25 +472,30 @@ async fn turn_start_emits_notifications_and_accepts_model_override() -> Result<( // Ensure the second turn has a different id than the first. assert_ne!(turn.id, turn2.id); - // Expect a second turn/started notification as well. - let _notif2: JSONRPCNotification = timeout( + let notif2: JSONRPCNotification = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_notification_message("turn/started"), ) .await??; + let started2: TurnStartedNotification = + serde_json::from_value(notif2.params.expect("params must be present"))?; + assert_eq!(started2.thread_id, thread.id); + assert_eq!(started2.turn.id, turn2.id); + assert_eq!(started2.turn.status, TurnStatus::InProgress); - let completed_notif: JSONRPCNotification = timeout( + let completed_notif2: JSONRPCNotification = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_notification_message("turn/completed"), ) .await??; - let completed: TurnCompletedNotification = serde_json::from_value( - completed_notif + let completed2: TurnCompletedNotification = serde_json::from_value( + completed_notif2 .params .expect("turn/completed params must be present"), )?; - assert_eq!(completed.thread_id, thread.id); - assert_eq!(completed.turn.status, TurnStatus::Completed); + assert_eq!(completed2.thread_id, thread.id); + assert_eq!(completed2.turn.id, turn2.id); + assert_eq!(completed2.turn.status, TurnStatus::Completed); Ok(()) }