From 8ce48f9968fc3b8d6d16177d3d5deb0b6cb0676d Mon Sep 17 00:00:00 2001 From: Matthew Zeng Date: Wed, 29 Apr 2026 13:41:12 -0700 Subject: [PATCH 01/56] [tool_suggest] Improve tool_suggest triggering conditions. (#20091) ## Summary - Tighten `tool_suggest` guidance so it prefers explicit plugin install requests, while still allowing a connector install when the relevant plugin is already installed and a needed connector from that plugin is missing. - Tell the model not to call `tool_suggest` in parallel with other tools. ## Testing - `cargo test -p codex-tools tool_suggest` - `cargo test -p codex-core tool_suggest` --- .../search_tool/tool_suggest_description.md | 33 ++++++++-------- codex-rs/core/tests/suite/tool_suggest.rs | 13 ++++--- codex-rs/tools/src/tool_discovery.rs | 15 ++------ codex-rs/tools/src/tool_discovery_tests.rs | 32 ++++++++++++++-- .../tools/src/tool_registry_plan_tests.rs | 38 +++++++++++++------ 5 files changed, 83 insertions(+), 48 deletions(-) diff --git a/codex-rs/core/templates/search_tool/tool_suggest_description.md b/codex-rs/core/templates/search_tool/tool_suggest_description.md index ad0a50fcc1..9bed2d9d7b 100644 --- a/codex-rs/core/templates/search_tool/tool_suggest_description.md +++ b/codex-rs/core/templates/search_tool/tool_suggest_description.md @@ -1,26 +1,29 @@ # Tool suggestion discovery -Suggests a missing connector in an installed plugin, or in narrower cases a not installed but discoverable plugin, when the user clearly wants a capability that is not currently available in the active `tools` list. +Use this tool only to ask the user to install one known plugin or connector from the list below. The list contains known candidates that are not currently installed. -Use this ONLY when: -- You've already tried to find a matching available tool for the user's request but couldn't find a good match. This includes `tool_search` (if available) and other means. -- For connectors/apps that are not installed but needed for an installed plugin, suggest to install them if the task requirements match precisely. -- For plugins that are not installed but discoverable, only suggest discoverable and installable plugins when the user's intent very explicitly and unambiguously matches that plugin itself. Do not suggest a plugin just because one of its connectors or capabilities seems relevant. +Use this ONLY when all of the following are true: +- The user explicitly wants a specific plugin or connector that is not already available in the current context or active `tools` list. +- `tool_search` is not available, or it has already been called and did not find or make the requested tool callable. +- The tool is one of the known installable plugins or connectors listed below. Only ask to install tools from this list. -Tool suggestions should only use the discoverable tools listed here. DO NOT explore or recommend tools that are not on this list. +Do not use tool suggestion for adjacent capabilities, broad recommendations, or tools that merely seem useful. The user's intent must clearly match one listed tool. -Discoverable tools: +Known plugins/connectors available to install: {{discoverable_tools}} Workflow: -1. Ensure all possible means have been exhausted to find an existing available tool but none of them matches the request intent. -2. Match the user's request against the discoverable tools list above. Apply the stricter explicit-and-unambiguous rule for *discoverable tools* like plugin install suggestions; *missing tools* like connector install suggestions continue to use the normal clear-fit standard. -3. If one tool clearly fits, call `tool_suggest` with: +1. Check the current context and active `tools` list first. If `tool_search` is available, call `tool_search` before calling `tool_suggest`. Do not use tool suggestion if the needed tool is already available, found through `tool_search`, or callable after discovery. +2. Match the user's explicit request against the known plugin/connector list above. Only proceed when one listed plugin or connector exactly fits. +3. If we found both connectors and plugins to suggest, use plugins first, only use connectors if the corresponding plugin is installed but the connector is not. +4. If one tool clearly fits, call `tool_suggest` with: - `tool_type`: `connector` or `plugin` - - `action_type`: `install` or `enable` - - `tool_id`: exact id from the discoverable tools list above + - `action_type`: `install` + - `tool_id`: exact id from the known plugin/connector list above - `suggest_reason`: concise one-line user-facing reason this tool can help with the current request -4. After the suggestion flow completes: - - if the user finished the install or enable flow, continue by searching again or using the newly available tool - - if the user did not finish, continue without that tool, and don't suggest that tool again unless the user explicitly asks you to. +5. After the suggestion flow completes: + - if the user finished the install flow, continue by searching again or using the newly available tool + - if the user did not finish, continue without that tool, and don't suggest that tool again unless the user explicitly asks for it. + +IMPORTANT: DO NOT call this tool in parallel with other tools. diff --git a/codex-rs/core/tests/suite/tool_suggest.rs b/codex-rs/core/tests/suite/tool_suggest.rs index b5b3752e0c..6cb19d01a5 100644 --- a/codex-rs/core/tests/suite/tool_suggest.rs +++ b/codex-rs/core/tests/suite/tool_suggest.rs @@ -131,12 +131,13 @@ async fn tool_suggest_is_available_without_search_tool_after_discovery_attempts( let description = function_tool_description(&body, TOOL_SUGGEST_TOOL_NAME).expect("description"); - assert!( - description.contains( - "You've already tried to find a matching available tool for the user's request" - ) - ); - assert!(description.contains("This includes `tool_search` (if available) and other means.")); + assert!(description.contains( + "Use this tool only to ask the user to install one known plugin or connector from the list below" + )); + assert!(description.contains( + "`tool_search` is not available, or it has already been called and did not find or make the requested tool callable." + )); + assert!(description.contains("IMPORTANT: DO NOT call this tool in parallel with other tools.")); assert!(!description.contains("tool_search fails to find a good match")); Ok(()) diff --git a/codex-rs/tools/src/tool_discovery.rs b/codex-rs/tools/src/tool_discovery.rs index ddcd2f1ce3..74977dce38 100644 --- a/codex-rs/tools/src/tool_discovery.rs +++ b/codex-rs/tools/src/tool_discovery.rs @@ -272,11 +272,6 @@ pub fn collect_tool_search_source_infos<'a>( } pub fn create_tool_suggest_tool(discoverable_tools: &[ToolSuggestEntry]) -> ToolSpec { - let discoverable_tool_ids = discoverable_tools - .iter() - .map(|tool| tool.id.as_str()) - .collect::>() - .join(", "); let properties = BTreeMap::from([ ( "tool_type".to_string(), @@ -287,15 +282,11 @@ pub fn create_tool_suggest_tool(discoverable_tools: &[ToolSuggestEntry]) -> Tool ), ( "action_type".to_string(), - JsonSchema::string(Some( - "Suggested action for the tool. Use \"install\" or \"enable\".".to_string(), - )), + JsonSchema::string(Some("Suggested action for the tool. Use \"install\".".to_string())), ), ( "tool_id".to_string(), - JsonSchema::string(Some(format!( - "Connector or plugin id to suggest. Must be one of: {discoverable_tool_ids}." - ))), + JsonSchema::string(Some("Connector or plugin id to suggest.".to_string())), ), ( "suggest_reason".to_string(), @@ -308,7 +299,7 @@ pub fn create_tool_suggest_tool(discoverable_tools: &[ToolSuggestEntry]) -> Tool let discoverable_tools = format_discoverable_tools(discoverable_tools); let description = format!( - "# Tool suggestion discovery\n\nSuggests a missing connector in an installed plugin, or in narrower cases a not installed but discoverable plugin, when the user clearly wants a capability that is not currently available in the active `tools` list.\n\nUse this ONLY when:\n- You've already tried to find a matching available tool for the user's request but couldn't find a good match. This includes `{TOOL_SEARCH_TOOL_NAME}` (if available) and other means.\n- For connectors/apps that are not installed but needed for an installed plugin, suggest to install them if the task requirements match precisely.\n- For plugins that are not installed but discoverable, only suggest discoverable and installable plugins when the user's intent very explicitly and unambiguously matches that plugin itself. Do not suggest a plugin just because one of its connectors or capabilities seems relevant.\n\nTool suggestions should only use the discoverable tools listed here. DO NOT explore or recommend tools that are not on this list.\n\nDiscoverable tools:\n{discoverable_tools}\n\nWorkflow:\n\n1. Ensure all possible means have been exhausted to find an existing available tool but none of them matches the request intent.\n2. Match the user's request against the discoverable tools list above. Apply the stricter explicit-and-unambiguous rule for *discoverable tools* like plugin install suggestions; *missing tools* like connector install suggestions continue to use the normal clear-fit standard.\n3. If one tool clearly fits, call `{TOOL_SUGGEST_TOOL_NAME}` with:\n - `tool_type`: `connector` or `plugin`\n - `action_type`: `install` or `enable`\n - `tool_id`: exact id from the discoverable tools list above\n - `suggest_reason`: concise one-line user-facing reason this tool can help with the current request\n4. After the suggestion flow completes:\n - if the user finished the install or enable flow, continue by searching again or using the newly available tool\n - if the user did not finish, continue without that tool, and don't suggest that tool again unless the user explicitly asks for it." + "# Tool suggestion discovery\n\nUse this tool only to ask the user to install one known plugin or connector from the list below. The list contains known candidates that are not currently installed.\n\nUse this ONLY when all of the following are true:\n- The user explicitly wants a specific plugin or connector that is not already available in the current context or active `tools` list.\n- `{TOOL_SEARCH_TOOL_NAME}` is not available, or it has already been called and did not find or make the requested tool callable.\n- The tool is one of the known installable plugins or connectors listed below. Only ask to install tools from this list.\n\nDo not use tool suggestion for adjacent capabilities, broad recommendations, or tools that merely seem useful. The user's intent must clearly match one listed tool.\n\nKnown plugins/connectors available to install:\n{discoverable_tools}\n\nWorkflow:\n\n1. Check the current context and active `tools` list first. If `{TOOL_SEARCH_TOOL_NAME}` is available, call `{TOOL_SEARCH_TOOL_NAME}` before calling `{TOOL_SUGGEST_TOOL_NAME}`. Do not use tool suggestion if the needed tool is already available, found through `{TOOL_SEARCH_TOOL_NAME}`, or callable after discovery.\n2. Match the user's explicit request against the known plugin/connector list above. Only proceed when one listed plugin or connector exactly fits.\n3. If we found both connectors and plugins to suggest, use plugins first, only use connectors if the corresponding plugin is installed but the connector is not.\n4. If one tool clearly fits, call `{TOOL_SUGGEST_TOOL_NAME}` with:\n - `tool_type`: `connector` or `plugin`\n - `action_type`: `install`\n - `tool_id`: exact id from the known plugin/connector list above\n - `suggest_reason`: concise one-line user-facing reason this tool can help with the current request\n5. After the suggestion flow completes:\n - if the user finished the install flow, continue by searching again or using the newly available tool\n - if the user did not finish, continue without that tool, and don't suggest that tool again unless the user explicitly asks for it.\n\nIMPORTANT: DO NOT call this tool in parallel with other tools." ); ToolSpec::Function(ResponsesApiTool { diff --git a/codex-rs/tools/src/tool_discovery_tests.rs b/codex-rs/tools/src/tool_discovery_tests.rs index 9afd32231e..9edbccffaa 100644 --- a/codex-rs/tools/src/tool_discovery_tests.rs +++ b/codex-rs/tools/src/tool_discovery_tests.rs @@ -50,6 +50,32 @@ fn create_tool_search_tool_deduplicates_and_renders_enabled_sources() { #[test] fn create_tool_suggest_tool_uses_plugin_summary_fallback() { + let expected_description = concat!( + "# Tool suggestion discovery\n\n", + "Use this tool only to ask the user to install one known plugin or connector from the list below. The list contains known candidates that are not currently installed.\n\n", + "Use this ONLY when all of the following are true:\n", + "- The user explicitly wants a specific plugin or connector that is not already available in the current context or active `tools` list.\n", + "- `tool_search` is not available, or it has already been called and did not find or make the requested tool callable.\n", + "- The tool is one of the known installable plugins or connectors listed below. Only ask to install tools from this list.\n\n", + "Do not use tool suggestion for adjacent capabilities, broad recommendations, or tools that merely seem useful. The user's intent must clearly match one listed tool.\n\n", + "Known plugins/connectors available to install:\n", + "- GitHub (id: `github`, type: plugin, action: install): skills; MCP servers: github-mcp; app connectors: github-app\n", + "- Slack (id: `slack@openai-curated`, type: connector, action: install): No description provided.\n\n", + "Workflow:\n\n", + "1. Check the current context and active `tools` list first. If `tool_search` is available, call `tool_search` before calling `tool_suggest`. Do not use tool suggestion if the needed tool is already available, found through `tool_search`, or callable after discovery.\n", + "2. Match the user's explicit request against the known plugin/connector list above. Only proceed when one listed plugin or connector exactly fits.\n", + "3. If we found both connectors and plugins to suggest, use plugins first, only use connectors if the corresponding plugin is installed but the connector is not.\n", + "4. If one tool clearly fits, call `tool_suggest` with:\n", + " - `tool_type`: `connector` or `plugin`\n", + " - `action_type`: `install`\n", + " - `tool_id`: exact id from the known plugin/connector list above\n", + " - `suggest_reason`: concise one-line user-facing reason this tool can help with the current request\n", + "5. After the suggestion flow completes:\n", + " - if the user finished the install flow, continue by searching again or using the newly available tool\n", + " - if the user did not finish, continue without that tool, and don't suggest that tool again unless the user explicitly asks for it.\n\n", + "IMPORTANT: DO NOT call this tool in parallel with other tools.", + ); + assert_eq!( create_tool_suggest_tool(&[ ToolSuggestEntry { @@ -73,14 +99,14 @@ fn create_tool_suggest_tool_uses_plugin_summary_fallback() { ]), ToolSpec::Function(ResponsesApiTool { name: "tool_suggest".to_string(), - description: "# Tool suggestion discovery\n\nSuggests a missing connector in an installed plugin, or in narrower cases a not installed but discoverable plugin, when the user clearly wants a capability that is not currently available in the active `tools` list.\n\nUse this ONLY when:\n- You've already tried to find a matching available tool for the user's request but couldn't find a good match. This includes `tool_search` (if available) and other means.\n- For connectors/apps that are not installed but needed for an installed plugin, suggest to install them if the task requirements match precisely.\n- For plugins that are not installed but discoverable, only suggest discoverable and installable plugins when the user's intent very explicitly and unambiguously matches that plugin itself. Do not suggest a plugin just because one of its connectors or capabilities seems relevant.\n\nTool suggestions should only use the discoverable tools listed here. DO NOT explore or recommend tools that are not on this list.\n\nDiscoverable tools:\n- GitHub (id: `github`, type: plugin, action: install): skills; MCP servers: github-mcp; app connectors: github-app\n- Slack (id: `slack@openai-curated`, type: connector, action: install): No description provided.\n\nWorkflow:\n\n1. Ensure all possible means have been exhausted to find an existing available tool but none of them matches the request intent.\n2. Match the user's request against the discoverable tools list above. Apply the stricter explicit-and-unambiguous rule for *discoverable tools* like plugin install suggestions; *missing tools* like connector install suggestions continue to use the normal clear-fit standard.\n3. If one tool clearly fits, call `tool_suggest` with:\n - `tool_type`: `connector` or `plugin`\n - `action_type`: `install` or `enable`\n - `tool_id`: exact id from the discoverable tools list above\n - `suggest_reason`: concise one-line user-facing reason this tool can help with the current request\n4. After the suggestion flow completes:\n - if the user finished the install or enable flow, continue by searching again or using the newly available tool\n - if the user did not finish, continue without that tool, and don't suggest that tool again unless the user explicitly asks for it.".to_string(), + description: expected_description.to_string(), strict: false, defer_loading: None, parameters: JsonSchema::object(BTreeMap::from([ ( "action_type".to_string(), JsonSchema::string(Some( - "Suggested action for the tool. Use \"install\" or \"enable\"." + "Suggested action for the tool. Use \"install\"." .to_string(), ),), ), @@ -94,7 +120,7 @@ fn create_tool_suggest_tool_uses_plugin_summary_fallback() { ( "tool_id".to_string(), JsonSchema::string(Some( - "Connector or plugin id to suggest. Must be one of: slack@openai-curated, github." + "Connector or plugin id to suggest." .to_string(), ),), ), diff --git a/codex-rs/tools/src/tool_registry_plan_tests.rs b/codex-rs/tools/src/tool_registry_plan_tests.rs index 4570034238..8d3ab5253c 100644 --- a/codex-rs/tools/src/tool_registry_plan_tests.rs +++ b/codex-rs/tools/src/tool_registry_plan_tests.rs @@ -1721,17 +1721,18 @@ fn tool_suggest_can_be_registered_without_search_tool() { ); assert_contains_tool_names(&tools, &[TOOL_SUGGEST_TOOL_NAME]); + let tool_suggest = find_tool(&tools, TOOL_SUGGEST_TOOL_NAME); + assert!(tool_suggest.supports_parallel_tool_calls); assert_lacks_tool_name(&tools, TOOL_SEARCH_TOOL_NAME); - let tool_suggest = find_tool(&tools, TOOL_SUGGEST_TOOL_NAME); let ToolSpec::Function(ResponsesApiTool { description, .. }) = &tool_suggest.spec else { panic!("expected function tool"); }; assert!(description.contains( - "Suggests a missing connector in an installed plugin, or in narrower cases a not installed but discoverable plugin" + "Use this tool only to ask the user to install one known plugin or connector from the list below. The list contains known candidates that are not currently installed." )); assert!(description.contains( - "You've already tried to find a matching available tool for the user's request but couldn't find a good match. This includes `tool_search` (if available) and other means." + "`tool_search` is not available, or it has already been called and did not find or make the requested tool callable." )); } @@ -1776,13 +1777,17 @@ fn tool_suggest_description_lists_discoverable_tools() { })), ]; - let (tools, _) = build_specs_with_discoverable_tools( + let (tools, handlers) = build_specs_with_discoverable_tools( &tools_config, /*mcp_tools*/ None, /*deferred_mcp_tools*/ None, Some(discoverable_tools), &[], ); + assert!(handlers.contains(&ToolHandlerSpec { + name: ToolName::plain(TOOL_SUGGEST_TOOL_NAME), + kind: ToolHandlerKind::ToolSuggest, + })); let tool_suggest = find_tool(&tools, TOOL_SUGGEST_TOOL_NAME); let ToolSpec::Function(ResponsesApiTool { @@ -1794,7 +1799,7 @@ fn tool_suggest_description_lists_discoverable_tools() { panic!("expected function tool"); }; assert!(description.contains( - "Suggests a missing connector in an installed plugin, or in narrower cases a not installed but discoverable plugin" + "Use this tool only to ask the user to install one known plugin or connector from the list below. The list contains known candidates that are not currently installed." )); assert!(description.contains("Google Calendar")); assert!(description.contains("Gmail")); @@ -1802,28 +1807,37 @@ fn tool_suggest_description_lists_discoverable_tools() { assert!(description.contains("Plan events and schedules.")); assert!(description.contains("Find and summarize email threads.")); assert!(description.contains("id: `sample@test`, type: plugin, action: install")); - assert!(description.contains("`action_type`: `install` or `enable`")); + assert!(description.contains("`action_type`: `install`")); assert!( description.contains("skills; MCP servers: sample-docs; app connectors: connector_sample") ); assert!( description.contains( - "You've already tried to find a matching available tool for the user's request but couldn't find a good match. This includes `tool_search` (if available) and other means." + "The user explicitly wants a specific plugin or connector that is not already available in the current context or active `tools` list." ) ); assert!(description.contains( - "For connectors/apps that are not installed but needed for an installed plugin, suggest to install them if the task requirements match precisely." + "`tool_search` is not available, or it has already been called and did not find or make the requested tool callable." )); assert!(description.contains( - "For plugins that are not installed but discoverable, only suggest discoverable and installable plugins when the user's intent very explicitly and unambiguously matches that plugin itself." + "The tool is one of the known installable plugins or connectors listed below. Only ask to install tools from this list." )); assert!(description.contains( - "Do not suggest a plugin just because one of its connectors or capabilities seems relevant." + "Do not use tool suggestion for adjacent capabilities, broad recommendations, or tools that merely seem useful." + )); + assert!(description.contains("IMPORTANT: DO NOT call this tool in parallel with other tools.")); + assert!(description.contains( + "Do not use tool suggestion if the needed tool is already available, found through `tool_search`, or callable after discovery." )); assert!(description.contains( - "Apply the stricter explicit-and-unambiguous rule for *discoverable tools* like plugin install suggestions; *missing tools* like connector install suggestions continue to use the normal clear-fit standard." + "If `tool_search` is available, call `tool_search` before calling `tool_suggest`." + )); + assert!(!description.contains("targeted lookup")); + assert!(!description.contains("broad or speculative searches")); + assert!(description.contains("Only proceed when one listed plugin or connector exactly fits.")); + assert!(description.contains( + "If we found both connectors and plugins to suggest, use plugins first, only use connectors if the corresponding plugin is installed but the connector is not." )); - assert!(description.contains("DO NOT explore or recommend tools that are not on this list.")); assert!(!description.contains("{{discoverable_tools}}")); assert!(!description.contains("tool_search fails to find a good match")); let (_, required) = expect_object_schema(parameters); From b15074d0a42e87465a39f78f033d8e3b74db6011 Mon Sep 17 00:00:00 2001 From: sayan-oai Date: Wed, 29 Apr 2026 13:47:20 -0700 Subject: [PATCH 02/56] app-server: fix outgoing sender test setup (#20258) ## Why [#17088](https://github.com/openai/codex/pull/17088) changed `OutgoingMessageSender::new` to require an `AnalyticsEventsClient`, but one `command_exec` test added earlier on `main` still called the old one-argument constructor. That leaves current `main` failing to compile in Bazel and argument-comment-lint jobs. ## What changed - Pass `AnalyticsEventsClient::disabled()` to the missed `OutgoingMessageSender::new` test call site in `command_exec.rs`. ## Verification - `cargo test -p codex-app-server timeout_or_cancellation_reports_cancellation_without_timeout_exit_code` --- codex-rs/app-server/src/command_exec.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/codex-rs/app-server/src/command_exec.rs b/codex-rs/app-server/src/command_exec.rs index 22ac92686e..699556dd5b 100644 --- a/codex-rs/app-server/src/command_exec.rs +++ b/codex-rs/app-server/src/command_exec.rs @@ -890,7 +890,10 @@ mod tests { manager .start(StartCommandExecParams { - outgoing: Arc::new(OutgoingMessageSender::new(tx)), + outgoing: Arc::new(OutgoingMessageSender::new( + tx, + codex_analytics::AnalyticsEventsClient::disabled(), + )), request_id: request_id.clone(), process_id: Some("proc-101".to_string()), exec_request: ExecRequest::new( From 973c5c823ed1b8ae76ab5a677c5864f624a2fafb Mon Sep 17 00:00:00 2001 From: rhan-oai Date: Wed, 29 Apr 2026 13:50:47 -0700 Subject: [PATCH 03/56] [app-server] type client response payloads (#20050) ## Why `pr17088` adds typed server-originated request/response plumbing, but successful client responses are still erased into bare JSON-RPC `result` values before app-server can make any typed decision about them. This precursor PR keeps successful client responses typed until the outgoing response seam. It is intentionally limited to protocol/app-server plumbing so the analytics behavior change can review separately on top. ## What changed - Add `ClientResponsePayload` as the pre-serialization client response body type. - Route app-server successful response paths through the typed payload seam while preserving existing handler-local analytics behavior. - Keep `InterruptConversation` JSON-RPC-only because it has no `ClientResponse` variant. - Move the new payload conversion tests into a dedicated protocol test module. ## Verification - `cargo check -p codex-app-server` - `cargo test -p codex-app-server-protocol` --- codex-rs/analytics/src/client.rs | 19 ++- codex-rs/analytics/src/client_tests.rs | 79 +++++++++++++ .../src/protocol/common.rs | 110 ++++++++++++++++++ .../src/protocol/common_tests.rs | 44 +++++++ .../app-server/src/codex_message_processor.rs | 11 +- codex-rs/app-server/src/message_processor.rs | 45 ++++--- codex-rs/app-server/src/outgoing_message.rs | 57 ++++++--- 7 files changed, 323 insertions(+), 42 deletions(-) create mode 100644 codex-rs/analytics/src/client_tests.rs create mode 100644 codex-rs/app-server-protocol/src/protocol/common_tests.rs diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index b149614353..2ff4c794ed 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -186,11 +186,22 @@ impl AnalyticsEventsClient { ))); } - pub fn track_request(&self, connection_id: u64, request_id: RequestId, request: ClientRequest) { + pub fn track_request( + &self, + connection_id: u64, + request_id: RequestId, + request: &ClientRequest, + ) { + if !matches!( + request, + ClientRequest::TurnStart { .. } | ClientRequest::TurnSteer { .. } + ) { + return; + } self.record_fact(AnalyticsFact::ClientRequest { connection_id, request_id, - request: Box::new(request), + request: Box::new(request.clone()), }); } @@ -324,6 +335,10 @@ impl AnalyticsEventsClient { } } +#[cfg(test)] +#[path = "client_tests.rs"] +mod tests; + async fn send_track_events( auth_manager: &AuthManager, base_url: &str, diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs new file mode 100644 index 0000000000..57679f24a1 --- /dev/null +++ b/codex-rs/analytics/src/client_tests.rs @@ -0,0 +1,79 @@ +use super::AnalyticsEventsClient; +use super::AnalyticsEventsQueue; +use crate::facts::AnalyticsFact; +use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadArchiveParams; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnSteerParams; +use std::collections::HashSet; +use std::sync::Arc; +use std::sync::Mutex; +use tokio::sync::mpsc; +use tokio::sync::mpsc::error::TryRecvError; + +fn client_with_receiver() -> (AnalyticsEventsClient, mpsc::Receiver) { + let (sender, receiver) = mpsc::channel(4); + let queue = AnalyticsEventsQueue { + sender, + app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), + plugin_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), + }; + (AnalyticsEventsClient { queue: Some(queue) }, receiver) +} + +fn sample_turn_start_request() -> ClientRequest { + ClientRequest::TurnStart { + request_id: RequestId::Integer(1), + params: TurnStartParams { + thread_id: "thread-1".to_string(), + input: Vec::new(), + ..Default::default() + }, + } +} + +fn sample_turn_steer_request() -> ClientRequest { + ClientRequest::TurnSteer { + request_id: RequestId::Integer(2), + params: TurnSteerParams { + thread_id: "thread-1".to_string(), + expected_turn_id: "turn-1".to_string(), + input: Vec::new(), + responsesapi_client_metadata: None, + }, + } +} + +fn sample_thread_archive_request() -> ClientRequest { + ClientRequest::ThreadArchive { + request_id: RequestId::Integer(3), + params: ThreadArchiveParams { + thread_id: "thread-1".to_string(), + }, + } +} + +#[test] +fn track_request_only_enqueues_analytics_relevant_requests() { + let (client, mut receiver) = client_with_receiver(); + + for (request_id, request) in [ + (RequestId::Integer(1), sample_turn_start_request()), + (RequestId::Integer(2), sample_turn_steer_request()), + ] { + client.track_request(/*connection_id*/ 7, request_id, &request); + assert!(matches!( + receiver.try_recv(), + Ok(AnalyticsFact::ClientRequest { .. }) + )); + } + + let ignored_request = sample_thread_archive_request(); + client.track_request( + /*connection_id*/ 7, + RequestId::Integer(3), + &ignored_request, + ); + assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); +} diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index e89af03d85..94659ee738 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -157,6 +157,7 @@ macro_rules! client_request_definitions { params: $(#[$params_meta:meta])* $params:ty, $(inspect_params: $inspect_params:tt,)? serialization: $serialization:ident $( ( $($serialization_args:tt)* ) )?, + $(manual_payload_conversion: $manual_payload_conversion:ident,)? response: $response:ty, } ),* $(,)? @@ -243,8 +244,100 @@ macro_rules! client_request_definitions { }) .unwrap_or_else(|| "".to_string()) } + + pub fn into_jsonrpc_parts( + self, + ) -> std::result::Result<(RequestId, crate::Result), serde_json::Error> { + match self { + $( + Self::$variant { request_id, response } => { + serde_json::to_value(response).map(|result| (request_id, result)) + } + )* + } + } } + #[derive(Debug, Clone)] + #[allow(clippy::large_enum_variant)] + pub enum ClientResponsePayload { + $( $variant($response), )* + InterruptConversation(v1::InterruptConversationResponse), + } + + impl ClientResponsePayload { + pub fn into_jsonrpc_parts_and_payload( + self, + request_id: RequestId, + ) -> std::result::Result< + (RequestId, crate::Result, Option), + serde_json::Error, + > { + match self { + $( + Self::$variant(response) => { + let result = serde_json::to_value(&response)?; + Ok((request_id, result, Some(Self::$variant(response)))) + } + )* + Self::InterruptConversation(response) => { + serde_json::to_value(response).map(|result| (request_id, result, None)) + } + } + } + + pub fn into_client_response(self, request_id: RequestId) -> Option { + match self { + $( + Self::$variant(response) => { + Some(ClientResponse::$variant { + request_id, + response, + }) + } + )* + Self::InterruptConversation(_) => None, + } + } + + pub fn into_jsonrpc_parts( + self, + request_id: RequestId, + ) -> std::result::Result<(RequestId, crate::Result), serde_json::Error> { + self.to_jsonrpc_parts(request_id) + } + + pub fn to_jsonrpc_parts( + &self, + request_id: RequestId, + ) -> std::result::Result<(RequestId, crate::Result), serde_json::Error> { + match self { + $( + Self::$variant(response) => { + serde_json::to_value(response).map(|result| (request_id, result)) + } + )* + Self::InterruptConversation(response) => { + serde_json::to_value(response).map(|result| (request_id, result)) + } + } + } + } + + impl From for ClientResponsePayload { + fn from(response: v1::InterruptConversationResponse) -> Self { + Self::InterruptConversation(response) + } + } + + $( + client_response_payload_from_impl!( + $variant, + $response + $(, $manual_payload_conversion)? + ); + )* + impl crate::experimental_api::ExperimentalApi for ClientRequest { fn experimental_reason(&self) -> Option<&'static str> { match self { @@ -317,6 +410,17 @@ macro_rules! client_request_definitions { }; } +macro_rules! client_response_payload_from_impl { + ($variant:ident, $response:ty) => { + impl From<$response> for ClientResponsePayload { + fn from(response: $response) -> Self { + Self::$variant(response) + } + } + }; + ($variant:ident, $response:ty, manual) => {}; +} + client_request_definitions! { Initialize { params: v1::InitializeParams, @@ -789,11 +893,13 @@ client_request_definitions! { ConfigValueWrite => "config/value/write" { params: v2::ConfigValueWriteParams, serialization: global("config"), + manual_payload_conversion: manual, response: v2::ConfigWriteResponse, }, ConfigBatchWrite => "config/batchWrite" { params: v2::ConfigBatchWriteParams, serialization: global("config"), + manual_payload_conversion: manual, response: v2::ConfigWriteResponse, }, @@ -2766,3 +2872,7 @@ mod tests { ); } } + +#[cfg(test)] +#[path = "common_tests.rs"] +mod common_tests; diff --git a/codex-rs/app-server-protocol/src/protocol/common_tests.rs b/codex-rs/app-server-protocol/src/protocol/common_tests.rs new file mode 100644 index 0000000000..83e5d37117 --- /dev/null +++ b/codex-rs/app-server-protocol/src/protocol/common_tests.rs @@ -0,0 +1,44 @@ +use super::*; +use anyhow::Result; +use codex_protocol::protocol::TurnAbortReason; +use pretty_assertions::assert_eq; +use serde_json::json; + +#[test] +fn client_response_payload_returns_jsonrpc_parts_and_client_response() -> Result<()> { + let (request_id, result, payload) = + ClientResponsePayload::ThreadArchive(v2::ThreadArchiveResponse {}) + .into_jsonrpc_parts_and_payload(RequestId::Integer(7))?; + + assert_eq!(request_id, RequestId::Integer(7)); + assert_eq!(result, json!({})); + + let Some(ClientResponse::ThreadArchive { + request_id, + response: _, + }) = payload.and_then(|payload| payload.into_client_response(RequestId::Integer(7))) + else { + panic!("expected thread/archive client response"); + }; + assert_eq!(request_id, RequestId::Integer(7)); + Ok(()) +} + +#[test] +fn interrupt_conversation_payload_stays_jsonrpc_only() -> Result<()> { + let (request_id, result, payload) = + ClientResponsePayload::InterruptConversation(v1::InterruptConversationResponse { + abort_reason: TurnAbortReason::Interrupted, + }) + .into_jsonrpc_parts_and_payload(RequestId::Integer(8))?; + + assert_eq!(request_id, RequestId::Integer(8)); + assert_eq!( + result, + json!({ + "abortReason": "interrupted", + }) + ); + assert!(payload.is_none()); + Ok(()) +} diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index e53b5e2c20..d250959d8d 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -43,6 +43,7 @@ use codex_app_server_protocol::CancelLoginAccountResponse; use codex_app_server_protocol::CancelLoginAccountStatus; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ClientResponse; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::CodexErrorInfo; use codex_app_server_protocol::CollaborationModeListParams; use codex_app_server_protocol::CollaborationModeListResponse; @@ -2118,7 +2119,7 @@ impl CodexMessageProcessor { let result = self .exec_one_off_command_inner(request_id.clone(), params) .await - .map(|()| None::); + .map(|()| None::); self.send_optional_result(request_id, result).await; } @@ -2864,7 +2865,6 @@ impl CodexMessageProcessor { response: response.clone(), }, ); - listener_task_context .outgoing .send_response(request_id, response) @@ -3544,7 +3544,7 @@ impl CodexMessageProcessor { let result = self .thread_rollback_start(&request_id, params) .await - .map(|()| None::); + .map(|()| None::); self.send_optional_result(request_id, result).await; } @@ -4401,6 +4401,7 @@ impl CodexMessageProcessor { permission_profile, reasoning_effort: session_configured.reasoning_effort, }; + self.analytics_events_client.track_response( request_id.connection_id.0, ClientResponse::ThreadResume { @@ -4408,7 +4409,6 @@ impl CodexMessageProcessor { response: response.clone(), }, ); - let connection_id = request_id.connection_id; let token_usage_thread = include_turns.then(|| response.thread.clone()); self.outgoing.send_response(request_id, response).await; @@ -5027,7 +5027,6 @@ impl CodexMessageProcessor { response: response.clone(), }, ); - let connection_id = request_id.connection_id; let token_usage_thread = include_turns.then(|| response.thread.clone()); self.outgoing.send_response(request_id, response).await; @@ -5811,7 +5810,7 @@ impl CodexMessageProcessor { request_id: ConnectionRequestId, result: Result, JSONRPCErrorError>, ) where - T: serde::Serialize, + T: Into, { match result { Ok(Some(response)) => self.outgoing.send_response(request_id, response).await, diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 56d5ff183a..1a28f8e278 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -37,6 +37,7 @@ use codex_app_server_protocol::ChatgptAuthTokensRefreshResponse; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientNotification; use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::ConfigBatchWriteParams; use codex_app_server_protocol::ConfigValueWriteParams; use codex_app_server_protocol::ConfigWarningNotification; @@ -732,15 +733,11 @@ impl MessageProcessor { return Err(invalid_request(experimental_required_message(reason))); } let connection_id = connection_request_id.connection_id; - if let ClientRequest::TurnStart { request_id, .. } - | ClientRequest::TurnSteer { request_id, .. } = &codex_request - { - self.analytics_events_client.track_request( - connection_id.0, - request_id.clone(), - codex_request.clone(), - ); - } + self.analytics_events_client.track_request( + connection_id.0, + connection_request_id.request_id.clone(), + &codex_request, + ); let serialization_scope = codex_request.serialization_scope(); let app_server_client_name = session.app_server_client_name().map(str::to_string); @@ -992,7 +989,12 @@ impl MessageProcessor { params: ConfigValueWriteParams, ) { let result = self.config_api.write_value(params).await; - self.handle_config_mutation_result(request_id, result).await + self.handle_config_mutation_result( + request_id, + result, + ClientResponsePayload::ConfigValueWrite, + ) + .await } async fn handle_config_batch_write( @@ -1001,7 +1003,12 @@ impl MessageProcessor { params: ConfigBatchWriteParams, ) { let result = self.config_api.batch_write(params).await; - self.handle_config_mutation_result(request_id, result).await; + self.handle_config_mutation_result( + request_id, + result, + ClientResponsePayload::ConfigBatchWrite, + ) + .await; } async fn handle_experimental_feature_enablement_set( @@ -1015,7 +1022,12 @@ impl MessageProcessor { .set_experimental_feature_enablement(params) .await; let is_ok = result.is_ok(); - self.handle_config_mutation_result(request_id, result).await; + self.handle_config_mutation_result( + request_id, + result, + ClientResponsePayload::ExperimentalFeatureEnablementSet, + ) + .await; if should_refresh_apps_list && is_ok { self.refresh_apps_list_after_experimental_feature_enablement_set() .await; @@ -1091,15 +1103,18 @@ impl MessageProcessor { }); } - async fn handle_config_mutation_result( + async fn handle_config_mutation_result( &self, request_id: ConnectionRequestId, result: std::result::Result, + wrap_success: impl FnOnce(T) -> ClientResponsePayload, ) { match result { Ok(response) => { self.handle_config_mutation().await; - self.outgoing.send_response(request_id, response).await; + self.outgoing + .send_response_as(request_id, wrap_success(response)) + .await; } Err(error) => self.outgoing.send_error(request_id, error).await, } @@ -1177,7 +1192,7 @@ impl MessageProcessor { device_key_requests_allowed: bool, run_request: F, ) where - R: serde::Serialize + Send + 'static, + R: Into + Send + 'static, F: FnOnce(DeviceKeyApi) -> Fut + Send + 'static, Fut: Future> + Send + 'static, { diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 300a7044e5..588b4b53eb 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -5,6 +5,7 @@ use std::sync::atomic::AtomicI64; use std::sync::atomic::Ordering; use codex_analytics::AnalyticsEventsClient; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::Result; @@ -188,11 +189,10 @@ impl ThreadScopedOutgoingMessageSender { .await } - pub(crate) async fn send_response( - &self, - request_id: ConnectionRequestId, - response: T, - ) { + pub(crate) async fn send_response(&self, request_id: ConnectionRequestId, response: T) + where + T: Into, + { self.outgoing.send_response(request_id, response).await; } @@ -482,21 +482,28 @@ impl OutgoingMessageSender { } } - pub(crate) async fn send_response( + pub(crate) async fn send_response(&self, request_id: ConnectionRequestId, response: T) + where + T: Into, + { + self.send_response_as(request_id, response.into()).await; + } + + pub(crate) async fn send_response_as( &self, request_id: ConnectionRequestId, - response: T, + response: ClientResponsePayload, ) { + let connection_id = request_id.connection_id; + let serialized_response = response.into_jsonrpc_parts(request_id.request_id.clone()); let request_context = self.take_request_context(&request_id).await; - match serde_json::to_value(response) { - Ok(result) => { - let outgoing_message = OutgoingMessage::Response(OutgoingResponse { - id: request_id.request_id.clone(), - result, - }); + + match serialized_response { + Ok((id, result)) => { + let outgoing_message = OutgoingMessage::Response(OutgoingResponse { id, result }); self.send_outgoing_message_to_connection( request_context, - request_id.connection_id, + connection_id, outgoing_message, "response", ) @@ -592,11 +599,13 @@ impl OutgoingMessageSender { request_id: ConnectionRequestId, result: std::result::Result, ) where - T: Serialize, + T: Into, E: Into, { match result { - Ok(response) => self.send_response(request_id, response).await, + Ok(response) => { + self.send_response(request_id, response).await; + } Err(error) => self.send_error(request_id, error).await, } } @@ -966,7 +975,12 @@ mod tests { }; outgoing - .send_response(request_id.clone(), json!({ "ok": true })) + .send_response( + request_id.clone(), + ClientResponsePayload::ThreadArchive( + codex_app_server_protocol::ThreadArchiveResponse {}, + ), + ) .await; let envelope = timeout(Duration::from_secs(1), rx.recv()) @@ -985,7 +999,7 @@ mod tests { panic!("expected response message"); }; assert_eq!(response.id, request_id.request_id); - assert_eq!(response.result, json!({ "ok": true })); + assert_eq!(response.result, json!({})); } other => panic!("expected targeted response envelope, got: {other:?}"), } @@ -1011,7 +1025,12 @@ mod tests { assert_eq!(outgoing.request_context_count().await, 1); outgoing - .send_response(request_id, json!({ "ok": true })) + .send_response( + request_id, + ClientResponsePayload::ThreadArchive( + codex_app_server_protocol::ThreadArchiveResponse {}, + ), + ) .await; assert_eq!(outgoing.request_context_count().await, 0); From afbddabc8bee29e72f19e866c46db9b9f7a2f836 Mon Sep 17 00:00:00 2001 From: xli-oai Date: Wed, 29 Apr 2026 14:01:11 -0700 Subject: [PATCH 04/56] Require remote plugin detail before uninstall (#19966) ## Summary - Fetch remote plugin detail before sending the uninstall request. - Use the detail response to derive the marketplace namespace and plugin name for cache cleanup. - Stop the uninstall before the backend POST if detail lookup fails, so backend state and local cache state do not diverge. ## Testing - `just fmt` - `cargo test -p codex-app-server plugin_uninstall` - `cargo test -p codex-core-plugins` - `git diff --check` --- .../tests/suite/v2/plugin_uninstall.rs | 54 ++------ codex-rs/core-plugins/src/remote.rs | 124 ++++++------------ 2 files changed, 53 insertions(+), 125 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs b/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs index e0f71f11b5..5a05816b0d 100644 --- a/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs +++ b/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs @@ -196,7 +196,6 @@ async fn plugin_uninstall_writes_remote_plugin_to_cloud_when_remote_plugin_enabl )?; mount_remote_plugin_detail(&server, REMOTE_PLUGIN_ID, "1.0.0", "GLOBAL").await; - mount_empty_remote_installed_plugins(&server).await; Mock::given(method("POST")) .and(path(format!( @@ -269,7 +268,6 @@ async fn plugin_uninstall_uses_detail_scope_for_cache_namespace() -> Result<()> AuthCredentialsStoreMode::File, )?; mount_remote_plugin_detail(&server, REMOTE_PLUGIN_ID, "1.0.0", "WORKSPACE").await; - mount_empty_remote_installed_plugins(&server).await; Mock::given(method("POST")) .and(path(format!( @@ -326,7 +324,7 @@ async fn plugin_uninstall_uses_detail_scope_for_cache_namespace() -> Result<()> } #[tokio::test] -async fn plugin_uninstall_posts_even_when_remote_detail_fetch_fails() -> Result<()> { +async fn plugin_uninstall_rejects_before_post_when_remote_detail_fetch_fails() -> Result<()> { let codex_home = TempDir::new()?; let server = MockServer::start().await; write_remote_plugin_catalog_config( @@ -342,19 +340,6 @@ async fn plugin_uninstall_posts_even_when_remote_detail_fetch_fails() -> Result< AuthCredentialsStoreMode::File, )?; - Mock::given(method("POST")) - .and(path(format!( - "/backend-api/plugins/{REMOTE_PLUGIN_ID}/uninstall" - ))) - .and(header("authorization", "Bearer chatgpt-token")) - .and(header("chatgpt-account-id", "account-123")) - .respond_with( - ResponseTemplate::new(200) - .set_body_string(format!(r#"{{"id":"{REMOTE_PLUGIN_ID}","enabled":false}}"#)), - ) - .mount(&server) - .await; - let legacy_remote_plugin_cache_root = codex_home .path() .join(format!("plugins/cache/chatgpt-global/{REMOTE_PLUGIN_ID}")); @@ -368,22 +353,29 @@ async fn plugin_uninstall_posts_even_when_remote_detail_fetch_fails() -> Result< plugin_id: REMOTE_PLUGIN_ID.to_string(), }) .await?; - let response: JSONRPCResponse = timeout( + let err = timeout( DEFAULT_TIMEOUT, - mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + mcp.read_stream_until_error_message(RequestId::Integer(request_id)), ) .await??; - let response: PluginUninstallResponse = to_response(response)?; - assert_eq!(response, PluginUninstallResponse {}); + assert_eq!(err.error.code, -32600); + assert!(err.error.message.contains("remote plugin catalog request")); + wait_for_remote_plugin_request_count( + &server, + "GET", + &format!("/ps/plugins/{REMOTE_PLUGIN_ID}"), + /*expected_count*/ 1, + ) + .await?; wait_for_remote_plugin_request_count( &server, "POST", &format!("/plugins/{REMOTE_PLUGIN_ID}/uninstall"), - /*expected_count*/ 1, + /*expected_count*/ 0, ) .await?; - assert!(!legacy_remote_plugin_cache_root.exists()); + assert!(legacy_remote_plugin_cache_root.exists()); Ok(()) } @@ -557,24 +549,6 @@ async fn mount_remote_plugin_detail( .await; } -async fn mount_empty_remote_installed_plugins(server: &MockServer) { - Mock::given(method("GET")) - .and(path("/backend-api/ps/plugins/installed")) - .and(header("authorization", "Bearer chatgpt-token")) - .and(header("chatgpt-account-id", "account-123")) - .respond_with(ResponseTemplate::new(200).set_body_string( - r#"{ - "plugins": [], - "pagination": { - "limit": 50, - "next_page_token": null - } -}"#, - )) - .mount(server) - .await; -} - async fn wait_for_remote_plugin_request_count( server: &MockServer, method_name: &str, diff --git a/codex-rs/core-plugins/src/remote.rs b/codex-rs/core-plugins/src/remote.rs index 032313ca1f..e6256691f5 100644 --- a/codex-rs/core-plugins/src/remote.rs +++ b/codex-rs/core-plugins/src/remote.rs @@ -13,10 +13,8 @@ use std::collections::BTreeMap; use std::collections::BTreeSet; use std::collections::HashSet; use std::fs; -use std::path::Path; use std::path::PathBuf; use std::time::Duration; -use tracing::warn; pub const REMOTE_GLOBAL_MARKETPLACE_NAME: &str = "chatgpt-global"; pub const REMOTE_WORKSPACE_MARKETPLACE_NAME: &str = "chatgpt-workspace"; @@ -476,27 +474,6 @@ async fn fetch_remote_plugin_detail_with_download_url_option( .await } -async fn fetch_remote_plugin_detail_by_id( - config: &RemotePluginServiceConfig, - auth: &CodexAuth, - plugin_id: &str, -) -> Result { - let plugin = fetch_plugin_detail( - config, auth, plugin_id, /*include_download_urls*/ false, - ) - .await?; - let scope = plugin.scope; - build_remote_plugin_detail( - config, - auth, - scope, - scope.marketplace_name().to_string(), - plugin_id, - plugin, - ) - .await -} - async fn build_remote_plugin_detail( config: &RemotePluginServiceConfig, auth: &CodexAuth, @@ -589,6 +566,12 @@ pub async fn uninstall_remote_plugin( plugin_id: &str, ) -> Result<(), RemotePluginCatalogError> { let auth = ensure_chatgpt_auth(auth)?; + let plugin = fetch_plugin_detail( + config, auth, plugin_id, /*include_download_urls*/ false, + ) + .await?; + let marketplace_name = plugin.scope.marketplace_name().to_string(); + let plugin_name = plugin.name; let base_url = config.chatgpt_base_url.trim_end_matches('/'); let url = format!("{base_url}/plugins/{plugin_id}/uninstall"); @@ -609,19 +592,9 @@ pub async fn uninstall_remote_plugin( }); } - let remote_detail = match fetch_remote_plugin_detail_by_id(config, auth, plugin_id).await { - Ok(remote_detail) => Some(remote_detail), - Err(err) => { - warn!( - plugin_id, - "failed to read remote plugin details after uninstall; skipping named cache removal: {err}" - ); - None - } - }; let legacy_plugin_id = plugin_id.to_string(); tokio::task::spawn_blocking(move || { - remove_remote_plugin_cache(codex_home, remote_detail, legacy_plugin_id) + remove_remote_plugin_cache(codex_home, marketplace_name, plugin_name, legacy_plugin_id) }) .await .map_err(|err| { @@ -636,65 +609,46 @@ pub async fn uninstall_remote_plugin( fn remove_remote_plugin_cache( codex_home: PathBuf, - remote_detail: Option, + marketplace_name: String, + plugin_name: String, legacy_plugin_id: String, ) -> Result<(), String> { - if let Some(remote_detail) = remote_detail { - let marketplace_name = remote_detail.marketplace_name; - let plugin_name = remote_detail.summary.name; - let store = PluginStore::try_new(codex_home.clone()) - .map_err(|err| format!("failed to resolve remote plugin cache root: {err}"))?; - let plugin_id = PluginId::new(plugin_name.clone(), marketplace_name.clone()).map_err( - |err| { - format!( - "invalid remote plugin cache id for `{plugin_name}` in `{marketplace_name}`: {err}" - ) - }, - )?; - let plugin_cache_root = store.plugin_base_root(&plugin_id); - store.uninstall(&plugin_id).map_err(|err| { + let store = PluginStore::try_new(codex_home.clone()) + .map_err(|err| format!("failed to resolve remote plugin cache root: {err}"))?; + let plugin_id = + PluginId::new(plugin_name.clone(), marketplace_name.clone()).map_err(|err| { format!( - "failed to remove remote plugin cache entry {}: {err}", - plugin_cache_root.display() + "invalid remote plugin cache id for `{plugin_name}` in `{marketplace_name}`: {err}" ) })?; - - let legacy_remote_plugin_cache_root = codex_home - .join(PLUGINS_CACHE_DIR) - .join(marketplace_name) - .join(legacy_plugin_id); - if legacy_remote_plugin_cache_root != plugin_cache_root.as_path() { - remove_path_if_exists(&legacy_remote_plugin_cache_root)?; - } - return Ok(()); - } - - for scope in RemotePluginScope::all() { - let legacy_remote_plugin_cache_root = codex_home - .join(PLUGINS_CACHE_DIR) - .join(scope.marketplace_name()) - .join(&legacy_plugin_id); - remove_path_if_exists(&legacy_remote_plugin_cache_root)?; - } - Ok(()) -} - -fn remove_path_if_exists(path: &Path) -> Result<(), String> { - if !path.exists() { - return Ok(()); - } - - let result = if path.is_dir() { - fs::remove_dir_all(path) - } else { - fs::remove_file(path) - }; - result.map_err(|err| { + let plugin_cache_root = store.plugin_base_root(&plugin_id); + store.uninstall(&plugin_id).map_err(|err| { format!( "failed to remove remote plugin cache entry {}: {err}", - path.display() + plugin_cache_root.display() ) - }) + })?; + + let legacy_remote_plugin_cache_root = codex_home + .join(PLUGINS_CACHE_DIR) + .join(marketplace_name) + .join(legacy_plugin_id); + if legacy_remote_plugin_cache_root != plugin_cache_root.as_path() + && legacy_remote_plugin_cache_root.exists() + { + let result = if legacy_remote_plugin_cache_root.is_dir() { + fs::remove_dir_all(&legacy_remote_plugin_cache_root) + } else { + fs::remove_file(&legacy_remote_plugin_cache_root) + }; + result.map_err(|err| { + format!( + "failed to remove remote plugin cache entry {}: {err}", + legacy_remote_plugin_cache_root.display() + ) + })?; + } + Ok(()) } fn build_remote_plugin_summary( From 72a39e3a96e96fe8802b0f779ec82d8aceb5d24e Mon Sep 17 00:00:00 2001 From: rhan-oai Date: Wed, 29 Apr 2026 14:22:39 -0700 Subject: [PATCH 05/56] [app-server] centralize client response analytics (#20059) ## Why The precursor PR keeps successful client responses typed until app-server's outgoing response seam. This follow-up uses that seam to move successful client-response analytics out of individual handlers and into the shared sender path, while keeping filtering decisions inside `codex-analytics`. ## What changed - Emit successful client-response analytics centrally from `OutgoingMessageSender::send_response`. - Remove duplicate handler-local response tracking for the current thread/turn lifecycle responses. - Keep analytics ingestion selective inside `AnalyticsEventsClient`, so unrelated client traffic is ignored before cloning or boxing. - Collapse client-response analytics facts onto one typed path and normalize payloads in the reducer. - Add direct client-filter coverage plus sender-level coverage for the centralized forwarding path. ## Verification - `cargo test -p codex-analytics` - `cargo test -p codex-app-server outgoing_message::tests --lib` --- .../analytics/src/analytics_client_tests.rs | 191 ++++++++++++------ codex-rs/analytics/src/client.rs | 28 ++- codex-rs/analytics/src/client_tests.rs | 141 ++++++++++++- codex-rs/analytics/src/facts.rs | 5 +- codex-rs/analytics/src/reducer.rs | 5 +- .../app-server/src/codex_message_processor.rs | 38 ---- codex-rs/app-server/src/outgoing_message.rs | 14 +- 7 files changed, 311 insertions(+), 111 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 9f88ceeb94..8fd1ee7dc3 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -59,7 +59,7 @@ use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientRequest; -use codex_app_server_protocol::ClientResponse; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::CodexErrorInfo; use codex_app_server_protocol::InitializeCapabilities; use codex_app_server_protocol::InitializeParams; @@ -71,6 +71,8 @@ use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::SessionSource as AppServerSessionSource; use codex_app_server_protocol::Thread; +use codex_app_server_protocol::ThreadArchiveParams; +use codex_app_server_protocol::ThreadArchiveResponse; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus; @@ -141,23 +143,24 @@ fn sample_thread_with_source( } } -fn sample_thread_start_response(thread_id: &str, ephemeral: bool, model: &str) -> ClientResponse { - ClientResponse::ThreadStart { - request_id: RequestId::Integer(1), - response: ThreadStartResponse { - thread: sample_thread(thread_id, ephemeral), - model: model.to_string(), - model_provider: "openai".to_string(), - service_tier: None, - cwd: test_path_buf("/tmp").abs(), - instruction_sources: Vec::new(), - approval_policy: AppServerAskForApproval::OnFailure, - approvals_reviewer: AppServerApprovalsReviewer::User, - sandbox: AppServerSandboxPolicy::DangerFullAccess, - permission_profile: Some(sample_permission_profile()), - reasoning_effort: None, - }, - } +fn sample_thread_start_response( + thread_id: &str, + ephemeral: bool, + model: &str, +) -> ClientResponsePayload { + ClientResponsePayload::ThreadStart(ThreadStartResponse { + thread: sample_thread(thread_id, ephemeral), + model: model.to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: test_path_buf("/tmp").abs(), + instruction_sources: Vec::new(), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + permission_profile: Some(sample_permission_profile()), + reasoning_effort: None, + }) } fn sample_permission_profile() -> AppServerPermissionProfile { @@ -183,7 +186,11 @@ fn sample_runtime_metadata() -> CodexRuntimeMetadata { } } -fn sample_thread_resume_response(thread_id: &str, ephemeral: bool, model: &str) -> ClientResponse { +fn sample_thread_resume_response( + thread_id: &str, + ephemeral: bool, + model: &str, +) -> ClientResponsePayload { sample_thread_resume_response_with_source( thread_id, ephemeral, @@ -197,23 +204,20 @@ fn sample_thread_resume_response_with_source( ephemeral: bool, model: &str, source: AppServerSessionSource, -) -> ClientResponse { - ClientResponse::ThreadResume { - request_id: RequestId::Integer(2), - response: ThreadResumeResponse { - thread: sample_thread_with_source(thread_id, ephemeral, source), - model: model.to_string(), - model_provider: "openai".to_string(), - service_tier: None, - cwd: test_path_buf("/tmp").abs(), - instruction_sources: Vec::new(), - approval_policy: AppServerAskForApproval::OnFailure, - approvals_reviewer: AppServerApprovalsReviewer::User, - sandbox: AppServerSandboxPolicy::DangerFullAccess, - permission_profile: Some(sample_permission_profile()), - reasoning_effort: None, - }, - } +) -> ClientResponsePayload { + ClientResponsePayload::ThreadResume(ThreadResumeResponse { + thread: sample_thread_with_source(thread_id, ephemeral, source), + model: model.to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: test_path_buf("/tmp").abs(), + instruction_sources: Vec::new(), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + permission_profile: Some(sample_permission_profile()), + reasoning_effort: None, + }) } fn sample_turn_start_request(thread_id: &str, request_id: i64) -> ClientRequest { @@ -235,21 +239,18 @@ fn sample_turn_start_request(thread_id: &str, request_id: i64) -> ClientRequest } } -fn sample_turn_start_response(turn_id: &str, request_id: i64) -> ClientResponse { - ClientResponse::TurnStart { - request_id: RequestId::Integer(request_id), - response: codex_app_server_protocol::TurnStartResponse { - turn: Turn { - id: turn_id.to_string(), - items: vec![], - status: AppServerTurnStatus::InProgress, - error: None, - started_at: None, - completed_at: None, - duration_ms: None, - }, +fn sample_turn_start_response(turn_id: &str) -> ClientResponsePayload { + ClientResponsePayload::TurnStart(codex_app_server_protocol::TurnStartResponse { + turn: Turn { + id: turn_id.to_string(), + items: vec![], + status: AppServerTurnStatus::InProgress, + error: None, + started_at: None, + completed_at: None, + duration_ms: None, }, - } + }) } fn sample_turn_started_notification(thread_id: &str, turn_id: &str) -> ServerNotification { @@ -355,13 +356,10 @@ fn sample_turn_steer_request( } } -fn sample_turn_steer_response(turn_id: &str, request_id: i64) -> ClientResponse { - ClientResponse::TurnSteer { - request_id: RequestId::Integer(request_id), - response: TurnSteerResponse { - turn_id: turn_id.to_string(), - }, - } +fn sample_turn_steer_response(turn_id: &str) -> ClientResponsePayload { + ClientResponsePayload::TurnSteer(TurnSteerResponse { + turn_id: turn_id.to_string(), + }) } fn no_active_turn_steer_error() -> JSONRPCErrorError { @@ -488,6 +486,7 @@ async fn ingest_turn_prerequisites( .ingest( AnalyticsFact::ClientResponse { connection_id: 7, + request_id: RequestId::Integer(1), response: Box::new(sample_thread_start_response( "thread-2", /*ephemeral*/ false, "gpt-5", )), @@ -512,7 +511,8 @@ async fn ingest_turn_prerequisites( .ingest( AnalyticsFact::ClientResponse { connection_id: 7, - response: Box::new(sample_turn_start_response("turn-2", /*request_id*/ 3)), + request_id: RequestId::Integer(3), + response: Box::new(sample_turn_start_response("turn-2")), }, out, ) @@ -864,6 +864,7 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize .ingest( AnalyticsFact::ClientResponse { connection_id: 7, + request_id: RequestId::Integer(1), response: Box::new(sample_thread_start_response( "thread-no-client", /*ephemeral*/ false, @@ -908,6 +909,7 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize .ingest( AnalyticsFact::ClientResponse { connection_id: 7, + request_id: RequestId::Integer(2), response: Box::new(sample_thread_resume_response( "thread-1", /*ephemeral*/ true, "gpt-5", )), @@ -954,6 +956,65 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize ); } +#[tokio::test] +async fn unrelated_client_requests_are_ignored_by_reducer() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + + reducer + .ingest( + AnalyticsFact::ClientRequest { + connection_id: 7, + request_id: RequestId::Integer(3), + request: Box::new(ClientRequest::ThreadArchive { + request_id: RequestId::Integer(3), + params: ThreadArchiveParams { + thread_id: "thread-2".to_string(), + }, + }), + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + request_id: RequestId::Integer(3), + response: Box::new(sample_turn_start_response("turn-2")), + }, + &mut events, + ) + .await; + + assert!( + events.is_empty(), + "unrelated requests must not create pending turn state" + ); +} + +#[tokio::test] +async fn unrelated_client_responses_are_ignored_by_reducer() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + + ingest_initialize(&mut reducer, &mut events).await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + request_id: RequestId::Integer(9), + response: Box::new(ClientResponsePayload::ThreadArchive( + ThreadArchiveResponse {}, + )), + }, + &mut events, + ) + .await; + + assert!(events.is_empty()); +} + #[tokio::test] async fn compaction_event_ingests_custom_fact() { let mut reducer = AnalyticsReducer::default(); @@ -988,6 +1049,7 @@ async fn compaction_event_ingests_custom_fact() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, + request_id: RequestId::Integer(2), response: Box::new(sample_thread_resume_response_with_source( "thread-1", /*ephemeral*/ false, @@ -1099,6 +1161,7 @@ async fn guardian_review_event_ingests_custom_fact_with_optional_target_item() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, + request_id: RequestId::Integer(1), response: Box::new(sample_thread_start_response( "thread-guardian", /*ephemeral*/ false, @@ -1881,7 +1944,8 @@ async fn accepted_turn_steer_emits_expected_event() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, - response: Box::new(sample_turn_steer_response("turn-2", /*request_id*/ 4)), + request_id: RequestId::Integer(4), + response: Box::new(sample_turn_steer_response("turn-2")), }, &mut out, ) @@ -2047,7 +2111,8 @@ async fn turn_start_error_response_discards_pending_start_request() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, - response: Box::new(sample_turn_start_response("turn-2", /*request_id*/ 3)), + request_id: RequestId::Integer(3), + response: Box::new(sample_turn_start_response("turn-2")), }, &mut out, ) @@ -2176,7 +2241,8 @@ async fn accepted_steers_increment_turn_steer_count() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, - response: Box::new(sample_turn_steer_response("turn-2", /*request_id*/ 4)), + request_id: RequestId::Integer(4), + response: Box::new(sample_turn_steer_response("turn-2")), }, &mut out, ) @@ -2222,7 +2288,8 @@ async fn accepted_steers_increment_turn_steer_count() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, - response: Box::new(sample_turn_steer_response("turn-2", /*request_id*/ 6)), + request_id: RequestId::Integer(6), + response: Box::new(sample_turn_steer_response("turn-2")), }, &mut out, ) diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index 2ff4c794ed..d54c53ede9 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -22,7 +22,7 @@ use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnTokenUsageFact; use crate::reducer::AnalyticsReducer; use codex_app_server_protocol::ClientRequest; -use codex_app_server_protocol::ClientResponse; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::RequestId; @@ -295,9 +295,25 @@ impl AnalyticsEventsClient { } } - pub fn track_response(&self, connection_id: u64, response: ClientResponse) { + pub fn track_response( + &self, + connection_id: u64, + request_id: RequestId, + response: ClientResponsePayload, + ) { + if !matches!( + response, + ClientResponsePayload::ThreadStart(_) + | ClientResponsePayload::ThreadResume(_) + | ClientResponsePayload::ThreadFork(_) + | ClientResponsePayload::TurnStart(_) + | ClientResponsePayload::TurnSteer(_) + ) { + return; + } self.record_fact(AnalyticsFact::ClientResponse { connection_id, + request_id, response: Box::new(response), }); } @@ -335,10 +351,6 @@ impl AnalyticsEventsClient { } } -#[cfg(test)] -#[path = "client_tests.rs"] -mod tests; - async fn send_track_events( auth_manager: &AuthManager, base_url: &str, @@ -379,3 +391,7 @@ async fn send_track_events( } } } + +#[cfg(test)] +#[path = "client_tests.rs"] +mod tests; diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index 57679f24a1..da18da2529 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -1,11 +1,30 @@ use super::AnalyticsEventsClient; use super::AnalyticsEventsQueue; use crate::facts::AnalyticsFact; +use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; +use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::ClientResponsePayload; +use codex_app_server_protocol::PermissionProfile as AppServerPermissionProfile; use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy; +use codex_app_server_protocol::SessionSource as AppServerSessionSource; +use codex_app_server_protocol::Thread; use codex_app_server_protocol::ThreadArchiveParams; +use codex_app_server_protocol::ThreadArchiveResponse; +use codex_app_server_protocol::ThreadForkResponse; +use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus; +use codex_app_server_protocol::Turn; use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::TurnStatus as AppServerTurnStatus; use codex_app_server_protocol::TurnSteerParams; +use codex_app_server_protocol::TurnSteerResponse; +use codex_protocol::models::PermissionProfile as CorePermissionProfile; +use codex_utils_absolute_path::test_support::PathBufExt; +use codex_utils_absolute_path::test_support::test_path_buf; use std::collections::HashSet; use std::sync::Arc; use std::sync::Mutex; @@ -13,7 +32,7 @@ use tokio::sync::mpsc; use tokio::sync::mpsc::error::TryRecvError; fn client_with_receiver() -> (AnalyticsEventsClient, mpsc::Receiver) { - let (sender, receiver) = mpsc::channel(4); + let (sender, receiver) = mpsc::channel(8); let queue = AnalyticsEventsQueue { sender, app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), @@ -54,6 +73,100 @@ fn sample_thread_archive_request() -> ClientRequest { } } +fn sample_thread(thread_id: &str) -> Thread { + Thread { + id: thread_id.to_string(), + forked_from_id: None, + preview: "first prompt".to_string(), + ephemeral: false, + model_provider: "openai".to_string(), + created_at: 1, + updated_at: 2, + status: AppServerThreadStatus::Idle, + path: None, + cwd: test_path_buf("/tmp").abs(), + cli_version: "0.0.0".to_string(), + source: AppServerSessionSource::Exec, + agent_nickname: None, + agent_role: None, + git_info: None, + name: None, + turns: Vec::new(), + } +} + +fn sample_permission_profile() -> AppServerPermissionProfile { + CorePermissionProfile::Disabled.into() +} + +fn sample_thread_start_response() -> ClientResponsePayload { + ClientResponsePayload::ThreadStart(ThreadStartResponse { + thread: sample_thread("thread-1"), + model: "gpt-5".to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: test_path_buf("/tmp").abs(), + instruction_sources: Vec::new(), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + permission_profile: Some(sample_permission_profile()), + reasoning_effort: None, + }) +} + +fn sample_thread_resume_response() -> ClientResponsePayload { + ClientResponsePayload::ThreadResume(ThreadResumeResponse { + thread: sample_thread("thread-2"), + model: "gpt-5".to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: test_path_buf("/tmp").abs(), + instruction_sources: Vec::new(), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + permission_profile: Some(sample_permission_profile()), + reasoning_effort: None, + }) +} + +fn sample_thread_fork_response() -> ClientResponsePayload { + ClientResponsePayload::ThreadFork(ThreadForkResponse { + thread: sample_thread("thread-3"), + model: "gpt-5".to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: test_path_buf("/tmp").abs(), + instruction_sources: Vec::new(), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + permission_profile: Some(sample_permission_profile()), + reasoning_effort: None, + }) +} + +fn sample_turn_start_response() -> ClientResponsePayload { + ClientResponsePayload::TurnStart(TurnStartResponse { + turn: Turn { + id: "turn-1".to_string(), + items: Vec::new(), + status: AppServerTurnStatus::InProgress, + error: None, + started_at: None, + completed_at: None, + duration_ms: None, + }, + }) +} + +fn sample_turn_steer_response() -> ClientResponsePayload { + ClientResponsePayload::TurnSteer(TurnSteerResponse { + turn_id: "turn-2".to_string(), + }) +} + #[test] fn track_request_only_enqueues_analytics_relevant_requests() { let (client, mut receiver) = client_with_receiver(); @@ -77,3 +190,29 @@ fn track_request_only_enqueues_analytics_relevant_requests() { ); assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); } + +#[test] +fn track_response_only_enqueues_analytics_relevant_responses() { + let (client, mut receiver) = client_with_receiver(); + + for (request_id, response) in [ + (RequestId::Integer(1), sample_thread_start_response()), + (RequestId::Integer(2), sample_thread_resume_response()), + (RequestId::Integer(3), sample_thread_fork_response()), + (RequestId::Integer(4), sample_turn_start_response()), + (RequestId::Integer(5), sample_turn_steer_response()), + ] { + client.track_response(/*connection_id*/ 7, request_id, response); + assert!(matches!( + receiver.try_recv(), + Ok(AnalyticsFact::ClientResponse { .. }) + )); + } + + client.track_response( + /*connection_id*/ 7, + RequestId::Integer(6), + ClientResponsePayload::ThreadArchive(ThreadArchiveResponse {}), + ); + assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); +} diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index 86beb0b88f..424dd523b2 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -2,7 +2,7 @@ use crate::events::AppServerRpcTransport; use crate::events::CodexRuntimeMetadata; use crate::events::GuardianReviewEventParams; use codex_app_server_protocol::ClientRequest; -use codex_app_server_protocol::ClientResponse; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::RequestId; @@ -281,7 +281,8 @@ pub(crate) enum AnalyticsFact { }, ClientResponse { connection_id: u64, - response: Box, + request_id: RequestId, + response: Box, }, ErrorResponse { connection_id: u64, diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 33e89df011..43da35c47c 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -181,9 +181,12 @@ impl AnalyticsReducer { } AnalyticsFact::ClientResponse { connection_id, + request_id, response, } => { - self.ingest_response(connection_id, *response, out); + if let Some(response) = response.into_client_response(request_id) { + self.ingest_response(connection_id, response, out); + } } AnalyticsFact::ErrorResponse { connection_id, diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index d250959d8d..a8b3beb871 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -42,7 +42,6 @@ use codex_app_server_protocol::CancelLoginAccountParams; use codex_app_server_protocol::CancelLoginAccountResponse; use codex_app_server_protocol::CancelLoginAccountStatus; use codex_app_server_protocol::ClientRequest; -use codex_app_server_protocol::ClientResponse; use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::CodexErrorInfo; use codex_app_server_protocol::CollaborationModeListParams; @@ -2856,15 +2855,6 @@ impl CodexMessageProcessor { match result { Ok((response, notif)) => { - listener_task_context - .analytics_events_client - .track_response( - request_id.connection_id.0, - ClientResponse::ThreadStart { - request_id: request_id.request_id.clone(), - response: response.clone(), - }, - ); listener_task_context .outgoing .send_response(request_id, response) @@ -4402,13 +4392,6 @@ impl CodexMessageProcessor { reasoning_effort: session_configured.reasoning_effort, }; - self.analytics_events_client.track_response( - request_id.connection_id.0, - ClientResponse::ThreadResume { - request_id: request_id.request_id.clone(), - response: response.clone(), - }, - ); let connection_id = request_id.connection_id; let token_usage_thread = include_turns.then(|| response.thread.clone()); self.outgoing.send_response(request_id, response).await; @@ -5020,13 +5003,6 @@ impl CodexMessageProcessor { return; } }; - self.analytics_events_client.track_response( - request_id.connection_id.0, - ClientResponse::ThreadFork { - request_id: request_id.request_id.clone(), - response: response.clone(), - }, - ); let connection_id = request_id.connection_id; let token_usage_thread = include_turns.then(|| response.thread.clone()); self.outgoing.send_response(request_id, response).await; @@ -6644,13 +6620,6 @@ impl CodexMessageProcessor { match result { Ok(response) => { - 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(error) => { @@ -6821,13 +6790,6 @@ impl CodexMessageProcessor { match result { Ok(response) => { - self.analytics_events_client.track_response( - request_id.connection_id.0, - ClientResponse::TurnSteer { - request_id: request_id.request_id.clone(), - response: response.clone(), - }, - ); self.outgoing.send_response(request_id, response).await; } Err(error) => { diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 588b4b53eb..34441f83a0 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -495,7 +495,19 @@ impl OutgoingMessageSender { response: ClientResponsePayload, ) { let connection_id = request_id.connection_id; - let serialized_response = response.into_jsonrpc_parts(request_id.request_id.clone()); + let request_id_for_analytics = request_id.request_id.clone(); + let serialized_response = response + .into_jsonrpc_parts_and_payload(request_id.request_id.clone()) + .map(|(id, result, response)| { + if let Some(response) = response { + self.analytics_events_client.track_response( + connection_id.0, + request_id_for_analytics, + response, + ); + } + (id, result) + }); let request_context = self.take_request_context(&request_id).await; match serialized_response { From 8d5da3ffe58cc13380226a253a7e22a3a2a2abce Mon Sep 17 00:00:00 2001 From: xli-oai Date: Wed, 29 Apr 2026 14:45:27 -0700 Subject: [PATCH 06/56] Fallback login callback port when default is busy (#19334) ## Summary - Keep the preferred ChatGPT login callback port `1455` first. - Preserve the existing `/cancel` recovery for stale Codex login servers. - Fall back to the registered localhost callback port `1457` when `1455` remains unavailable. ## Why Cursor and Codex Desktop both use the ChatGPT account login callback server. On Windows, Cursor can already be listening on `127.0.0.1:1455` / `[::1]:1455`, causing Codex Desktop sign-in to fail with: `Local callback port 1455 is already in use on this machine.` Codex already attempted to cancel a stale Codex login server on that port, but if the listener does not release the port, the old behavior was to fail. The new behavior falls back to `1457`, which matches the fixed redirect URI being registered server-side in `openai/openai#863817`. This keeps the OAuth `redirect_uri` inside Hydra's exact allow-list instead of choosing an arbitrary ephemeral port. ## Validation - `just fmt` - `cargo test -p codex-login` - `git diff --check HEAD~1..HEAD` --- codex-rs/login/src/server.rs | 25 +++++-- .../login/tests/suite/login_server_e2e.rs | 70 +++++++++++++++++++ 2 files changed, 91 insertions(+), 4 deletions(-) diff --git a/codex-rs/login/src/server.rs b/codex-rs/login/src/server.rs index 0c7b810184..f6987d46d9 100644 --- a/codex-rs/login/src/server.rs +++ b/codex-rs/login/src/server.rs @@ -50,6 +50,8 @@ use tracing::warn; const DEFAULT_ISSUER: &str = "https://auth.openai.com"; const DEFAULT_PORT: u16 = 1455; +// Keep in sync with the Codex CLI Hydra redirect URI allow-list. +const FALLBACK_PORT: u16 = 1457; static LOGIN_ERROR_PAGE_TEMPLATE: LazyLock