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

Move item event mapping into app-server-protocol (#20299)

Browse Source 
## Why

Follow-up to #20291.

The v2 item-event-to-notification translation had been embedded in
`app-server/src/bespoke_event_handling.rs`, which made it hard to reuse
anywhere else. This PR moves that stateless mapping into shared protocol
code so other entry points can produce the same `ServerNotification`
payloads without copying app-server logic.

That also lets `thread-manager-sample` demonstrate the same notification
surface that the app server exposes, instead of only printing the final
assistant message.

## What changed

- move `item_event_to_server_notification` into
`codex-app-server-protocol::protocol::event_mapping`
- keep the mapper tests next to the shared implementation in
`codex-app-server-protocol`
- re-export the mapper from `codex-core-api` so lightweight consumers
can use it without reaching into `app-server-protocol` directly
- simplify `app-server/src/bespoke_event_handling.rs` so it delegates
the stateless event-to-notification projection to the shared helper
- update `thread-manager-sample` to:
  - print mapped notifications as newline-delimited JSON
  - use the shared mapper through `codex-core-api`
- enable the default feature set so the sample exposes the normal tool
surface
- use a `read_only` permission profile so shell commands can run in the
sample without widening permissions

## Testing

- `cargo test -p codex-app-server-protocol`
- `cargo test -p codex-core-api`
- `cargo test -p codex-app-server bespoke_event_handling::tests`
- `cargo test -p codex-thread-manager-sample`
- `cargo run -p codex-thread-manager-sample -- "briefly explore the repo
with pwd and ls, then summarize it"`
This commit is contained in:
pakrym-oai
2026-04-30 11:02:13 -07:00
committed by GitHub
Unverified
parent c70cdc108f
commit 5cc5f12efc
9 changed files with 955 additions and 899 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+2
View File
@@ -2485,6 +2485,7 @@ name = "codex-core-api"
version = "0.0.0"
dependencies = [
"codex-analytics",
"codex-app-server-protocol",
"codex-arg0",
"codex-config",
"codex-core",
@@ -3515,6 +3516,7 @@ dependencies = [
"anyhow",
"clap",
"codex-core-api",
"serde_json",
"tracing",
]
codex-rs/app-server-protocol/src/lib.rs
+1
View File
@@ -14,6 +14,7 @@ pub use export::generate_ts_with_options;
pub use export::generate_types;
pub use jsonrpc_lite::*;
pub use protocol::common::*;
pub use protocol::event_mapping::*;
pub use protocol::item_builders::*;
pub use protocol::thread_history::*;
pub use protocol::v1::ApplyPatchApprovalParams;
codex-rs/app-server-protocol/src/protocol/event_mapping.rs
+807
View File
@@ -0,0 +1,807 @@
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::build_file_change_begin_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::FileChangeOutputDeltaNotification;
use crate::protocol::v2::FileChangePatchUpdatedNotification;
use crate::protocol::v2::ItemCompletedNotification;
use crate::protocol::v2::ItemStartedNotification;
use crate::protocol::v2::McpToolCallError;
use crate::protocol::v2::McpToolCallResult;
use crate::protocol::v2::McpToolCallStatus;
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 serde_json::Value as JsonValue;
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,
is_file_change_output: bool,
) -> 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,
})
}
EventMsg::McpToolCallBegin(begin_event) => {
let item = ThreadItem::McpToolCall {
id: begin_event.call_id,
server: begin_event.invocation.server,
tool: begin_event.invocation.tool,
status: McpToolCallStatus::InProgress,
arguments: begin_event.invocation.arguments.unwrap_or(JsonValue::Null),
mcp_app_resource_uri: begin_event.mcp_app_resource_uri,
result: None,
error: None,
duration_ms: None,
};
ServerNotification::ItemStarted(ItemStartedNotification {
thread_id,
turn_id,
item,
})
}
EventMsg::McpToolCallEnd(end_event) => {
let status = if end_event.is_success() {
McpToolCallStatus::Completed
} else {
McpToolCallStatus::Failed
};
let duration_ms = i64::try_from(end_event.duration.as_millis()).ok();
let (result, error) = match &end_event.result {
Ok(value) => (
Some(Box::new(McpToolCallResult {
content: value.content.clone(),
structured_content: value.structured_content.clone(),
meta: value.meta.clone(),
})),
None,
),
Err(message) => (
None,
Some(McpToolCallError {
message: message.clone(),
}),
),
};
let item = ThreadItem::McpToolCall {
id: end_event.call_id,
server: end_event.invocation.server,
tool: end_event.invocation.tool,
status,
arguments: end_event.invocation.arguments.unwrap_or(JsonValue::Null),
mcp_app_resource_uri: end_event.mcp_app_resource_uri,
result,
error,
duration_ms,
};
ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id,
turn_id,
item,
})
}
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,
})
}
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,
})
}
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,
})
}
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,
})
}
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,
})
}
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,
})
}
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,
})
}
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,
})
}
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,
})
}
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,
})
}
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(),
})
}
EventMsg::ItemCompleted(item_completed_event) => {
ServerNotification::ItemCompleted(ItemCompletedNotification {
thread_id,
turn_id,
item: item_completed_event.item.into(),
})
}
EventMsg::PatchApplyBegin(patch_begin_event) => {
ServerNotification::ItemStarted(ItemStartedNotification {
thread_id,
turn_id,
item: build_file_change_begin_item(&patch_begin_event),
})
}
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),
})
}
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();
if is_file_change_output {
ServerNotification::FileChangeOutputDelta(FileChangeOutputDeltaNotification {
thread_id,
turn_id,
item_id,
delta,
})
} else {
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),
})
}
_ => unreachable!("unsupported item event"),
}
}
#[cfg(test)]
mod tests {
use super::*;
use codex_protocol::ThreadId;
use codex_protocol::mcp::CallToolResult;
use codex_protocol::protocol::CollabResumeBeginEvent;
use codex_protocol::protocol::CollabResumeEndEvent;
use codex_protocol::protocol::McpInvocation;
use codex_protocol::protocol::McpToolCallBeginEvent;
use codex_protocol::protocol::McpToolCallEndEvent;
use pretty_assertions::assert_eq;
use rmcp::model::Content;
use std::time::Duration;
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:?}"),
}
}
#[test]
fn collab_resume_begin_maps_to_item_started_resume_agent() {
let event = CollabResumeBeginEvent {
call_id: "call-1".to_string(),
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",
/*is_file_change_output*/ false,
);
assert_item_started_server_notification(
notification,
ItemStartedNotification {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
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(),
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",
/*is_file_change_output*/ false,
);
assert_item_completed_server_notification(
notification,
ItemCompletedNotification {
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
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 mcp_tool_call_begin_maps_to_item_started_notification_with_args() {
let begin_event = McpToolCallBeginEvent {
call_id: "call_123".to_string(),
invocation: McpInvocation {
server: "codex".to_string(),
tool: "list_mcp_resources".to_string(),
arguments: Some(serde_json::json!({"server": ""})),
},
mcp_app_resource_uri: Some("ui://widget/list-resources.html".to_string()),
};
let notification = item_event_to_server_notification(
EventMsg::McpToolCallBegin(begin_event.clone()),
"thread-1",
"turn_1",
/*is_file_change_output*/ false,
);
assert_item_started_server_notification(
notification,
ItemStartedNotification {
thread_id: "thread-1".to_string(),
turn_id: "turn_1".to_string(),
item: ThreadItem::McpToolCall {
id: begin_event.call_id,
server: begin_event.invocation.server,
tool: begin_event.invocation.tool,
status: McpToolCallStatus::InProgress,
arguments: serde_json::json!({"server": ""}),
mcp_app_resource_uri: Some("ui://widget/list-resources.html".to_string()),
result: None,
error: None,
duration_ms: None,
},
},
);
}
#[test]
fn mcp_tool_call_begin_maps_to_item_started_notification_without_args() {
let begin_event = McpToolCallBeginEvent {
call_id: "call_456".to_string(),
invocation: McpInvocation {
server: "codex".to_string(),
tool: "list_mcp_resources".to_string(),
arguments: None,
},
mcp_app_resource_uri: None,
};
let notification = item_event_to_server_notification(
EventMsg::McpToolCallBegin(begin_event.clone()),
"thread-2",
"turn_2",
/*is_file_change_output*/ false,
);
assert_item_started_server_notification(
notification,
ItemStartedNotification {
thread_id: "thread-2".to_string(),
turn_id: "turn_2".to_string(),
item: ThreadItem::McpToolCall {
id: begin_event.call_id,
server: begin_event.invocation.server,
tool: begin_event.invocation.tool,
status: McpToolCallStatus::InProgress,
arguments: JsonValue::Null,
mcp_app_resource_uri: None,
result: None,
error: None,
duration_ms: None,
},
},
);
}
#[test]
fn mcp_tool_call_end_maps_to_item_completed_notification_on_success() {
let content = vec![
serde_json::to_value(Content::text("{\"resources\":[]}"))
.expect("content should serialize"),
];
let result = CallToolResult {
content: content.clone(),
is_error: Some(false),
structured_content: None,
meta: Some(serde_json::json!({
"ui/resourceUri": "ui://widget/list-resources.html"
})),
};
let end_event = McpToolCallEndEvent {
call_id: "call_789".to_string(),
invocation: McpInvocation {
server: "codex".to_string(),
tool: "list_mcp_resources".to_string(),
arguments: Some(serde_json::json!({"server": ""})),
},
mcp_app_resource_uri: Some("ui://widget/list-resources.html".to_string()),
duration: Duration::from_nanos(92708),
result: Ok(result),
};
let notification = item_event_to_server_notification(
EventMsg::McpToolCallEnd(end_event.clone()),
"thread-3",
"turn_3",
/*is_file_change_output*/ false,
);
assert_item_completed_server_notification(
notification,
ItemCompletedNotification {
thread_id: "thread-3".to_string(),
turn_id: "turn_3".to_string(),
item: ThreadItem::McpToolCall {
id: end_event.call_id,
server: end_event.invocation.server,
tool: end_event.invocation.tool,
status: McpToolCallStatus::Completed,
arguments: serde_json::json!({"server": ""}),
mcp_app_resource_uri: Some("ui://widget/list-resources.html".to_string()),
result: Some(Box::new(McpToolCallResult {
content,
structured_content: None,
meta: Some(serde_json::json!({
"ui/resourceUri": "ui://widget/list-resources.html"
})),
})),
error: None,
duration_ms: Some(0),
},
},
);
}
#[test]
fn mcp_tool_call_end_maps_to_item_completed_notification_on_error() {
let end_event = McpToolCallEndEvent {
call_id: "call_err".to_string(),
invocation: McpInvocation {
server: "codex".to_string(),
tool: "list_mcp_resources".to_string(),
arguments: None,
},
mcp_app_resource_uri: None,
duration: Duration::from_millis(1),
result: Err("boom".to_string()),
};
let notification = item_event_to_server_notification(
EventMsg::McpToolCallEnd(end_event.clone()),
"thread-4",
"turn_4",
/*is_file_change_output*/ false,
);
assert_item_completed_server_notification(
notification,
ItemCompletedNotification {
thread_id: "thread-4".to_string(),
turn_id: "turn_4".to_string(),
item: ThreadItem::McpToolCall {
id: end_event.call_id,
server: end_event.invocation.server,
tool: end_event.invocation.tool,
status: McpToolCallStatus::Failed,
arguments: JsonValue::Null,
mcp_app_resource_uri: None,
result: None,
error: Some(McpToolCallError {
message: "boom".to_string(),
}),
duration_ms: Some(1),
},
},
);
}
}
codex-rs/app-server-protocol/src/protocol/mod.rs
+1
View File
@@ -2,6 +2,7 @@
// Exposes protocol pieces used by `lib.rs` via `pub use protocol::common::*;`.
pub mod common;
pub mod event_mapping;
pub mod item_builders;
mod mappers;
mod serde_helpers;
codex-rs/app-server/src/bespoke_event_handling.rs
+71 -880
View File
File diff suppressed because it is too large Load Diff
codex-rs/core-api/Cargo.toml
+1
View File
@@ -13,6 +13,7 @@ path = "src/lib.rs"
workspace = true
[dependencies]
codex-app-server-protocol = { workspace = true }
codex-arg0 = { workspace = true }
codex-analytics = { workspace = true }
codex-config = { workspace = true }
codex-rs/core-api/src/lib.rs
+3
View File
@@ -3,6 +3,8 @@
#![deny(private_bounds, private_interfaces, unreachable_pub)]
pub use codex_analytics::AnalyticsEventsClient;
pub use codex_app_server_protocol::ServerNotification;
pub use codex_app_server_protocol::item_event_to_server_notification;
pub use codex_arg0::Arg0DispatchPaths;
pub use codex_arg0::arg0_dispatch_or_else;
pub use codex_config::ConfigLayerStack;
@@ -42,6 +44,7 @@ pub use codex_exec_server::EnvironmentManager;
pub use codex_exec_server::EnvironmentManagerArgs;
pub use codex_exec_server::ExecServerRuntimePaths;
pub use codex_features::Feature;
pub use codex_features::Features;
pub use codex_login::AuthManager;
pub use codex_login::default_client::set_default_originator;
pub use codex_model_provider_info::OPENAI_PROVIDER_ID;
codex-rs/thread-manager-sample/Cargo.toml
+1
View File
@@ -10,6 +10,7 @@ workspace = true
[dependencies]
anyhow = { workspace = true }
clap = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
# Keep this sample limited to a single Codex workspace dependency.
# Add new Codex surface area to `codex-core-api` instead of depending on
# additional `codex-*` crates here.
codex-rs/thread-manager-sample/src/main.rs
+68 -19
View File
@@ -23,6 +23,7 @@ use codex_core_api::EnvironmentManager;
use codex_core_api::EnvironmentManagerArgs;
use codex_core_api::EventMsg;
use codex_core_api::ExecServerRuntimePaths;
use codex_core_api::Features;
use codex_core_api::GhostSnapshotConfig;
use codex_core_api::History;
use codex_core_api::MemoriesConfig;
@@ -53,13 +54,14 @@ use codex_core_api::WebSearchMode;
use codex_core_api::arg0_dispatch_or_else;
use codex_core_api::built_in_model_providers;
use codex_core_api::find_codex_home;
use codex_core_api::item_event_to_server_notification;
use codex_core_api::set_default_originator;
use codex_core_api::thread_store_from_config;
#[derive(Debug, Parser)]
#[command(
name = "codex-thread-manager-sample",
about = "Run one Codex turn through ThreadManager and print the final assistant output."
about = "Run one Codex turn through ThreadManager and print mapped notifications as newline-delimited JSON."
)]
struct Args {
/// Override the model for this run.
@@ -125,19 +127,14 @@ async fn run_main(arg0_paths: Arg0DispatchPaths) -> anyhow::Result<()> {
.await
.context("start Codex thread")?;
let turn_output = run_turn(&thread, prompt).await;
let thread_id_string = thread_id.to_string();
let turn_output = run_turn(&thread, &thread_id_string, prompt).await;
let shutdown_result = thread.shutdown_and_wait().await;
let _ = thread_manager.remove_thread(&thread_id).await;
let output = turn_output?;
turn_output?;
shutdown_result.context("shut down Codex thread")?;
let mut stdout = std::io::stdout().lock();
stdout.write_all(output.as_bytes())?;
if !output.ends_with('\n') {
stdout.write_all(b"\n")?;
}
Ok(())
}
@@ -151,7 +148,7 @@ fn new_config(model: Option<String>, arg0_paths: Arg0DispatchPaths) -> anyhow::R
.context("OpenAI model provider should be available")?
.clone();
Ok(Config {
let mut config = Config {
config_layer_stack: ConfigLayerStack::default(),
startup_warnings: Vec::new(),
model,
@@ -164,7 +161,7 @@ fn new_config(model: Option<String>, arg0_paths: Arg0DispatchPaths) -> anyhow::R
personality: None,
permissions: Permissions {
approval_policy: Constrained::allow_any(AskForApproval::Never),
permission_profile: Constrained::allow_any(PermissionProfile::default()),
permission_profile: Constrained::allow_any(PermissionProfile::read_only()),
active_permission_profile: None,
network: None,
allow_login_shell: true,
@@ -261,10 +258,15 @@ fn new_config(model: Option<String>, arg0_paths: Arg0DispatchPaths) -> anyhow::R
feedback_enabled: false,
tool_suggest: ToolSuggestConfig::default(),
otel: OtelConfig::default(),
})
};
config
.features
.set(Features::with_defaults())
.context("configure default features")?;
Ok(config)
}
async fn run_turn(thread: &CodexThread, prompt: String) -> anyhow::Result<String> {
async fn run_turn(thread: &CodexThread, thread_id: &str, prompt: String) -> anyhow::Result<()> {
thread
.submit(Op::UserInput {
items: vec![UserInput::Text {
@@ -278,15 +280,62 @@ async fn run_turn(thread: &CodexThread, prompt: String) -> anyhow::Result<String
.await
.context("submit user input")?;
let mut last_agent_message = String::new();
let mut current_turn_id: Option<String> = None;
let mut stdout = std::io::stdout().lock();
loop {
let event = thread.next_event().await.context("read Codex event")?;
match event.msg {
EventMsg::TurnComplete(event) => {
return Ok(event.last_agent_message.unwrap_or(last_agent_message));
let notification = match &event.msg {
EventMsg::TurnStarted(event) => {
current_turn_id = Some(event.turn_id.clone());
None
}
EventMsg::AgentMessage(event) => {
last_agent_message = event.message;
EventMsg::DynamicToolCallResponse(_)
| EventMsg::McpToolCallBegin(_)
| EventMsg::McpToolCallEnd(_)
| EventMsg::CollabAgentSpawnBegin(_)
| EventMsg::CollabAgentSpawnEnd(_)
| EventMsg::CollabAgentInteractionBegin(_)
| EventMsg::CollabAgentInteractionEnd(_)
| EventMsg::CollabWaitingBegin(_)
| EventMsg::CollabWaitingEnd(_)
| EventMsg::CollabCloseBegin(_)
| EventMsg::CollabCloseEnd(_)
| EventMsg::CollabResumeBegin(_)
| EventMsg::CollabResumeEnd(_)
| EventMsg::AgentMessageContentDelta(_)
| EventMsg::PlanDelta(_)
| EventMsg::ReasoningContentDelta(_)
| EventMsg::ReasoningRawContentDelta(_)
| EventMsg::AgentReasoningSectionBreak(_)
| EventMsg::ItemStarted(_)
| EventMsg::ItemCompleted(_)
| EventMsg::PatchApplyBegin(_)
| EventMsg::PatchApplyUpdated(_)
| EventMsg::TerminalInteraction(_)
| EventMsg::ExecCommandBegin(_)
| EventMsg::ExecCommandOutputDelta(_)
| EventMsg::ExecCommandEnd(_) => Some(item_event_to_server_notification(
event.msg.clone(),
thread_id,
current_turn_id
.as_deref()
.context("mapped notification arrived before turn started")?,
/*is_file_change_output*/ false,
)),
_ => None,
};
if let Some(notification) = notification {
serde_json::to_writer(&mut stdout, &notification)
.context("serialize mapped notification")?;
stdout
.write_all(b"\n")
.context("write notification newline")?;
stdout.flush().context("flush notification output")?;
}
match event.msg {
EventMsg::TurnComplete(_) => {
return Ok(());
}
EventMsg::Error(event) => {
bail!(event.message);