mirror of
https://github.com/openai/codex.git
synced 2026-10-18 17:18:52 +00:00
Report completed sub-agent activity on parent turns (#40437)
## Why A Multi-Agent V2 child can finish after the parent turn that spawned it, so its successful completion needs to remain associated with that parent turn in notifications and history. ## What changed - Add `completed` to `SubAgentActivityKind` and emit the corresponding started and completed item events on the spawning parent turn when a child succeeds. - Persist completed activity as a canonical turn item in both legacy and paginated history, and expose it through app-server history and schemas. - Render completed activity as terminal in the TUI without counting it as a separate tool call or tool-runtime event. ## Testing - Cover successful completion in legacy and paginated history, late updates to completed parent turns, TUI rendering, analytics, and rollout tracing. - Verify that terminal child errors do not emit completed activity. GitOrigin-RevId: 6c71eb8b10e7327611c6af84f70e294e27d94d02
This commit is contained in:
@@ -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(),
|
||||
|
||||
@@ -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()) {
|
||||
|
||||
@@ -3672,7 +3672,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
+2
-1
@@ -19366,7 +19366,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
+2
-1
@@ -17118,7 +17118,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
+2
-1
@@ -565,7 +565,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
@@ -565,7 +565,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
@@ -711,7 +711,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
@@ -986,7 +986,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
@@ -777,7 +777,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
+2
-1
@@ -777,7 +777,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
@@ -777,7 +777,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
@@ -986,7 +986,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
@@ -777,7 +777,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
@@ -986,7 +986,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
+2
-1
@@ -777,7 +777,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
@@ -777,7 +777,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
+2
-1
@@ -711,7 +711,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
@@ -711,7 +711,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
@@ -711,7 +711,8 @@
|
||||
"enum": [
|
||||
"started",
|
||||
"interacted",
|
||||
"interrupted"
|
||||
"interrupted",
|
||||
"completed"
|
||||
],
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
BIN
Binary file not shown.
BIN
Binary file not shown.
@@ -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";
|
||||
|
||||
@@ -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![
|
||||
|
||||
@@ -1180,6 +1180,7 @@ pub enum SubAgentActivityKind {
|
||||
Started,
|
||||
Interacted,
|
||||
Interrupted,
|
||||
Completed,
|
||||
}
|
||||
|
||||
impl From<CoreSubAgentActivityKind> for SubAgentActivityKind {
|
||||
@@ -1188,6 +1189,7 @@ impl From<CoreSubAgentActivityKind> for SubAgentActivityKind {
|
||||
CoreSubAgentActivityKind::Started => SubAgentActivityKind::Started,
|
||||
CoreSubAgentActivityKind::Interacted => SubAgentActivityKind::Interacted,
|
||||
CoreSubAgentActivityKind::Interrupted => SubAgentActivityKind::Interrupted,
|
||||
CoreSubAgentActivityKind::Completed => SubAgentActivityKind::Completed,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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(),
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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('/')
|
||||
|
||||
@@ -268,6 +268,10 @@ impl TurnMetadataState {
|
||||
let _ = self.parent_turn_id.set(parent_turn_id);
|
||||
}
|
||||
|
||||
pub(crate) fn parent_turn_id(&self) -> Option<String> {
|
||||
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;
|
||||
|
||||
@@ -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::<Value>(&body).ok())
|
||||
.and_then(|body| {
|
||||
body["client_metadata"]["x-codex-turn-metadata"]
|
||||
.as_str()
|
||||
.and_then(|metadata| serde_json::from_str::<Value>(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(())
|
||||
}
|
||||
|
||||
|
||||
@@ -4117,6 +4117,7 @@ pub enum SubAgentActivityKind {
|
||||
Started,
|
||||
Interacted,
|
||||
Interrupted,
|
||||
Completed,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]
|
||||
|
||||
@@ -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<ToolRuntimeTr
|
||||
status: ExecutionStatus::Completed,
|
||||
payload: ToolRuntimePayload::CollabCloseEnd(event),
|
||||
}),
|
||||
EventMsg::SubAgentActivity(event) => 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(_)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(_)
|
||||
|
||||
@@ -173,6 +173,7 @@ fn activity_summary(item: &ThreadItem) -> Option<String> {
|
||||
SubAgentActivityKind::Started => "Started",
|
||||
SubAgentActivityKind::Interacted => "Contacted",
|
||||
SubAgentActivityKind::Interrupted => "Interrupted",
|
||||
SubAgentActivityKind::Completed => "Completed",
|
||||
};
|
||||
return bounded_summary(&format!("{action} {agent_path}"));
|
||||
}
|
||||
|
||||
+2
-2
@@ -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`
|
||||
|
||||
@@ -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(),
|
||||
};
|
||||
|
||||
@@ -291,7 +291,7 @@ pub(crate) fn sub_agent_activity_display(item: &ThreadItem) -> Option<SubAgentAc
|
||||
let is_running_hint = match kind {
|
||||
SubAgentActivityKind::Started => 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")
|
||||
|
||||
Reference in New Issue
Block a user