Skip to content

Commit 122f1c9

Browse files
owenlin0owenlin0
committed
feat(core): emit canonical sub-agent activity items
1 parent 1661367 commit 122f1c9

5 files changed

Lines changed: 103 additions & 76 deletions

File tree

‎codex-rs/app-server/src/bespoke_event_handling.rs‎

Lines changed: 58 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -852,23 +852,10 @@ pub(crate) async fn apply_bespoke_event_handling(
852852
);
853853
outgoing.send_server_notification(notification).await;
854854
}
855-
EventMsg::SubAgentActivity(activity) => {
856-
if activity.kind == SubAgentActivityKind::Interrupted
857-
&& thread_manager
858-
.get_thread(activity.agent_thread_id)
859-
.await
860-
.is_err()
861-
{
862-
thread_watch_manager
863-
.remove_thread(&activity.agent_thread_id.to_string())
864-
.await;
865-
}
866-
let notification = item_event_to_server_notification(
867-
EventMsg::SubAgentActivity(activity),
868-
&conversation_id.to_string(),
869-
&event_turn_id,
870-
);
871-
outgoing.send_server_notification(notification).await;
855+
EventMsg::SubAgentActivity(_) => {
856+
// Deprecated sub-agent activity events are still fanned out for raw-event and
857+
// rollout compatibility consumers. App-server v2 receives the canonical
858+
// SubAgentActivity item lifecycle instead.
872859
}
873860
EventMsg::CollabCloseEnd(end_event) => {
874861
if thread_manager
@@ -1033,7 +1020,13 @@ pub(crate) async fn apply_bespoke_event_handling(
10331020
}
10341021
}
10351022
EventMsg::ItemCompleted(event) => {
1036-
apply_canonical_item_completed_side_effects(&thread_state, &event.item).await;
1023+
apply_canonical_item_completed_side_effects(
1024+
&thread_manager,
1025+
&thread_watch_manager,
1026+
&thread_state,
1027+
&event.item,
1028+
)
1029+
.await;
10371030
let notification = item_event_to_server_notification(
10381031
EventMsg::ItemCompleted(event),
10391032
&conversation_id.to_string(),
@@ -1334,16 +1327,43 @@ async fn emit_turn_completed_with_status(
13341327
}
13351328

13361329
async fn apply_canonical_item_completed_side_effects(
1330+
thread_manager: &Arc<ThreadManager>,
1331+
thread_watch_manager: &ThreadWatchManager,
13371332
thread_state: &Arc<Mutex<ThreadState>>,
13381333
item: &CoreTurnItem,
13391334
) {
1340-
if let CoreTurnItem::CommandExecution(item) = item {
1341-
thread_state
1342-
.lock()
1343-
.await
1344-
.turn_summary
1345-
.command_execution_started
1346-
.remove(&item.id);
1335+
match item {
1336+
CoreTurnItem::CommandExecution(item) => {
1337+
thread_state
1338+
.lock()
1339+
.await
1340+
.turn_summary
1341+
.command_execution_started
1342+
.remove(&item.id);
1343+
}
1344+
CoreTurnItem::SubAgentActivity(activity)
1345+
if activity.kind == SubAgentActivityKind::Interrupted =>
1346+
{
1347+
remove_missing_thread_watch(
1348+
thread_manager,
1349+
thread_watch_manager,
1350+
activity.agent_thread_id,
1351+
)
1352+
.await;
1353+
}
1354+
_ => {}
1355+
}
1356+
}
1357+
1358+
async fn remove_missing_thread_watch(
1359+
thread_manager: &Arc<ThreadManager>,
1360+
thread_watch_manager: &ThreadWatchManager,
1361+
thread_id: ThreadId,
1362+
) {
1363+
if thread_manager.get_thread(thread_id).await.is_err() {
1364+
thread_watch_manager
1365+
.remove_thread(&thread_id.to_string())
1366+
.await;
13471367
}
13481368
}
13491369

@@ -2175,6 +2195,7 @@ mod tests {
21752195
use codex_protocol::items::DynamicToolCallItem;
21762196
use codex_protocol::items::DynamicToolCallStatus as CoreDynamicToolCallStatus;
21772197
use codex_protocol::items::HookPromptFragment;
2198+
use codex_protocol::items::SubAgentActivityItem;
21782199
use codex_protocol::items::TurnItem as CoreTurnItem;
21792200
use codex_protocol::items::build_hook_prompt_message;
21802201
use codex_protocol::models::FileSystemPermissions as CoreFileSystemPermissions;
@@ -2192,12 +2213,12 @@ mod tests {
21922213
use codex_protocol::protocol::EventMsg;
21932214
use codex_protocol::protocol::GuardianAssessmentEvent;
21942215
use codex_protocol::protocol::GuardianAssessmentStatus;
2216+
use codex_protocol::protocol::ItemCompletedEvent;
21952217
use codex_protocol::protocol::ItemStartedEvent;
21962218
use codex_protocol::protocol::RateLimitSnapshot;
21972219
use codex_protocol::protocol::RateLimitWindow;
21982220
use codex_protocol::protocol::RolloutItem;
21992221
use codex_protocol::protocol::SessionSource;
2200-
use codex_protocol::protocol::SubAgentActivityEvent;
22012222
use codex_protocol::protocol::TokenUsage;
22022223
use codex_protocol::protocol::TokenUsageInfo;
22032224
use codex_protocol::protocol::UserMessageEvent;
@@ -3420,13 +3441,17 @@ mod tests {
34203441
apply_bespoke_event_handling(
34213442
Event {
34223443
id: "turn-1".to_string(),
3423-
msg: EventMsg::SubAgentActivity(SubAgentActivityEvent {
3424-
event_id: "activity-1".to_string(),
3425-
occurred_at_ms: 42,
3426-
agent_thread_id: child_thread_id,
3427-
agent_path: AgentPath::try_from("/root/worker")
3428-
.expect("agent path should parse"),
3429-
kind: SubAgentActivityKind::Interrupted,
3444+
msg: EventMsg::ItemCompleted(ItemCompletedEvent {
3445+
thread_id: conversation_id,
3446+
turn_id: "turn-1".to_string(),
3447+
item: CoreTurnItem::SubAgentActivity(SubAgentActivityItem {
3448+
id: "activity-1".to_string(),
3449+
kind: SubAgentActivityKind::Interrupted,
3450+
agent_thread_id: child_thread_id,
3451+
agent_path: AgentPath::try_from("/root/worker")
3452+
.expect("agent path should parse"),
3453+
}),
3454+
completed_at_ms: 42,
34303455
}),
34313456
},
34323457
conversation_id,

‎codex-rs/core/src/tools/handlers/multi_agents_v2.rs‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,12 +12,13 @@ use crate::tools::handlers::parse_arguments;
1212
use crate::tools::registry::CoreToolRuntime;
1313
use crate::tools::registry::ToolExecutor;
1414
use codex_protocol::AgentPath;
15+
use codex_protocol::items::SubAgentActivityItem;
16+
use codex_protocol::items::TurnItem;
1517
use codex_protocol::models::ResponseInputItem;
1618
use codex_protocol::openai_models::ReasoningEffort;
1719
use codex_protocol::protocol::CollabWaitingBeginEvent;
1820
use codex_protocol::protocol::CollabWaitingEndEvent;
1921
use codex_protocol::protocol::InterAgentCommunication;
20-
use codex_protocol::protocol::SubAgentActivityEvent;
2122
use codex_protocol::protocol::SubAgentActivityKind;
2223
use codex_tools::ToolName;
2324
use serde::Deserialize;
@@ -39,6 +40,16 @@ mod send_message;
3940
mod spawn;
4041
pub(crate) mod wait;
4142

43+
pub(crate) async fn emit_sub_agent_activity(
44+
session: &crate::session::session::Session,
45+
turn: &crate::session::turn_context::TurnContext,
46+
item: SubAgentActivityItem,
47+
) {
48+
session
49+
.emit_turn_item_completed(turn, TurnItem::SubAgentActivity(item))
50+
.await;
51+
}
52+
4253
pub(super) fn communication_from_tool_message(
4354
author: AgentPath,
4455
recipient: AgentPath,

‎codex-rs/core/src/tools/handlers/multi_agents_v2/interrupt_agent.rs‎

Lines changed: 11 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
use super::*;
22
use crate::tools::handlers::multi_agents_spec::create_interrupt_agent_tool_v2;
3-
use crate::turn_timing::now_unix_timestamp_ms;
43
use codex_protocol::error::CodexErr;
54
use codex_tools::ToolSpec;
65

@@ -71,19 +70,17 @@ async fn handle_interrupt_agent(
7170
Err(err) => Err(collab_agent_error(agent_id, err)),
7271
};
7372
result?;
74-
session
75-
.send_event(
76-
&turn,
77-
SubAgentActivityEvent {
78-
event_id: call_id,
79-
occurred_at_ms: now_unix_timestamp_ms(),
80-
agent_thread_id: agent_id,
81-
agent_path: receiver_agent_path,
82-
kind: SubAgentActivityKind::Interrupted,
83-
}
84-
.into(),
85-
)
86-
.await;
73+
emit_sub_agent_activity(
74+
&session,
75+
&turn,
76+
SubAgentActivityItem {
77+
id: call_id,
78+
agent_thread_id: agent_id,
79+
agent_path: receiver_agent_path,
80+
kind: SubAgentActivityKind::Interrupted,
81+
},
82+
)
83+
.await;
8784

8885
Ok(InterruptAgentResult {
8986
previous_status: status,

‎codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs‎

Lines changed: 11 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@ use super::*;
77
use crate::agent_communication::AgentCommunicationContext;
88
use crate::agent_communication::AgentCommunicationKind;
99
use crate::tools::context::FunctionToolOutput;
10-
use crate::turn_timing::now_unix_timestamp_ms;
1110
use codex_protocol::protocol::InterAgentCommunication;
1211

1312
#[derive(Clone, Copy, PartialEq, Eq)]
@@ -115,19 +114,17 @@ pub(crate) async fn handle_message_string_tool(
115114
.await
116115
.map_err(|err| collab_agent_error(receiver_thread_id, err));
117116
result?;
118-
session
119-
.send_event(
120-
&turn,
121-
SubAgentActivityEvent {
122-
event_id: call_id,
123-
occurred_at_ms: now_unix_timestamp_ms(),
124-
agent_thread_id: receiver_thread_id,
125-
agent_path: receiver_agent_path,
126-
kind: SubAgentActivityKind::Interacted,
127-
}
128-
.into(),
129-
)
130-
.await;
117+
emit_sub_agent_activity(
118+
&session,
119+
&turn,
120+
SubAgentActivityItem {
121+
id: call_id,
122+
agent_thread_id: receiver_thread_id,
123+
agent_path: receiver_agent_path,
124+
kind: SubAgentActivityKind::Interacted,
125+
},
126+
)
127+
.await;
131128

132129
Ok(FunctionToolOutput::from_text(String::new(), Some(true)))
133130
}

‎codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs‎

Lines changed: 11 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,6 @@ use crate::agent_communication::AgentCommunicationKind;
99
use crate::tools::handlers::multi_agents_spec::SpawnAgentToolOptions;
1010
use crate::tools::handlers::multi_agents_spec::create_spawn_agent_tool_v2;
1111
use crate::tools::handlers::multi_agents_v2::message_tool::message_content;
12-
use crate::turn_timing::now_unix_timestamp_ms;
1312
use codex_protocol::AgentPath;
1413
use codex_tools::ToolSpec;
1514

@@ -140,19 +139,17 @@ async fn handle_spawn_agent(
140139
.as_ref()
141140
.and_then(|snapshot| snapshot.session_source.get_nickname())
142141
.or(spawned_agent.metadata.agent_nickname);
143-
session
144-
.send_event(
145-
&turn,
146-
SubAgentActivityEvent {
147-
event_id: call_id,
148-
occurred_at_ms: now_unix_timestamp_ms(),
149-
agent_thread_id: new_thread_id,
150-
agent_path: new_agent_path.clone(),
151-
kind: SubAgentActivityKind::Started,
152-
}
153-
.into(),
154-
)
155-
.await;
142+
emit_sub_agent_activity(
143+
&session,
144+
&turn,
145+
SubAgentActivityItem {
146+
id: call_id,
147+
agent_thread_id: new_thread_id,
148+
agent_path: new_agent_path.clone(),
149+
kind: SubAgentActivityKind::Started,
150+
},
151+
)
152+
.await;
156153
let role_tag = role_name.unwrap_or(DEFAULT_ROLE_NAME);
157154
turn.session_telemetry.counter(
158155
"codex.multi_agent.spawn",

0 commit comments

Comments
 (0)