agent/codex
1
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
f886e33e5a55fcb6637d1d05c0f44cab9fcb933f
codex/codex-rs/app-server-protocol/src/protocol/event_mapping.rs
T
jif fae2709320 multi-agent: add path-based v2 activity tracking (#27007) 
## Why

Multi-agent v2 identifies agents by canonical paths, but its tool
handlers still emitted the larger legacy collaboration begin/end events
built around nickname and role metadata. App-server, rollout-trace,
analytics, and TUI consumers therefore lacked one compact path-based
completion signal that behaved consistently across live events and
replay.

The TUI also needs a bounded `/agent` status surface for v2 agents. It
should use recent local activity for previews, refresh liveness without
loading full histories, and keep the legacy picker available when no
path-backed v2 agent is known.

## What changed

- Replace the v2 `spawn_agent`, `send_message`, `followup_task`, and
`interrupt_agent` legacy lifecycle emissions with a success-only
`SubAgentActivity` event. The event records the tool call ID, occurrence
time, affected thread, canonical agent path, and `started`,
`interacted`, or `interrupted` kind.
- Expose the activity as a completion-only app-server v2
`subAgentActivity` thread item in live notifications and reconstructed
history, regenerate the protocol schemas, and count it in sub-agent tool
analytics.
- Track canonical paths from live activity and loaded-thread metadata in
the TUI, and render the activity in live and replayed transcripts.
- Make `/agent` list running path-backed agents with summaries from
bounded local event buffers. Each summary is capped at 240 graphemes,
the scan is capped at six recent items, only the last three wrapped
lines are shown, and command output is omitted. Liveness falls back to
metadata-only `thread/read` when local turn state is unavailable.
- Persist the activity as a terminal rollout-trace runtime payload and
reduce it to the corresponding spawn, send, follow-up, or close
interaction edge. `interrupt_agent` is classified as a close-edge
operation.
- Preserve the legacy picker when no path-backed v2 agent is known.

## Compatibility

App-server v2 clients that consumed `collabAgentToolCall` begin/end
pairs for these tools must handle the new completion-only
`subAgentActivity` item. Legacy v1 collaboration behavior is unchanged.

## Screenshot

<img width="684" height="288" alt="Screenshot 2026-06-08 at 15 40 47"
src="https://github.com/user-attachments/assets/194b3cd0-619d-45fb-b587-cf3e2b1b8a1d"
/>

## Testing

- `just test -p codex-app-server-protocol`
- `just test -p codex-rollout-trace`
- Added focused coverage for activity analytics, terminal trace
serialization, spawn-edge reduction, `interrupt_agent` classification,
TUI status rendering without aggregated command output, and clearing
stale running state after a completed turn.
2026-06-09 12:14:48 +02:00

612 lines
25 KiB
Rust
Raw Blame History

use crate::protocol::common::ServerNotification;
use crate::protocol::item_builders::build_command_execution_begin_item;
use crate::protocol::item_builders::build_command_execution_end_item;
use crate::protocol::item_builders::convert_patch_changes;
use crate::protocol::v2::AgentMessageDeltaNotification;
use crate::protocol::v2::CollabAgentState;
use crate::protocol::v2::CollabAgentTool;
use crate::protocol::v2::CollabAgentToolCallStatus;
use crate::protocol::v2::CommandExecutionOutputDeltaNotification;
use crate::protocol::v2::DynamicToolCallOutputContentItem;
use crate::protocol::v2::DynamicToolCallStatus;
use crate::protocol::v2::FileChangePatchUpdatedNotification;
use crate::protocol::v2::ItemCompletedNotification;
use crate::protocol::v2::ItemStartedNotification;
use crate::protocol::v2::PlanDeltaNotification;
use crate::protocol::v2::ReasoningSummaryPartAddedNotification;
use crate::protocol::v2::ReasoningSummaryTextDeltaNotification;
use crate::protocol::v2::ReasoningTextDeltaNotification;
use crate::protocol::v2::TerminalInteractionNotification;
use crate::protocol::v2::ThreadItem;
use codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem as CoreDynamicToolCallOutputContentItem;
use codex_protocol::protocol::EventMsg;
use std::collections::HashMap;
/// Build the v2 app-server notification that directly corresponds to a single core event.
///
/// This only covers the stateless event-to-notification projections that have a one-to-one
/// mapping. Callers remain responsible for any surrounding state checks or side effects before
/// invoking this helper.
pub fn item_event_to_server_notification(
msg: EventMsg,
thread_id: &str,
turn_id: &str,
) -> ServerNotification {
let thread_id = thread_id.to_string();
let turn_id = turn_id.to_string();
match msg {
EventMsg::DynamicToolCallResponse(response) => {
let status = if response.success {
DynamicToolCallStatus::Completed
} else {
DynamicToolCallStatus::Failed
};
let duration_ms = i64::try_from(response.duration.as_millis()).ok();
let item = ThreadItem::DynamicToolCall {
id: response.call_id,
namespace: response.namespace,
tool: response.tool,
arguments: response.arguments,
status,
content_items: Some(
response
.content_items
.into_iter()
.map(|item| match item {
CoreDynamicToolCallOutputContentItem::InputText { text } => {
DynamicToolCallOutputContentItem::InputText { text }
}
CoreDynamicToolCallOutputContentItem::InputImage { image_url } => {
DynamicToolCallOutputContentItem::InputImage { image_url }
}
})
.collect(),
),
success: Some(response.success),
duration_ms,
};
ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id,
turn_id: response.turn_id,
item,
completed_at_ms: response.completed_at_ms,
})
}
EventMsg::CollabAgentSpawnBegin(begin_event) => {
let item = ThreadItem::CollabAgentToolCall {
id: begin_event.call_id,
tool: CollabAgentTool::SpawnAgent,
status: CollabAgentToolCallStatus::InProgress,
sender_thread_id: begin_event.sender_thread_id.to_string(),
receiver_thread_ids: Vec::new(),
prompt: Some(begin_event.prompt),
model: Some(begin_event.model),
reasoning_effort: Some(begin_event.reasoning_effort),
agents_states: HashMap::new(),
};
ServerNotification::ItemStarted(ItemStartedNotification {
thread_id,
turn_id,
item,
started_at_ms: begin_event.started_at_ms,
})
}
EventMsg::CollabAgentSpawnEnd(end_event) => {
let has_receiver = end_event.new_thread_id.is_some();
let status = match &end_event.status {
codex_protocol::protocol::AgentStatus::Errored(_)
| codex_protocol::protocol::AgentStatus::NotFound => {
CollabAgentToolCallStatus::Failed
}
_ if has_receiver => CollabAgentToolCallStatus::Completed,
_ => CollabAgentToolCallStatus::Failed,
};
let (receiver_thread_ids, agents_states) = match end_event.new_thread_id {
Some(id) => {
let receiver_id = id.to_string();
let received_status = CollabAgentState::from(end_event.status.clone());
(
vec![receiver_id.clone()],
[(receiver_id, received_status)].into_iter().collect(),
)
}
None => (Vec::new(), HashMap::new()),
};
let item = ThreadItem::CollabAgentToolCall {
id: end_event.call_id,
tool: CollabAgentTool::SpawnAgent,
status,
sender_thread_id: end_event.sender_thread_id.to_string(),
receiver_thread_ids,
prompt: Some(end_event.prompt),
model: Some(end_event.model),
reasoning_effort: Some(end_event.reasoning_effort),
agents_states,
};
ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id,
turn_id,
item,
completed_at_ms: end_event.completed_at_ms,
})
}
EventMsg::CollabAgentInteractionBegin(begin_event) => {
let receiver_thread_ids = vec![begin_event.receiver_thread_id.to_string()];
let item = ThreadItem::CollabAgentToolCall {
id: begin_event.call_id,
tool: CollabAgentTool::SendInput,
status: CollabAgentToolCallStatus::InProgress,
sender_thread_id: begin_event.sender_thread_id.to_string(),
receiver_thread_ids,
prompt: Some(begin_event.prompt),
model: None,
reasoning_effort: None,
agents_states: HashMap::new(),
};
ServerNotification::ItemStarted(ItemStartedNotification {
thread_id,
turn_id,
item,
started_at_ms: begin_event.started_at_ms,
})
}
EventMsg::CollabAgentInteractionEnd(end_event) => {
let status = match &end_event.status {
codex_protocol::protocol::AgentStatus::Errored(_)
| codex_protocol::protocol::AgentStatus::NotFound => {
CollabAgentToolCallStatus::Failed
}
_ => CollabAgentToolCallStatus::Completed,
};
let receiver_id = end_event.receiver_thread_id.to_string();
let received_status = CollabAgentState::from(end_event.status);
let item = ThreadItem::CollabAgentToolCall {
id: end_event.call_id,
tool: CollabAgentTool::SendInput,
status,
sender_thread_id: end_event.sender_thread_id.to_string(),
receiver_thread_ids: vec![receiver_id.clone()],
prompt: Some(end_event.prompt),
model: None,
reasoning_effort: None,
agents_states: [(receiver_id, received_status)].into_iter().collect(),
};
ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id,
turn_id,
item,
completed_at_ms: end_event.completed_at_ms,
})
}
EventMsg::SubAgentActivity(activity) => {
let item = ThreadItem::SubAgentActivity {
id: activity.event_id,
kind: activity.kind.into(),
agent_thread_id: activity.agent_thread_id.to_string(),
agent_path: String::from(activity.agent_path),
};
ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id,
turn_id,
item,
completed_at_ms: activity.occurred_at_ms,
})
}
EventMsg::CollabWaitingBegin(begin_event) => {
let receiver_thread_ids = begin_event
.receiver_thread_ids
.iter()
.map(ToString::to_string)
.collect();
let item = ThreadItem::CollabAgentToolCall {
id: begin_event.call_id,
tool: CollabAgentTool::Wait,
status: CollabAgentToolCallStatus::InProgress,
sender_thread_id: begin_event.sender_thread_id.to_string(),
receiver_thread_ids,
prompt: None,
model: None,
reasoning_effort: None,
agents_states: HashMap::new(),
};
ServerNotification::ItemStarted(ItemStartedNotification {
thread_id,
turn_id,
item,
started_at_ms: begin_event.started_at_ms,
})
}
EventMsg::CollabWaitingEnd(end_event) => {
let status = if end_event.statuses.values().any(|status| {
matches!(
status,
codex_protocol::protocol::AgentStatus::Errored(_)
| codex_protocol::protocol::AgentStatus::NotFound
)
}) {
CollabAgentToolCallStatus::Failed
} else {
CollabAgentToolCallStatus::Completed
};
let receiver_thread_ids = end_event.statuses.keys().map(ToString::to_string).collect();
let agents_states = end_event
.statuses
.iter()
.map(|(id, status)| (id.to_string(), CollabAgentState::from(status.clone())))
.collect();
let item = ThreadItem::CollabAgentToolCall {
id: end_event.call_id,
tool: CollabAgentTool::Wait,
status,
sender_thread_id: end_event.sender_thread_id.to_string(),
receiver_thread_ids,
prompt: None,
model: None,
reasoning_effort: None,
agents_states,
};
ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id,
turn_id,
item,
completed_at_ms: end_event.completed_at_ms,
})
}
EventMsg::CollabCloseBegin(begin_event) => {
let item = ThreadItem::CollabAgentToolCall {
id: begin_event.call_id,
tool: CollabAgentTool::CloseAgent,
status: CollabAgentToolCallStatus::InProgress,
sender_thread_id: begin_event.sender_thread_id.to_string(),
receiver_thread_ids: vec![begin_event.receiver_thread_id.to_string()],
prompt: None,
model: None,
reasoning_effort: None,
agents_states: HashMap::new(),
};
ServerNotification::ItemStarted(ItemStartedNotification {
thread_id,
turn_id,
item,
started_at_ms: begin_event.started_at_ms,
})
}
EventMsg::CollabCloseEnd(end_event) => {
let status = match &end_event.status {
codex_protocol::protocol::AgentStatus::Errored(_)
| codex_protocol::protocol::AgentStatus::NotFound => {
CollabAgentToolCallStatus::Failed
}
_ => CollabAgentToolCallStatus::Completed,
};
let receiver_id = end_event.receiver_thread_id.to_string();
let agents_states = [(
receiver_id.clone(),
CollabAgentState::from(end_event.status),
)]
.into_iter()
.collect();
let item = ThreadItem::CollabAgentToolCall {
id: end_event.call_id,
tool: CollabAgentTool::CloseAgent,
status,
sender_thread_id: end_event.sender_thread_id.to_string(),
receiver_thread_ids: vec![receiver_id],
prompt: None,
model: None,
reasoning_effort: None,
agents_states,
};
ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id,
turn_id,
item,
completed_at_ms: end_event.completed_at_ms,
})
}
EventMsg::CollabResumeBegin(begin_event) => {
let item = ThreadItem::CollabAgentToolCall {
id: begin_event.call_id,
tool: CollabAgentTool::ResumeAgent,
status: CollabAgentToolCallStatus::InProgress,
sender_thread_id: begin_event.sender_thread_id.to_string(),
receiver_thread_ids: vec![begin_event.receiver_thread_id.to_string()],
prompt: None,
model: None,
reasoning_effort: None,
agents_states: HashMap::new(),
};
ServerNotification::ItemStarted(ItemStartedNotification {
thread_id,
turn_id,
item,
started_at_ms: begin_event.started_at_ms,
})
}
EventMsg::CollabResumeEnd(end_event) => {
let status = match &end_event.status {
codex_protocol::protocol::AgentStatus::Errored(_)
| codex_protocol::protocol::AgentStatus::NotFound => {
CollabAgentToolCallStatus::Failed
}
_ => CollabAgentToolCallStatus::Completed,
};
let receiver_id = end_event.receiver_thread_id.to_string();
let agents_states = [(
receiver_id.clone(),
CollabAgentState::from(end_event.status),
)]
.into_iter()
.collect();
let item = ThreadItem::CollabAgentToolCall {
id: end_event.call_id,
tool: CollabAgentTool::ResumeAgent,
status,
sender_thread_id: end_event.sender_thread_id.to_string(),
receiver_thread_ids: vec![receiver_id],
prompt: None,
model: None,
reasoning_effort: None,
agents_states,
};
ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id,
turn_id,
item,
completed_at_ms: end_event.completed_at_ms,
})
}
EventMsg::AgentMessageContentDelta(event) => {
let codex_protocol::protocol::AgentMessageContentDeltaEvent { item_id, delta, .. } =
event;
ServerNotification::AgentMessageDelta(AgentMessageDeltaNotification {
thread_id,
turn_id,
item_id,
delta,
})
}
EventMsg::PlanDelta(event) => ServerNotification::PlanDelta(PlanDeltaNotification {
thread_id,
turn_id,
item_id: event.item_id,
delta: event.delta,
}),
EventMsg::ReasoningContentDelta(event) => {
ServerNotification::ReasoningSummaryTextDelta(ReasoningSummaryTextDeltaNotification {
thread_id,
turn_id,
item_id: event.item_id,
delta: event.delta,
summary_index: event.summary_index,
})
}
EventMsg::ReasoningRawContentDelta(event) => {
ServerNotification::ReasoningTextDelta(ReasoningTextDeltaNotification {
thread_id,
turn_id,
item_id: event.item_id,
delta: event.delta,
content_index: event.content_index,
})
}
EventMsg::AgentReasoningSectionBreak(event) => {
ServerNotification::ReasoningSummaryPartAdded(ReasoningSummaryPartAddedNotification {
thread_id,
turn_id,
item_id: event.item_id,
summary_index: event.summary_index,
})
}
EventMsg::ItemStarted(item_started_event) => {
ServerNotification::ItemStarted(ItemStartedNotification {
thread_id,
turn_id,
item: item_started_event.item.into(),
started_at_ms: item_started_event.started_at_ms,
})
}
EventMsg::ItemCompleted(item_completed_event) => {
ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id,
turn_id,
item: item_completed_event.item.into(),
completed_at_ms: item_completed_event.completed_at_ms,
})
}
EventMsg::PatchApplyUpdated(event) => {
ServerNotification::FileChangePatchUpdated(FileChangePatchUpdatedNotification {
thread_id,
turn_id,
item_id: event.call_id,
changes: convert_patch_changes(&event.changes),
})
}
EventMsg::ExecCommandBegin(exec_command_begin_event) => {
ServerNotification::ItemStarted(ItemStartedNotification {
thread_id,
turn_id,
item: build_command_execution_begin_item(&exec_command_begin_event),
started_at_ms: exec_command_begin_event.started_at_ms,
})
}
EventMsg::ExecCommandOutputDelta(exec_command_output_delta_event) => {
let item_id = exec_command_output_delta_event.call_id;
let delta = String::from_utf8_lossy(&exec_command_output_delta_event.chunk).to_string();
ServerNotification::CommandExecutionOutputDelta(
CommandExecutionOutputDeltaNotification {
thread_id,
turn_id,
item_id,
delta,
},
)
}
EventMsg::TerminalInteraction(terminal_event) => {
ServerNotification::TerminalInteraction(TerminalInteractionNotification {
thread_id,
turn_id,
item_id: terminal_event.call_id,
process_id: terminal_event.process_id,
stdin: terminal_event.stdin,
})
}
EventMsg::ExecCommandEnd(exec_command_end_event) => {
ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id,
turn_id,
item: build_command_execution_end_item(&exec_command_end_event),
completed_at_ms: exec_command_end_event.completed_at_ms,
})
}
_ => unreachable!("unsupported item event"),
}
}
#[cfg(test)]
mod tests {
use super::*;
use codex_protocol::ThreadId;
use codex_protocol::protocol::CollabResumeBeginEvent;
use codex_protocol::protocol::CollabResumeEndEvent;
use codex_protocol::protocol::ExecCommandOutputDeltaEvent;
use codex_protocol::protocol::ExecOutputStream;
use pretty_assertions::assert_eq;
fn assert_item_started_server_notification(
notification: ServerNotification,
expected: ItemStartedNotification,
) {
match notification {
ServerNotification::ItemStarted(payload) => assert_eq!(payload, expected),
other => panic!("expected item started notification, got {other:?}"),
}
}
fn assert_item_completed_server_notification(
notification: ServerNotification,
expected: ItemCompletedNotification,
) {
match notification {
ServerNotification::ItemCompleted(payload) => assert_eq!(payload, expected),
other => panic!("expected item completed notification, got {other:?}"),
}
}
fn assert_command_execution_output_delta_server_notification(
notification: ServerNotification,
expected: CommandExecutionOutputDeltaNotification,
) {
match notification {
ServerNotification::CommandExecutionOutputDelta(payload) => {
assert_eq!(payload, expected)
}
other => panic!("expected command execution output delta, got {other:?}"),
}
}
#[test]
fn collab_resume_begin_maps_to_item_started_resume_agent() {
let event = CollabResumeBeginEvent {
call_id: "call-1".to_string(),
started_at_ms: 123,
sender_thread_id: ThreadId::new(),
receiver_thread_id: ThreadId::new(),
receiver_agent_nickname: None,
receiver_agent_role: None,
};
let notification = item_event_to_server_notification(
EventMsg::CollabResumeBegin(event.clone()),
"thread-1",
"turn-1",
);
assert_item_started_server_notification(
notification,
ItemStartedNotification {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
started_at_ms: event.started_at_ms,
item: ThreadItem::CollabAgentToolCall {
id: event.call_id,
tool: CollabAgentTool::ResumeAgent,
status: CollabAgentToolCallStatus::InProgress,
sender_thread_id: event.sender_thread_id.to_string(),
receiver_thread_ids: vec![event.receiver_thread_id.to_string()],
prompt: None,
model: None,
reasoning_effort: None,
agents_states: HashMap::new(),
},
},
);
}
#[test]
fn collab_resume_end_maps_to_item_completed_resume_agent() {
let event = CollabResumeEndEvent {
call_id: "call-2".to_string(),
completed_at_ms: 456,
sender_thread_id: ThreadId::new(),
receiver_thread_id: ThreadId::new(),
receiver_agent_nickname: None,
receiver_agent_role: None,
status: codex_protocol::protocol::AgentStatus::NotFound,
};
let receiver_id = event.receiver_thread_id.to_string();
let notification = item_event_to_server_notification(
EventMsg::CollabResumeEnd(event.clone()),
"thread-2",
"turn-2",
);
assert_item_completed_server_notification(
notification,
ItemCompletedNotification {
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
completed_at_ms: event.completed_at_ms,
item: ThreadItem::CollabAgentToolCall {
id: event.call_id,
tool: CollabAgentTool::ResumeAgent,
status: CollabAgentToolCallStatus::Failed,
sender_thread_id: event.sender_thread_id.to_string(),
receiver_thread_ids: vec![receiver_id.clone()],
prompt: None,
model: None,
reasoning_effort: None,
agents_states: [(
receiver_id,
CollabAgentState::from(codex_protocol::protocol::AgentStatus::NotFound),
)]
.into_iter()
.collect(),
},
},
);
}
#[test]
fn exec_command_output_delta_maps_to_command_execution_output_delta() {
let notification = item_event_to_server_notification(
EventMsg::ExecCommandOutputDelta(ExecCommandOutputDeltaEvent {
call_id: "call-1".to_string(),
stream: ExecOutputStream::Stdout,
chunk: b"hello".to_vec(),
}),
"thread-1",
"turn-1",
);
assert_command_execution_output_delta_server_notification(
notification,
CommandExecutionOutputDeltaNotification {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
item_id: "call-1".to_string(),
delta: "hello".to_string(),
},
);
}
}
Reference in New Issue View Git Blame Copy Permalink