diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 56ecf981d7..b535c1a34d 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -5066,6 +5066,12 @@ async fn turn_event_counts_completed_tool_items() { agent_thread_id: "thread-child".to_string(), agent_path: "/root/child".to_string(), }, + ThreadItem::SubAgentActivity { + id: "sub-agent-activity-completed".to_string(), + kind: SubAgentActivityKind::Completed, + agent_thread_id: "thread-child".to_string(), + agent_path: "/root/child".to_string(), + }, ThreadItem::WebSearch(WebSearchItem { id: "web-1".to_string(), query: "codex".to_string(), diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 226bf54540..a25627c43c 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -143,6 +143,7 @@ use codex_app_server_protocol::RequestPermissionProfile; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ServerResponse; +use codex_app_server_protocol::SubAgentActivityKind; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::TurnSteerResponse; use codex_app_server_protocol::UserInput; @@ -476,6 +477,10 @@ impl TurnToolCounts { ThreadItem::FileChange { .. } => self.file_change += 1, ThreadItem::McpToolCall { .. } => self.mcp_tool_call += 1, ThreadItem::DynamicToolCall { .. } => self.dynamic_tool_call += 1, + ThreadItem::SubAgentActivity { + kind: SubAgentActivityKind::Completed, + .. + } => return, ThreadItem::CollabAgentToolCall { id, .. } | ThreadItem::SubAgentActivity { id, .. } => { if !self.subagent_tool_call_ids.insert(id.clone()) { diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index e97d56bbad..4f73d298a5 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -3672,7 +3672,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 188c15b611..a2f100c793 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -19366,7 +19366,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index a77cf152f0..9dc3448153 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -17118,7 +17118,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/ItemCompletedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ItemCompletedNotification.json index 463c63e3d2..bf28b03d62 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ItemCompletedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ItemCompletedNotification.json @@ -565,7 +565,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/ItemStartedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ItemStartedNotification.json index 1600d8e38e..4791191a32 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ItemStartedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ItemStartedNotification.json @@ -565,7 +565,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/ReviewStartResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ReviewStartResponse.json index 2f91b873de..0451f740f3 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ReviewStartResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ReviewStartResponse.json @@ -711,7 +711,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json index cbcca82aa3..0332dfde65 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json @@ -986,7 +986,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json index 8d47036ab7..105d934b73 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json @@ -777,7 +777,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadMetadataUpdateResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadMetadataUpdateResponse.json index 7e57650a26..19c11d8a4c 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadMetadataUpdateResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadMetadataUpdateResponse.json @@ -777,7 +777,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json index 7580d0fc5f..8da91c2226 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json @@ -777,7 +777,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json index 6cadd8aa83..390882d3d9 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json @@ -986,7 +986,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json index 5ab29235a1..4687c7e048 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json @@ -777,7 +777,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json index 7d1c095b37..ceb64b2d92 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json @@ -986,7 +986,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json index ce357835ee..9de1753c59 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json @@ -777,7 +777,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json index 46b73654d1..dc72d5ebc9 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json @@ -777,7 +777,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/TurnCompletedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/TurnCompletedNotification.json index c0f6759e8f..43060b2d44 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/TurnCompletedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/TurnCompletedNotification.json @@ -711,7 +711,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/TurnStartResponse.json b/codex-rs/app-server-protocol/schema/json/v2/TurnStartResponse.json index 276e23b7d0..5728fce1e1 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/TurnStartResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/TurnStartResponse.json @@ -711,7 +711,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/TurnStartedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/TurnStartedNotification.json index 2cc4bb10d1..b57f48f30b 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/TurnStartedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/TurnStartedNotification.json @@ -711,7 +711,8 @@ "enum": [ "started", "interacted", - "interrupted" + "interrupted", + "completed" ], "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index d996aed8f7..0d009b2036 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index bd468c2323..1f03c2a00c 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/SubAgentActivityKind.ts b/codex-rs/app-server-protocol/schema/typescript/v2/SubAgentActivityKind.ts index 5e3ce81e01..8dcaceed08 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/SubAgentActivityKind.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/SubAgentActivityKind.ts @@ -2,4 +2,4 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. -export type SubAgentActivityKind = "started" | "interacted" | "interrupted"; +export type SubAgentActivityKind = "started" | "interacted" | "interrupted" | "completed"; diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index 34afee0e16..566b99d12e 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -1631,6 +1631,7 @@ mod tests { use codex_protocol::items::EnteredReviewModeItem as CoreEnteredReviewModeItem; use codex_protocol::items::ExitedReviewModeItem as CoreExitedReviewModeItem; use codex_protocol::items::HookPromptFragment as CoreHookPromptFragment; + use codex_protocol::items::SubAgentActivityItem as CoreSubAgentActivityItem; use codex_protocol::items::TurnItem as CoreTurnItem; use codex_protocol::items::UserMessageItem as CoreUserMessageItem; use codex_protocol::items::build_hook_prompt_message; @@ -1654,6 +1655,7 @@ mod tests { use codex_protocol::protocol::McpToolCallEndEvent; use codex_protocol::protocol::PatchApplyBeginEvent; use codex_protocol::protocol::ReviewTarget; + use codex_protocol::protocol::SubAgentActivityKind as CoreSubAgentActivityKind; use codex_protocol::protocol::ThreadRolledBackEvent; use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::TurnAbortedEvent; @@ -4480,6 +4482,57 @@ mod tests { ); } + #[test] + fn completed_sub_agent_activity_updates_completed_parent_turn() { + let child_thread_id = ThreadId::new(); + let child_path = codex_protocol::AgentPath::root() + .join("worker") + .expect("worker path"); + let items = vec![ + RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_id: "turn-a".into(), + trace_id: None, + started_at: None, + model_context_window: None, + collaboration_mode_kind: Default::default(), + })), + RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent { + turn_id: "turn-a".into(), + started_at: None, + last_agent_message: None, + error: None, + completed_at: None, + duration_ms: None, + time_to_first_token_ms: None, + })), + RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent { + thread_id: ThreadId::new(), + turn_id: "turn-a".into(), + item: CoreTurnItem::SubAgentActivity(CoreSubAgentActivityItem { + id: "child-turn-completed".into(), + kind: CoreSubAgentActivityKind::Completed, + agent_thread_id: child_thread_id, + agent_path: child_path, + }), + started_at_ms: None, + completed_at_ms: 0, + })), + ]; + + let turns = build_turns_from_rollout_items(&items); + + assert_eq!(turns.len(), 1); + assert_eq!( + turns[0].items, + vec![ThreadItem::SubAgentActivity { + id: "child-turn-completed".into(), + kind: crate::protocol::v2::SubAgentActivityKind::Completed, + agent_thread_id: child_thread_id.to_string(), + agent_path: "/root/worker".into(), + }] + ); + } + #[test] fn ignores_plain_user_response_items_in_rollout_replay() { let items = vec![ diff --git a/codex-rs/app-server-protocol/src/protocol/v2/item.rs b/codex-rs/app-server-protocol/src/protocol/v2/item.rs index c2b59e4594..6881d12099 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/item.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/item.rs @@ -1180,6 +1180,7 @@ pub enum SubAgentActivityKind { Started, Interacted, Interrupted, + Completed, } impl From for SubAgentActivityKind { @@ -1188,6 +1189,7 @@ impl From for SubAgentActivityKind { CoreSubAgentActivityKind::Started => SubAgentActivityKind::Started, CoreSubAgentActivityKind::Interacted => SubAgentActivityKind::Interacted, CoreSubAgentActivityKind::Interrupted => SubAgentActivityKind::Interrupted, + CoreSubAgentActivityKind::Completed => SubAgentActivityKind::Completed, } } } diff --git a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs index 472856bffb..c031207c01 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -3187,7 +3187,7 @@ fn core_turn_item_into_thread_item_converts_supported_variants() { let sub_agent_activity_item = TurnItem::SubAgentActivity(SubAgentActivityItem { id: "activity-1".to_string(), - kind: CoreSubAgentActivityKind::Interrupted, + kind: CoreSubAgentActivityKind::Completed, agent_thread_id: receiver_thread_id, agent_path: codex_protocol::AgentPath::root() .join("worker") @@ -3198,7 +3198,7 @@ fn core_turn_item_into_thread_item_converts_supported_variants() { ThreadItem::from(sub_agent_activity_item), ThreadItem::SubAgentActivity { id: "activity-1".to_string(), - kind: SubAgentActivityKind::Interrupted, + kind: SubAgentActivityKind::Completed, agent_thread_id: receiver_thread_id.to_string(), agent_path: "/root/worker".to_string(), } diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index d97fb11cb5..0a8a605531 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -1640,6 +1640,7 @@ The app-server streams JSON-RPC notifications while a turn is running. Each turn - `fileChange` — `{id, changes, status}` describing proposed edits; `changes` list `{path, kind, diff}` and `status` is `inProgress`, `completed`, `failed`, or `declined`. - `mcpToolCall` — `{id, server, tool, status, arguments, appContext, mcpAppResourceUri?, pluginId, readOnlyHint, result?, error?}` describing MCP calls; `appContext` is `{connectorId, linkId, resourceUri, appName, actionName}` for calls through a trusted MCP app, where `connectorId` identifies the connector that owns the tool, `linkId` identifies the app link, `resourceUri` points to the widget template, `appName` is the connector's display name, and `actionName` is the stable connector `Action.name`. `readOnlyHint` is `true` for read-only tools, `false` for write-capable tools, and `null` when the annotation is unavailable, including older rollout entries. The hint describes tool capability, not whether an invocation succeeded or performed a write; use `status`, `result`, and `error` to determine the execution outcome. `appName` and `actionName` may be null for older rollout entries. The top-level `mcpAppResourceUri` is deprecated and temporarily duplicated for client migration. `tool` identifies the raw MCP tool. `status` is `inProgress`, `completed`, or `failed`. - `collabToolCall` — `{id, tool, status, senderThreadId, receiverThreadId?, newThreadId?, prompt?, agentStatus?}` describing collab tool calls (`spawn_agent`, `send_input`, `resume_agent`, `wait`, `close_agent`); `status` is `inProgress`, `completed`, or `failed`. +- `subAgentActivity` — `{id, kind, agentThreadId, agentPath}` describing Multi-Agent V2 lifecycle activity; `kind` is `started`, `interacted`, `interrupted`, or `completed`. A successful child completion is attributed to the parent turn that spawned it, so its `item/completed` notification may arrive after that turn's `turn/completed` notification and is included with that turn when history is read. - `webSearch` — `{id, query, action?, results?}` for a web search request issued by the agent; `action` mirrors the Responses API web_search action payload (`search`, `open_page`, `find_in_page`) and may be omitted until completion. For standalone web search, `results` contains the out-of-band structured result DTOs returned by `/v1/alpha/search`; clients should ignore result types and fields they do not understand. - `imageGeneration` — `{id, status, revisedPrompt, result, transparentBackground, savedPath?}` for a generated image. `transparentBackground` is `true` when the Images API reports a transparent background, `false` when it reports an opaque background, and `null` when the background is automatic, unavailable, or the item has not completed. The field is always present on v2 item payloads, including persisted and resumed items. - `imageView` — `{id, path}` emitted when the agent invokes the image viewer tool. diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index c84c85628f..f32efca616 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -24,6 +24,7 @@ use crate::thread_manager::ThreadIdGenerator; use crate::thread_manager::ThreadManagerState; use crate::thread_manager::default_thread_id_generator; use crate::thread_rollout_truncation::truncate_rollout_to_last_n_fork_turns; +use crate::turn_timing::now_unix_timestamp_ms; use codex_history::InitialHistory; use codex_history::ResumedHistory; use codex_history::RolloutItem; @@ -33,11 +34,17 @@ use codex_protocol::ThreadId; use codex_protocol::error::CodexErr; use codex_protocol::error::CodexErrorDetails; use codex_protocol::error::Result as CodexResult; +use codex_protocol::items::SubAgentActivityItem; +use codex_protocol::items::TurnItem; use codex_protocol::models::ContentItem; use codex_protocol::models::MessagePhase; use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::HasLegacyEvent; use codex_protocol::protocol::InterAgentCommunication; +use codex_protocol::protocol::ItemCompletedEvent; +use codex_protocol::protocol::ItemStartedEvent; use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::Op; use codex_protocol::protocol::SessionSource; @@ -233,6 +240,55 @@ impl AgentControl { .await } + pub(crate) async fn emit_sub_agent_activity( + &self, + thread_id: ThreadId, + turn_id: String, + item: SubAgentActivityItem, + ) -> CodexResult<()> { + let state = self.upgrade()?; + let thread = state.get_thread(thread_id).await?; + let started_at_ms = now_unix_timestamp_ms(); + let item = TurnItem::SubAgentActivity(item); + thread + .session + .send_event_raw(Event { + id: turn_id.clone(), + msg: EventMsg::ItemStarted(ItemStartedEvent { + thread_id, + turn_id: turn_id.clone(), + item: item.clone(), + started_at_ms, + }), + }) + .await; + let completed_at_ms = now_unix_timestamp_ms(); + let completed = ItemCompletedEvent { + thread_id, + turn_id: turn_id.clone(), + item, + started_at_ms: Some(started_at_ms), + completed_at_ms, + }; + thread + .session + .send_event_raw(Event { + id: turn_id.clone(), + msg: EventMsg::ItemCompleted(completed.clone()), + }) + .await; + for legacy in completed.as_legacy_events(/*show_raw_agent_reasoning*/ false) { + thread + .session + .send_event_raw(Event { + id: turn_id.clone(), + msg: legacy, + }) + .await; + } + Ok(()) + } + async fn send_inter_agent_communication_after_capacity_check( &self, agent_id: ThreadId, diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index ba3a2ab738..70bf327fbf 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -101,6 +101,7 @@ use codex_protocol::config_types::WebSearchMode; use codex_protocol::dynamic_tools::DynamicToolResponse; use codex_protocol::dynamic_tools::DynamicToolSpec; use codex_protocol::items::EnteredReviewModeItem; +use codex_protocol::items::SubAgentActivityItem; use codex_protocol::items::TurnItem; use codex_protocol::items::UserMessageItem; use codex_protocol::models::ActivePermissionProfile; @@ -126,6 +127,7 @@ use codex_protocol::protocol::MULTI_AGENT_MODE_OPEN_TAG; use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::RawResponseItemEvent; use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentActivityKind; use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadSettingsOverrides; @@ -2016,6 +2018,26 @@ impl Session { child_agent_path: &codex_protocol::AgentPath, status: AgentStatus, ) { + if matches!(status, AgentStatus::Completed(_)) + && let Some(parent_turn_id) = turn_context.turn_metadata_state.parent_turn_id() + && let Err(err) = self + .services + .agent_control + .emit_sub_agent_activity( + parent_thread_id, + parent_turn_id, + SubAgentActivityItem { + id: format!("subagent-completed-{}", turn_context.sub_id), + kind: SubAgentActivityKind::Completed, + agent_thread_id: self.thread_id, + agent_path: child_agent_path.clone(), + }, + ) + .await + { + debug!("failed to emit completed activity to parent thread {parent_thread_id}: {err}"); + } + let Some(parent_agent_path) = child_agent_path .as_str() .rsplit_once('/') diff --git a/codex-rs/core/src/turn_metadata.rs b/codex-rs/core/src/turn_metadata.rs index 8be5c1705f..ef7244f69e 100644 --- a/codex-rs/core/src/turn_metadata.rs +++ b/codex-rs/core/src/turn_metadata.rs @@ -268,6 +268,10 @@ impl TurnMetadataState { let _ = self.parent_turn_id.set(parent_turn_id); } + pub(crate) fn parent_turn_id(&self) -> Option { + self.parent_turn_id.get().cloned() + } + pub(crate) fn set_root_turn_id(&self, root_turn_id: String) { if root_turn_id.trim().is_empty() { return; diff --git a/codex-rs/core/tests/suite/subagent_notifications.rs b/codex-rs/core/tests/suite/subagent_notifications.rs index 4021048238..a34d0e84fc 100644 --- a/codex-rs/core/tests/suite/subagent_notifications.rs +++ b/codex-rs/core/tests/suite/subagent_notifications.rs @@ -9,6 +9,8 @@ use codex_history::RolloutItem; use codex_models_manager::bundled_models_response; use codex_protocol::ThreadId; use codex_protocol::config_types::ReasoningSummary; +use codex_protocol::items::SubAgentActivityItem; +use codex_protocol::items::TurnItem; use codex_protocol::models::PermissionProfile; use codex_protocol::openai_models::MultiAgentMessages; use codex_protocol::openai_models::MultiAgentRoleMessages; @@ -20,6 +22,7 @@ use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::MULTI_AGENT_MODE_OPEN_TAG; use codex_protocol::protocol::Op; use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentActivityKind; use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadSettingsOverrides; @@ -59,6 +62,7 @@ use std::time::Duration; use test_case::test_case; use tokio::time::Instant; use tokio::time::sleep; +use tokio::time::timeout; use tracing::Level; use tracing_test::internal::MockWriter; use wiremock::MockServer; @@ -1982,11 +1986,25 @@ enum CompletionScenario { TerminalError, } -#[test_case(CompletionScenario::Completed ; "completed")] -#[test_case(CompletionScenario::TerminalError ; "terminal_error")] +#[test_case( + CompletionScenario::Completed, + ThreadHistoryMode::Paginated; + "completed_paginated" +)] +#[test_case( + CompletionScenario::Completed, + ThreadHistoryMode::Legacy; + "completed_legacy" +)] +#[test_case( + CompletionScenario::TerminalError, + ThreadHistoryMode::Paginated; + "terminal_error" +)] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn plaintext_multi_agent_v2_completion_sends_agent_message( scenario: CompletionScenario, + history_mode: ThreadHistoryMode, ) -> Result<()> { let server = start_mock_server().await; let spawn_args = serde_json::to_string(&json!({ @@ -2018,14 +2036,28 @@ async fn plaintext_multi_agent_v2_completion_sends_agent_message( }; let child_request = mount_response_once_match( &server, - |req: &wiremock::Request| request_has_input_type(req, "agent_message"), + |req: &wiremock::Request| { + request_has_input_type(req, "agent_message") + && decoded_body(req) + .and_then(|body| serde_json::from_slice::(&body).ok()) + .and_then(|body| { + body["client_metadata"]["x-codex-turn-metadata"] + .as_str() + .and_then(|metadata| serde_json::from_str::(metadata).ok()) + }) + .is_some_and(|metadata| { + metadata.get("parent_turn_id").is_some_and(Value::is_string) + }) + }, sse_response(sse(child_events)).set_delay(Duration::from_secs(1)), ) .await; mount_sse_once_match( &server, |req: &wiremock::Request| { - body_contains(req, SPAWN_CALL_ID) && !body_contains(req, "Message Type: FINAL_ANSWER") + body_contains(req, SPAWN_CALL_ID) + && !request_has_input_type(req, "agent_message") + && !body_contains(req, "Message Type: FINAL_ANSWER") }, sse(vec![ ev_response_created("resp-parent-2"), @@ -2096,12 +2128,106 @@ async fn plaintext_multi_agent_v2_completion_sends_agent_message( config.model_provider.stream_max_retries = Some(0); config.model_provider.supports_websockets = false; }) + .with_history_mode(history_mode) .build(&server) .await?; test.submit_turn(TURN_1_PROMPT).await?; - let _ = wait_for_requests(&child_request).await?; - test.submit_turn(TURN_2_NO_WAIT_PROMPT).await?; + let deadline = Instant::now() + Duration::from_secs(2); + let (child_request, child_turn_metadata) = loop { + let child_request = child_request.requests().into_iter().find_map(|request| { + let body = request.body_json(); + let turn_metadata: Value = + serde_json::from_str(body["client_metadata"]["x-codex-turn-metadata"].as_str()?) + .ok()?; + turn_metadata + .get("parent_turn_id") + .and_then(Value::as_str)?; + Some((request, turn_metadata)) + }); + if let Some(child_request) = child_request { + break child_request; + } + assert!( + Instant::now() < deadline, + "timed out waiting for child request" + ); + sleep(Duration::from_millis(10)).await; + }; + let expected_completed_activity = if matches!(scenario, CompletionScenario::Completed) { + let child_body = child_request.body_json(); + let parent_turn_id = child_turn_metadata["parent_turn_id"] + .as_str() + .expect("child parent turn ID") + .to_string(); + let child_turn_id = child_body["client_metadata"]["turn_id"] + .as_str() + .expect("child turn ID") + .to_string(); + let child_thread_id = ThreadId::from_string( + child_body["client_metadata"]["thread_id"] + .as_str() + .expect("child thread ID"), + )?; + Some((parent_turn_id, child_turn_id, child_thread_id)) + } else { + None + }; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: TURN_2_NO_WAIT_PROMPT.to_string(), + text_elements: Vec::new(), + }])) + .await?; + let (completed_activity_started, completed_activity_completed) = + timeout(Duration::from_secs(5), async { + let mut active_turn_id = None; + let mut completed_activity_started = None; + let mut completed_activity_completed = None; + loop { + let event = test + .codex + .next_event() + .await + .expect("event stream should remain open"); + match event.msg { + EventMsg::TurnStarted(event) => { + active_turn_id = Some(event.turn_id); + } + EventMsg::ItemStarted(event) + if matches!( + &event.item, + TurnItem::SubAgentActivity(SubAgentActivityItem { + kind: SubAgentActivityKind::Completed, + .. + }) + ) => + { + completed_activity_started = Some(event); + } + EventMsg::ItemCompleted(event) + if matches!( + &event.item, + TurnItem::SubAgentActivity(SubAgentActivityItem { + kind: SubAgentActivityKind::Completed, + .. + }) + ) => + { + completed_activity_completed = Some(event); + } + EventMsg::TurnComplete(event) + if active_turn_id.as_deref() == Some(event.turn_id.as_str()) => + { + break; + } + _ => {} + } + } + (completed_activity_started, completed_activity_completed) + }) + .await + .expect("timed out waiting for parent turn completion"); let request = wait_for_requests(&agent_request) .await? @@ -2122,6 +2248,85 @@ async fn plaintext_multi_agent_v2_completion_sends_agent_message( })]) ); + if let Some((parent_turn_id, child_turn_id, child_thread_id)) = expected_completed_activity { + let started = completed_activity_started.expect("completed activity start event"); + let completed_event = + completed_activity_completed.expect("completed activity completion event"); + let TurnItem::SubAgentActivity(started_item) = &started.item else { + panic!("expected started sub-agent activity"); + }; + let TurnItem::SubAgentActivity(completed_event_item) = &completed_event.item else { + panic!("expected completed sub-agent activity"); + }; + assert_eq!( + ( + started.thread_id, + &started.turn_id, + started_item, + Some(started.started_at_ms), + ), + ( + completed_event.thread_id, + &completed_event.turn_id, + completed_event_item, + completed_event.started_at_ms, + ) + ); + assert_eq!(started.turn_id, parent_turn_id); + + test.codex.ensure_rollout_materialized().await; + test.codex.flush_rollout().await?; + let rollout = codex_rollout::RolloutRecorder::get_rollout_history( + &test.codex.rollout_path().expect("parent rollout path"), + ) + .await?; + assert!( + !rollout.get_rollout_items().iter().any(|item| matches!( + item, + RolloutItem::EventMsg(EventMsg::SubAgentActivity(activity)) + if activity.kind == SubAgentActivityKind::Completed + )), + "legacy completed activity should not be persisted without its parent turn ID" + ); + let completed = rollout + .get_rollout_items() + .iter() + .find_map(|item| match item { + RolloutItem::EventMsg(EventMsg::ItemCompleted(completed)) + if matches!( + &completed.item, + TurnItem::SubAgentActivity(SubAgentActivityItem { + kind: SubAgentActivityKind::Completed, + .. + }) + ) => + { + Some(completed) + } + _ => None, + }) + .expect("persisted completed sub-agent activity"); + + assert_eq!(completed.turn_id, parent_turn_id); + let TurnItem::SubAgentActivity(completed_item) = &completed.item else { + panic!("expected completed sub-agent activity"); + }; + assert_eq!( + completed_item, + &SubAgentActivityItem { + id: format!("subagent-completed-{child_turn_id}"), + kind: SubAgentActivityKind::Completed, + agent_thread_id: child_thread_id, + agent_path: codex_protocol::AgentPath::root() + .join("worker") + .expect("worker path"), + } + ); + } else { + assert!(completed_activity_started.is_none()); + assert!(completed_activity_completed.is_none()); + } + Ok(()) } diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index cd20f787c4..8ea5a4ac75 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -4117,6 +4117,7 @@ pub enum SubAgentActivityKind { Started, Interacted, Interrupted, + Completed, } #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] diff --git a/codex-rs/rollout-trace/src/protocol_event.rs b/codex-rs/rollout-trace/src/protocol_event.rs index e0a03a3737..8f0441981a 100644 --- a/codex-rs/rollout-trace/src/protocol_event.rs +++ b/codex-rs/rollout-trace/src/protocol_event.rs @@ -22,6 +22,7 @@ use codex_protocol::protocol::PatchApplyBeginEvent; use codex_protocol::protocol::PatchApplyEndEvent; use codex_protocol::protocol::PatchApplyStatus; use codex_protocol::protocol::SubAgentActivityEvent; +use codex_protocol::protocol::SubAgentActivityKind; use codex_protocol::protocol::TurnAbortReason; use serde::Serialize; use std::time::Duration; @@ -352,11 +353,14 @@ pub(crate) fn tool_runtime_trace_event(event: &EventMsg) -> Option Some(ToolRuntimeTraceEvent::Ended { - tool_call_id: &event.event_id, - status: ExecutionStatus::Completed, - payload: ToolRuntimePayload::SubAgentActivity(event), - }), + EventMsg::SubAgentActivity(event) if event.kind != SubAgentActivityKind::Completed => { + Some(ToolRuntimeTraceEvent::Ended { + tool_call_id: &event.event_id, + status: ExecutionStatus::Completed, + payload: ToolRuntimePayload::SubAgentActivity(event), + }) + } + EventMsg::SubAgentActivity(_) => None, EventMsg::Error(_) | EventMsg::Warning(_) | EventMsg::GuardianWarning(_) diff --git a/codex-rs/rollout-trace/src/protocol_event_tests.rs b/codex-rs/rollout-trace/src/protocol_event_tests.rs index 1f75c61748..c7c53c3ef3 100644 --- a/codex-rs/rollout-trace/src/protocol_event_tests.rs +++ b/codex-rs/rollout-trace/src/protocol_event_tests.rs @@ -50,6 +50,20 @@ fn sub_agent_activity_is_a_terminal_tool_runtime_event() -> anyhow::Result<()> { Ok(()) } +#[test] +fn completed_sub_agent_activity_is_not_a_tool_runtime_event() -> anyhow::Result<()> { + let event = EventMsg::SubAgentActivity(SubAgentActivityEvent { + event_id: "child-turn-completed".to_string(), + occurred_at_ms: 1234, + agent_thread_id: ThreadId::new(), + agent_path: AgentPath::try_from("/root/reviewer").map_err(anyhow::Error::msg)?, + kind: SubAgentActivityKind::Completed, + }); + + assert!(tool_runtime_trace_event(&event).is_none()); + Ok(()) +} + #[test] fn exec_command_trace_payloads_use_inferred_native_cwd() -> anyhow::Result<()> { // Convention inference depends on the URI spelling, not the test host, so exercise both diff --git a/codex-rs/rollout-trace/src/reducer/tool/agents.rs b/codex-rs/rollout-trace/src/reducer/tool/agents.rs index 10b02f237f..7c71ddef7d 100644 --- a/codex-rs/rollout-trace/src/reducer/tool/agents.rs +++ b/codex-rs/rollout-trace/src/reducer/tool/agents.rs @@ -235,6 +235,7 @@ impl TraceReducer { target_thread_id, Some(wall_time_unix_ms), ), + (_, SubAgentActivityKind::Completed) => Ok(()), _ => bail!( "sub-agent activity {:?} does not match tool call kind {tool_kind:?}", payload.kind diff --git a/codex-rs/rollout/src/policy.rs b/codex-rs/rollout/src/policy.rs index 372c0afaf5..5c6c9e026f 100644 --- a/codex-rs/rollout/src/policy.rs +++ b/codex-rs/rollout/src/policy.rs @@ -3,6 +3,7 @@ use crate::protocol::EventMsg; use codex_extension_items::ExtensionItem; use codex_protocol::items::TurnItem; use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::SubAgentActivityKind; use codex_protocol::protocol::ThreadHistoryMode; /// Whether a rollout `item` should be persisted in rollout files. @@ -89,12 +90,18 @@ pub fn should_persist_event_msg(ev: &EventMsg, history_mode: ThreadHistoryMode) match ev { EventMsg::ItemCompleted(event) => { // Paginated rollouts store TurnItems. - // Legacy rollouts keep only items with no raw ResponseItem or legacy equivalent. + // Legacy rollouts keep only items with no lossless raw ResponseItem or legacy + // equivalent. matches!(history_mode, ThreadHistoryMode::Paginated) || matches!( event.item, TurnItem::Plan(_) | TurnItem::Extension(ExtensionItem::Sleep(_)) ) + || matches!( + &event.item, + TurnItem::SubAgentActivity(item) + if item.kind == SubAgentActivityKind::Completed + ) } EventMsg::TokenCount(_) | EventMsg::ThreadGoalUpdated(_) @@ -116,8 +123,13 @@ pub fn should_persist_event_msg(ev: &EventMsg, history_mode: ThreadHistoryMode) | EventMsg::ContextCompacted(_) | EventMsg::McpToolCallEnd(_) | EventMsg::WebSearchEnd(_) - | EventMsg::ImageGenerationEnd(_) - | EventMsg::SubAgentActivity(_) => matches!(history_mode, ThreadHistoryMode::Legacy), + | EventMsg::ImageGenerationEnd(_) => { + matches!(history_mode, ThreadHistoryMode::Legacy) + } + EventMsg::SubAgentActivity(event) => { + matches!(history_mode, ThreadHistoryMode::Legacy) + && event.kind != SubAgentActivityKind::Completed + } // Transient, non-durable events. EventMsg::Error(_) diff --git a/codex-rs/tui/src/app/agent_status_feed.rs b/codex-rs/tui/src/app/agent_status_feed.rs index 39aa1f7461..3f78b24e70 100644 --- a/codex-rs/tui/src/app/agent_status_feed.rs +++ b/codex-rs/tui/src/app/agent_status_feed.rs @@ -173,6 +173,7 @@ fn activity_summary(item: &ThreadItem) -> Option { SubAgentActivityKind::Started => "Started", SubAgentActivityKind::Interacted => "Contacted", SubAgentActivityKind::Interrupted => "Interrupted", + SubAgentActivityKind::Completed => "Completed", }; return bounded_summary(&format!("{action} {agent_path}")); } diff --git a/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__app_server_sub_agent_activity_renders_once.snap b/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__app_server_sub_agent_activity_renders_once.snap index 5cda26aecf..f364105995 100644 --- a/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__app_server_sub_agent_activity_renders_once.snap +++ b/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__app_server_sub_agent_activity_renders_once.snap @@ -1,5 +1,5 @@ --- -source: tui/src/chatwidget/tests.rs +source: tui/src/chatwidget/tests/app_server.rs expression: rendered --- -• Interacted with `/root/researcher` +• Completed `/root/researcher` diff --git a/codex-rs/tui/src/chatwidget/tests/app_server.rs b/codex-rs/tui/src/chatwidget/tests/app_server.rs index 56ba697a23..70df67db08 100644 --- a/codex-rs/tui/src/chatwidget/tests/app_server.rs +++ b/codex-rs/tui/src/chatwidget/tests/app_server.rs @@ -963,7 +963,7 @@ async fn live_app_server_sub_agent_activity_renders_once() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; let activity = AppServerThreadItem::SubAgentActivity { id: "activity-1".to_string(), - kind: codex_app_server_protocol::SubAgentActivityKind::Interacted, + kind: codex_app_server_protocol::SubAgentActivityKind::Completed, agent_thread_id: ThreadId::new().to_string(), agent_path: "/root/researcher".to_string(), }; diff --git a/codex-rs/tui/src/multi_agents.rs b/codex-rs/tui/src/multi_agents.rs index 5d652cbd10..9d571ba95a 100644 --- a/codex-rs/tui/src/multi_agents.rs +++ b/codex-rs/tui/src/multi_agents.rs @@ -291,7 +291,7 @@ pub(crate) fn sub_agent_activity_display(item: &ThreadItem) -> Option true, SubAgentActivityKind::Interacted => return None, - SubAgentActivityKind::Interrupted => false, + SubAgentActivityKind::Interrupted | SubAgentActivityKind::Completed => false, }; Some(SubAgentActivityDisplay { thread_id: parse_thread_id(agent_thread_id)?, @@ -318,6 +318,7 @@ pub(crate) fn sub_agent_activity_summary(kind: SubAgentActivityKind, agent_path: SubAgentActivityKind::Started => format!("Started `{agent_path}`"), SubAgentActivityKind::Interacted => format!("Interacted with `{agent_path}`"), SubAgentActivityKind::Interrupted => format!("Interrupted `{agent_path}`"), + SubAgentActivityKind::Completed => format!("Completed `{agent_path}`"), } } @@ -326,6 +327,7 @@ fn sub_agent_activity_title(kind: SubAgentActivityKind, agent_path: &str) -> Lin SubAgentActivityKind::Started => ("Started ", agent_path), SubAgentActivityKind::Interacted => ("Interacted with ", agent_path), SubAgentActivityKind::Interrupted => ("Interrupted ", agent_path), + SubAgentActivityKind::Completed => ("Completed ", agent_path), }; title_spans_line(vec![ Span::from(prefix).bold(), @@ -691,6 +693,26 @@ mod tests { assert_eq!(sub_agent_activity_display(&item), None); } + #[test] + fn completed_sub_agent_activity_stops_running_liveness() { + let thread_id = ThreadId::new(); + let item = ThreadItem::SubAgentActivity { + id: "activity-1".to_string(), + kind: SubAgentActivityKind::Completed, + agent_thread_id: thread_id.to_string(), + agent_path: "/root/child".to_string(), + }; + + assert_eq!( + sub_agent_activity_display(&item), + Some(SubAgentActivityDisplay { + thread_id, + agent_path: "/root/child".to_string(), + is_running_hint: false, + }) + ); + } + #[test] fn collab_events_snapshot() { let sender_thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000001")