openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-30 19:18:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge e061022857 into sapling-pr-archive-bolinfest

Browse Source
This commit is contained in:
Michael Bolin
2026-07-06 20:51:42 -07:00
committed by GitHub
parent 5b9fbf1d88 e061022857
commit cf5ca70a7d
32 changed files with 1003 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.codex/environments/environment.toml
+4
View File
@@ -2,6 +2,10 @@
version = 1
name = "codex"
# TODO(anp) make it optional to specify this field
[setup]
script = ""
[[actions]]
name = "Run"
icon = "run"
codex-rs/codex-api/src/common.rs
+21
View File
@@ -104,6 +104,11 @@ pub enum ResponseEvent {
delta: String,
summary_index: i64,
},
ReasoningSummaryDone {
item_id: String,
text: String,
summary_index: i64,
},
ReasoningContentDelta {
delta: String,
content_index: i64,
@@ -148,6 +153,17 @@ pub struct Reasoning {
pub context: Option<ReasoningContext>,
}
#[derive(Debug, Serialize, Clone, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum ReasoningSummaryDelivery {
SequentialCutoff,
}
#[derive(Debug, Serialize, Clone, PartialEq)]
pub struct StreamOptions {
pub reasoning_summary_delivery: ReasoningSummaryDelivery,
}
#[derive(Debug, Serialize, Default, Clone, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum TextFormatType {
@@ -209,6 +225,8 @@ pub struct ResponsesApiRequest {
pub reasoning: Option<Reasoning>,
pub store: bool,
pub stream: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub stream_options: Option<StreamOptions>,
pub include: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub service_tier: Option<String>,
@@ -233,6 +251,7 @@ impl From<&ResponsesApiRequest> for ResponseCreateWsRequest {
reasoning: request.reasoning.clone(),
store: request.store,
stream: request.stream,
stream_options: request.stream_options.clone(),
include: request.include.clone(),
service_tier: request.service_tier.clone(),
prompt_cache_key: request.prompt_cache_key.clone(),
@@ -258,6 +277,8 @@ pub struct ResponseCreateWsRequest {
pub reasoning: Option<Reasoning>,
pub store: bool,
pub stream: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub stream_options: Option<StreamOptions>,
pub include: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub service_tier: Option<String>,
codex-rs/codex-api/src/endpoint/responses_websocket.rs
+1
View File
@@ -851,6 +851,7 @@ mod tests {
reasoning: None,
store: false,
stream: true,
stream_options: None,
include: vec!["reasoning.encrypted_content".to_string()],
service_tier: Some("priority".to_string()),
prompt_cache_key: Some("cache-key".to_string()),
codex-rs/codex-api/src/lib.rs
+2
View File
@@ -34,11 +34,13 @@ pub use crate::common::RawMemory;
pub use crate::common::RawMemoryMetadata;
pub use crate::common::Reasoning;
pub use crate::common::ReasoningContext;
pub use crate::common::ReasoningSummaryDelivery;
pub use crate::common::ResponseCreateWsRequest;
pub use crate::common::ResponseEvent;
pub use crate::common::ResponseStream;
pub use crate::common::ResponsesApiRequest;
pub use crate::common::ResponsesWsRequest;
pub use crate::common::StreamOptions;
pub use crate::common::TextControls;
pub use crate::common::WS_REQUEST_HEADER_TRACEPARENT_CLIENT_METADATA_KEY;
pub use crate::common::WS_REQUEST_HEADER_TRACESTATE_CLIENT_METADATA_KEY;
codex-rs/codex-api/src/sse/responses.rs
+38
View File
@@ -167,6 +167,7 @@ pub struct ResponsesStreamEvent {
item_id: Option<String>,
call_id: Option<String>,
delta: Option<String>,
text: Option<String>,
summary_index: Option<i64>,
content_index: Option<i64>,
safety_buffering: Option<Value>,
@@ -358,6 +359,17 @@ pub fn process_responses_event(
}));
}
}
"response.reasoning_summary_text.done" => {
if let (Some(item_id), Some(text), Some(summary_index)) =
(event.item_id, event.text, event.summary_index)
{
return Ok(Some(ResponseEvent::ReasoningSummaryDone {
item_id,
text,
summary_index,
}));
}
}
"response.reasoning_text.delta" => {
if let (Some(delta), Some(content_index)) = (event.delta, event.content_index) {
return Ok(Some(ResponseEvent::ReasoningContentDelta {
@@ -796,6 +808,32 @@ mod tests {
}
}
#[tokio::test]
async fn parses_reasoning_summary_done() {
let events = run_sse(vec![
json!({
"type": "response.reasoning_summary_text.done",
"item_id": "reasoning-1",
"summary_index": 0,
"text": "Checking",
}),
json!({
"type": "response.completed",
"response": { "id": "resp1" },
}),
])
.await;
assert_matches!(
&events[0],
ResponseEvent::ReasoningSummaryDone {
item_id,
text,
summary_index: 0,
} if item_id == "reasoning-1" && text == "Checking"
);
}
#[tokio::test]
async fn error_when_missing_completed() {
let item1 = json!({
codex-rs/codex-api/tests/clients.rs
+3
View File
@@ -321,6 +321,7 @@ async fn responses_client_stream_request_preserves_item_ids() -> Result<()> {
reasoning: None,
store: false,
stream: true,
stream_options: None,
include: Vec::new(),
service_tier: None,
prompt_cache_key: None,
@@ -407,6 +408,7 @@ async fn streaming_client_retries_on_transport_error() -> Result<()> {
reasoning: None,
store: false,
stream: true,
stream_options: None,
include: Vec::new(),
service_tier: None,
prompt_cache_key: None,
@@ -526,6 +528,7 @@ async fn azure_store_sends_ids_and_headers() -> Result<()> {
reasoning: None,
store: true,
stream: true,
stream_options: None,
include: Vec::new(),
service_tier: None,
prompt_cache_key: None,
codex-rs/core/config.schema.json
+6
View File
@@ -469,6 +469,9 @@
"computer_use": {
"type": "boolean"
},
"concurrent_reasoning_summaries": {
"type": "boolean"
},
"connectors": {
"type": "boolean"
},
@@ -4838,6 +4841,9 @@
"computer_use": {
"type": "boolean"
},
"concurrent_reasoning_summaries": {
"type": "boolean"
},
"connectors": {
"type": "boolean"
},
codex-rs/core/src/client.rs
+15 -1
View File
@@ -56,6 +56,7 @@ use codex_api::ResponsesWebsocketConnection as ApiWebSocketConnection;
use codex_api::ResponsesWsRequest;
use codex_api::SharedAuthProvider;
use codex_api::SseTelemetry;
use codex_api::StreamOptions;
use codex_api::TransportError;
use codex_api::WebsocketTelemetry;
use codex_api::auth_header_telemetry;
@@ -207,6 +208,7 @@ struct ModelClientState {
include_timing_metrics: bool,
beta_features_header: Option<String>,
item_ids_enabled: bool,
concurrent_reasoning_summaries_enabled: bool,
include_attestation: bool,
attestation_provider: Option<Arc<dyn AttestationProvider>>,
disable_websockets: AtomicBool,
@@ -316,6 +318,7 @@ fn responses_request_properties_match(
reasoning: previous_reasoning,
store: previous_store,
stream: previous_stream,
stream_options: _,
include: previous_include,
service_tier: previous_service_tier,
prompt_cache_key: previous_prompt_cache_key,
@@ -332,6 +335,7 @@ fn responses_request_properties_match(
reasoning: current_reasoning,
store: current_store,
stream: current_stream,
stream_options: _,
include: current_include,
service_tier: current_service_tier,
prompt_cache_key: current_prompt_cache_key,
@@ -347,6 +351,8 @@ fn responses_request_properties_match(
&& previous_reasoning == current_reasoning
&& previous_store == current_store
&& previous_stream == current_stream
// Stream options control delivery for this response, not the context
// referenced by `previous_response_id`.
&& previous_include == current_include
&& previous_service_tier == current_service_tier
&& previous_prompt_cache_key == current_prompt_cache_key
@@ -416,6 +422,7 @@ impl ModelClient {
include_timing_metrics: bool,
beta_features_header: Option<String>,
item_ids_enabled: bool,
concurrent_reasoning_summaries_enabled: bool,
attestation_provider: Option<Arc<dyn AttestationProvider>>,
http_client_factory: HttpClientFactory,
) -> Self {
@@ -439,6 +446,7 @@ impl ModelClient {
include_timing_metrics,
beta_features_header,
item_ids_enabled,
concurrent_reasoning_summaries_enabled,
include_attestation,
attestation_provider,
disable_websockets: AtomicBool::new(false),
@@ -830,7 +838,8 @@ impl ModelClient {
responses_metadata: &CodexResponsesMetadata,
) -> Result<ResponsesApiRequest> {
let mut input = prompt.get_formatted_input_for_request(model_info.use_responses_lite);
if !self.state.provider.info().is_openai() {
let is_openai = self.state.provider.info().is_openai();
if !is_openai {
input
.iter_mut()
.for_each(ResponseItem::clear_internal_chat_message_metadata_passthrough);
@@ -858,6 +867,10 @@ impl ModelClient {
} else {
(prompt.base_instructions.text.clone(), Some(tools))
};
let stream_options = (self.state.concurrent_reasoning_summaries_enabled && is_openai)
.then_some(StreamOptions {
reasoning_summary_delivery: codex_api::ReasoningSummaryDelivery::SequentialCutoff,
});
let reasoning = Self::build_reasoning(model_info, effort, summary);
let include = if reasoning.is_some() {
vec!["reasoning.encrypted_content".to_string()]
@@ -892,6 +905,7 @@ impl ModelClient {
reasoning,
store: provider.is_azure_responses_endpoint(),
stream: true,
stream_options,
include,
service_tier,
prompt_cache_key,
codex-rs/core/src/client_common_tests.rs
+4
View File
@@ -116,6 +116,7 @@ fn serializes_text_verbosity_when_set() {
reasoning: None,
store: false,
stream: true,
stream_options: None,
include: vec![],
prompt_cache_key: None,
service_tier: None,
@@ -163,6 +164,7 @@ fn serializes_text_schema_with_strict_format() {
reasoning: None,
store: false,
stream: true,
stream_options: None,
include: vec![],
prompt_cache_key: None,
service_tier: None,
@@ -224,6 +226,7 @@ fn omits_text_when_not_set() {
reasoning: None,
store: false,
stream: true,
stream_options: None,
include: vec![],
prompt_cache_key: None,
service_tier: None,
@@ -247,6 +250,7 @@ fn serializes_flex_service_tier_when_set() {
reasoning: None,
store: false,
stream: true,
stream_options: None,
include: vec![],
prompt_cache_key: None,
service_tier: Some(ServiceTier::Flex.to_string()),
codex-rs/core/src/client_tests.rs
+3
View File
@@ -105,6 +105,7 @@ fn test_model_client_with_thread_id(
/*include_timing_metrics*/ false,
/*beta_features_header*/ None,
/*item_ids_enabled*/ false,
/*concurrent_reasoning_summaries_enabled*/ false,
/*attestation_provider*/ None,
HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
)
@@ -150,6 +151,7 @@ async fn compact_uses_bearer_after_agent_identity_session_fallback() -> anyhow::
/*include_timing_metrics*/ false,
/*beta_features_header*/ None,
/*item_ids_enabled*/ false,
/*concurrent_reasoning_summaries_enabled*/ false,
/*attestation_provider*/ None,
HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
);
@@ -825,6 +827,7 @@ fn model_client_with_counting_attestation(
/*include_timing_metrics*/ false,
/*beta_features_header*/ None,
/*item_ids_enabled*/ false,
/*concurrent_reasoning_summaries_enabled*/ false,
Some(Arc::new(CountingAttestationProvider {
calls: attestation_calls.clone(),
})),
codex-rs/core/src/exec_policy.rs
+9 -2
View File
@@ -554,12 +554,19 @@ pub fn format_exec_policy_error_with_source(error: &ExecPolicyError) -> String {
}
}
async fn load_exec_policy_with_warning(
pub(crate) async fn load_exec_policy_with_warning(
config_stack: &ConfigLayerStack,
) -> Result<(Policy, Option<ExecPolicyError>), ExecPolicyError> {
match load_exec_policy(config_stack).await {
Ok(policy) => Ok((policy, None)),
Err(err @ ExecPolicyError::ParsePolicy { .. }) => Ok((Policy::empty(), Some(err))),
Err(err @ ExecPolicyError::ParsePolicy { .. }) => {
let policy = config_stack
.requirements()
.exec_policy
.as_deref()
.map_or_else(Policy::empty, |policy| policy.as_ref().clone());
Ok((policy, Some(err)))
}
Err(err) => Err(err),
}
}
codex-rs/core/src/exec_policy_tests.rs
+36
View File
@@ -374,6 +374,42 @@ async fn merges_requirements_exec_policy_network_rules() -> anyhow::Result<()> {
Ok(())
}
#[tokio::test]
async fn malformed_custom_rules_preserve_requirements_exec_policy() -> anyhow::Result<()> {
let temp_dir = tempdir()?;
let policy_dir = temp_dir.path().join(RULES_DIR_NAME);
fs::create_dir_all(&policy_dir)?;
fs::write(policy_dir.join("broken.rules"), "prefix_rule(")?;
let mut requirements_exec_policy = Policy::empty();
requirements_exec_policy.add_prefix_rule(&["rm".to_string()], Decision::Forbidden)?;
let requirements = ConfigRequirements {
exec_policy: Some(Sourced::new(
RequirementsExecPolicy::new(requirements_exec_policy),
RequirementSource::Unknown,
)),
..ConfigRequirements::default()
};
let dot_codex_folder = AbsolutePathBuf::from_absolute_path(temp_dir.path())?;
let layer = ConfigLayerEntry::new(
ConfigLayerSource::Project { dot_codex_folder },
TomlValue::Table(Default::default()),
);
let config_stack =
ConfigLayerStack::new(vec![layer], requirements, ConfigRequirementsToml::default())?;
let (policy, warning) = load_exec_policy_with_warning(&config_stack).await?;
assert!(matches!(warning, Some(ExecPolicyError::ParsePolicy { .. })));
assert_eq!(
policy
.check_multiple([vec!["rm".to_string()]].iter(), &|_| Decision::Allow)
.decision,
Decision::Forbidden
);
Ok(())
}
#[tokio::test]
async fn preserves_host_executables_when_requirements_overlay_is_present() -> anyhow::Result<()> {
let temp_dir = tempdir()?;
codex-rs/core/src/network_proxy_loader.rs
+13 -14
View File
@@ -2,9 +2,8 @@ use crate::config::find_codex_home;
use crate::config::is_builtin_permission_profile_name;
use crate::config::reject_unknown_builtin_permission_profile;
use crate::config::resolve_permission_profile;
use crate::exec_policy::ExecPolicyError;
use crate::exec_policy::format_exec_policy_error_with_source;
use crate::exec_policy::load_exec_policy;
use crate::exec_policy::load_exec_policy_with_warning;
use anyhow::Context;
use anyhow::Result;
use codex_config::CONFIG_TOML_FILE;
@@ -64,13 +63,15 @@ async fn build_config_state_with_mtimes() -> Result<(ConfigState, Vec<LayerMtime
.await
.context("failed to load Codex config")?;
let (exec_policy, warning) = match load_exec_policy(&config_layer_stack).await {
Ok(policy) => (policy, None),
Err(err @ ExecPolicyError::ParsePolicy { .. }) => {
(codex_execpolicy::Policy::empty(), Some(err))
}
Err(err) => return Err(err.into()),
};
let layer_mtimes = collect_layer_mtimes(&config_layer_stack);
let state = build_config_state_from_layers(&config_layer_stack).await?;
Ok((state, layer_mtimes))
}
async fn build_config_state_from_layers(
config_layer_stack: &ConfigLayerStack,
) -> Result<ConfigState> {
let (exec_policy, warning) = load_exec_policy_with_warning(config_layer_stack).await?;
if let Some(err) = warning.as_ref() {
tracing::warn!(
"failed to parse execpolicy while building network proxy state: {}",
@@ -78,12 +79,10 @@ async fn build_config_state_with_mtimes() -> Result<(ConfigState, Vec<LayerMtime
);
}
let config = config_from_layers(&config_layer_stack, &exec_policy)?;
let config = config_from_layers(config_layer_stack, &exec_policy)?;
let constraints = enforce_trusted_constraints(&config_layer_stack, &config)?;
let layer_mtimes = collect_layer_mtimes(&config_layer_stack);
let state = build_config_state(config, constraints)?;
Ok((state, layer_mtimes))
let constraints = enforce_trusted_constraints(config_layer_stack, &config)?;
build_config_state(config, constraints)
}
fn collect_layer_mtimes(stack: &ConfigLayerStack) -> Vec<LayerMtime> {
codex-rs/core/src/network_proxy_loader_tests.rs
+50
View File
@@ -5,6 +5,9 @@ use codex_config::ConfigLayerSource;
use codex_config::ConfigLayerStack;
use codex_config::ConfigRequirements;
use codex_config::ConfigRequirementsToml;
use codex_config::RequirementSource;
use codex_config::RequirementsExecPolicy;
use codex_config::Sourced;
use codex_config::permissions_toml::NetworkDomainPermissionToml;
use codex_config::permissions_toml::NetworkDomainPermissionsToml;
use codex_execpolicy::Decision;
@@ -12,6 +15,8 @@ use codex_execpolicy::NetworkRuleProtocol;
use codex_execpolicy::Policy;
use pretty_assertions::assert_eq;
use std::collections::BTreeMap;
use std::fs;
use tempfile::tempdir;
#[test]
fn higher_precedence_profile_network_overlays_domain_entries() {
@@ -225,6 +230,51 @@ fn execpolicy_network_rules_overlay_network_lists() {
);
}
#[tokio::test]
async fn malformed_custom_rules_preserve_managed_denied_domain() {
let temp_dir = tempdir().expect("create temp dir");
let policy_dir = temp_dir.path().join("rules");
fs::create_dir_all(&policy_dir).expect("create policy dir");
fs::write(policy_dir.join("broken.rules"), "prefix_rule(").expect("write malformed policy");
let mut requirements_exec_policy = Policy::empty();
requirements_exec_policy
.add_network_rule(
"blocked.example.com",
NetworkRuleProtocol::Https,
Decision::Forbidden,
/*justification*/ None,
)
.expect("managed network rule should be valid");
let requirements = ConfigRequirements {
exec_policy: Some(Sourced::new(
RequirementsExecPolicy::new(requirements_exec_policy),
RequirementSource::Unknown,
)),
..ConfigRequirements::default()
};
let dot_codex_folder = AbsolutePathBuf::from_absolute_path(temp_dir.path())
.expect("dot codex folder should be absolute");
let layers = ConfigLayerStack::new(
vec![ConfigLayerEntry::new(
ConfigLayerSource::Project { dot_codex_folder },
toml::Value::Table(Default::default()),
)],
requirements,
ConfigRequirementsToml::default(),
)
.expect("layer stack should be valid");
let state = build_config_state_from_layers(&layers)
.await
.expect("proxy state should tolerate malformed custom rules");
assert_eq!(
state.config.network.denied_domains(),
Some(vec!["blocked.example.com".to_string()])
);
}
#[test]
fn apply_network_constraints_includes_allow_all_unix_sockets_flag() {
let config: toml::Value = toml::from_str(
codex-rs/core/src/session/mod.rs
+3
View File
@@ -1794,6 +1794,9 @@ impl Session {
let show_raw_agent_reasoning = self.show_raw_agent_reasoning();
for legacy in legacy_source.as_legacy_events(show_raw_agent_reasoning) {
self.services
.rollout_thread_trace
.record_tool_call_event(turn_context.sub_id.clone(), &legacy);
let legacy_event = Event {
id: turn_context.sub_id.clone(),
msg: legacy,
codex-rs/core/src/session/session.rs
+3
View File
@@ -1111,6 +1111,9 @@ impl Session {
config.features.enabled(Feature::RuntimeMetrics),
Self::build_model_client_beta_features_header(config.as_ref()),
/*item_ids_enabled*/ config.features.enabled(Feature::ItemIds),
/*concurrent_reasoning_summaries_enabled*/ config
.features
.enabled(Feature::ConcurrentReasoningSummaries),
attestation_provider,
config.http_client_factory(),
)
codex-rs/core/src/session/tests.rs
+9
View File
@@ -492,6 +492,7 @@ fn test_model_client_session() -> crate::client::ModelClientSession {
/*include_timing_metrics*/ false,
/*beta_features_header*/ None,
/*item_ids_enabled*/ false,
/*concurrent_reasoning_summaries_enabled*/ false,
/*attestation_provider*/ None,
HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
)
@@ -5463,6 +5464,10 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
config.features.enabled(Feature::RuntimeMetrics),
Session::build_model_client_beta_features_header(config.as_ref()),
/*item_ids_enabled*/ config.features.enabled(Feature::ItemIds),
/*concurrent_reasoning_summaries_enabled*/
config
.features
.enabled(Feature::ConcurrentReasoningSummaries),
/*attestation_provider*/ None,
config.http_client_factory(),
),
@@ -7590,6 +7595,10 @@ where
config.features.enabled(Feature::RuntimeMetrics),
Session::build_model_client_beta_features_header(config.as_ref()),
/*item_ids_enabled*/ config.features.enabled(Feature::ItemIds),
/*concurrent_reasoning_summaries_enabled*/
config
.features
.enabled(Feature::ConcurrentReasoningSummaries),
/*attestation_provider*/ None,
config.http_client_factory(),
),
codex-rs/core/src/session/turn.rs
+45
View File
@@ -1909,6 +1909,11 @@ async fn try_run_sampling_request(
turn_context.provider.info().name.as_str(),
);
let sampling_timing_guard = turn_context.turn_timing_state.begin_sampling();
let uses_sequential_cutoff_reasoning_summaries = turn_context
.config
.features
.enabled(Feature::ConcurrentReasoningSummaries)
&& turn_context.provider.info().is_openai();
let mut stream = client_session
.stream(
prompt,
@@ -2291,6 +2296,9 @@ async fn try_run_sampling_request(
delta,
summary_index,
} => {
if uses_sequential_cutoff_reasoning_summaries {
continue;
}
if let Some(active) = active_item.as_ref() {
if !active_item_is_streaming_to_client {
continue;
@@ -2309,6 +2317,9 @@ async fn try_run_sampling_request(
}
}
ResponseEvent::ReasoningSummaryPartAdded { summary_index } => {
if uses_sequential_cutoff_reasoning_summaries {
continue;
}
if let Some(active) = active_item.as_ref() {
if !active_item_is_streaming_to_client {
continue;
@@ -2323,6 +2334,40 @@ async fn try_run_sampling_request(
error_or_panic("ReasoningSummaryPartAdded without active item".to_string());
}
}
ResponseEvent::ReasoningSummaryDone {
item_id,
text,
summary_index,
} => {
if !uses_sequential_cutoff_reasoning_summaries {
continue;
}
let Some(active) = active_item.as_ref() else {
continue;
};
if !active_item_is_streaming_to_client || active.id() != item_id {
continue;
}
if summary_index > 0 {
sess.send_event(
&turn_context,
EventMsg::AgentReasoningSectionBreak(AgentReasoningSectionBreakEvent {
item_id: item_id.clone(),
summary_index,
}),
)
.await;
}
let event = ReasoningContentDeltaEvent {
thread_id: sess.thread_id.to_string(),
turn_id: turn_context.sub_id.clone(),
item_id,
delta: text,
summary_index,
};
sess.send_event(&turn_context, EventMsg::ReasoningContentDelta(event))
.await;
}
ResponseEvent::ReasoningContentDelta {
delta,
content_index,
codex-rs/core/src/turn_timing.rs
+1
View File
@@ -334,6 +334,7 @@ fn response_event_records_turn_ttft(event: &ResponseEvent) -> bool {
}
ResponseEvent::OutputTextDelta(_)
| ResponseEvent::ReasoningSummaryDelta { .. }
| ResponseEvent::ReasoningSummaryDone { .. }
| ResponseEvent::ReasoningContentDelta { .. } => true,
ResponseEvent::Created
| ResponseEvent::ServerModel(_)
codex-rs/core/tests/responses_headers.rs
+3
View File
@@ -131,6 +131,7 @@ async fn responses_stream_includes_subagent_header_on_review() {
/*include_timing_metrics*/ false,
/*beta_features_header*/ None,
/*item_ids_enabled*/ false,
/*concurrent_reasoning_summaries_enabled*/ false,
/*attestation_provider*/ None,
config.http_client_factory(),
);
@@ -266,6 +267,7 @@ async fn responses_stream_includes_subagent_header_on_other() {
/*include_timing_metrics*/ false,
/*beta_features_header*/ None,
/*item_ids_enabled*/ false,
/*concurrent_reasoning_summaries_enabled*/ false,
/*attestation_provider*/ None,
config.http_client_factory(),
);
@@ -387,6 +389,7 @@ async fn responses_respects_model_info_overrides_from_config() {
/*include_timing_metrics*/ false,
/*beta_features_header*/ None,
/*item_ids_enabled*/ false,
/*concurrent_reasoning_summaries_enabled*/ false,
/*attestation_provider*/ None,
config.http_client_factory(),
);
codex-rs/core/tests/suite/client.rs
+58
View File
@@ -1334,6 +1334,10 @@ async fn send_provider_auth_request(server: &MockServer, auth: ModelProviderAuth
/*include_timing_metrics*/ false,
/*beta_features_header*/ None,
/*item_ids_enabled*/ config.features.enabled(Feature::ItemIds),
/*concurrent_reasoning_summaries_enabled*/
config
.features
.enabled(Feature::ConcurrentReasoningSummaries),
/*attestation_provider*/ None,
config.http_client_factory(),
);
@@ -2377,6 +2381,9 @@ async fn configured_reasoning_summary_is_sent() -> anyhow::Result<()> {
let TestCodex { codex, .. } = test_codex()
.with_config(|config| {
config.model_reasoning_summary = Some(ReasoningSummary::Concise);
let _ = config
.features
.enable(Feature::ConcurrentReasoningSummaries);
})
.build(&server)
.await?;
@@ -2413,6 +2420,55 @@ async fn configured_reasoning_summary_is_sent() -> anyhow::Result<()> {
.and_then(|reasoning| reasoning.get("context")),
None
);
pretty_assertions::assert_eq!(
request_body
.get("stream_options")
.and_then(|stream_options| stream_options.get("reasoning_summary_delivery"))
.and_then(|value| value.as_str()),
Some("sequential_cutoff")
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn sequential_cutoff_is_omitted_for_non_openai_provider() -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
let server = MockServer::start().await;
let resp_mock = mount_sse_once(
&server,
sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
)
.await;
let TestCodex { codex, .. } = test_codex()
.with_config(|config| {
config.model_provider.name = "mock".to_string();
let _ = config
.features
.enable(Feature::ConcurrentReasoningSummaries);
})
.build(&server)
.await?;
codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
responsesapi_client_metadata: None,
additional_context: Default::default(),
thread_settings: Default::default(),
})
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let request_body = resp_mock.single_request().body_json();
pretty_assertions::assert_eq!(request_body.get("stream_options"), None);
Ok(())
}
@@ -2584,6 +2640,7 @@ async fn reasoning_summary_is_omitted_when_disabled() -> anyhow::Result<()> {
.and_then(|reasoning| reasoning.get("summary")),
None
);
pretty_assertions::assert_eq!(request_body.get("stream_options"), None);
Ok(())
}
@@ -2949,6 +3006,7 @@ async fn azure_responses_request_includes_store_and_reasoning_ids() {
/*include_timing_metrics*/ false,
/*beta_features_header*/ None,
/*item_ids_enabled*/ false,
/*concurrent_reasoning_summaries_enabled*/ false,
/*attestation_provider*/ None,
config.http_client_factory(),
);
codex-rs/core/tests/suite/client_websockets.rs
+37 -5
View File
@@ -386,7 +386,13 @@ async fn responses_websocket_request_prewarm_reuses_connection() {
]])
.await;
let harness = websocket_harness_with_options(&server, /*runtime_metrics_enabled*/ true).await;
let mut provider = websocket_provider(&server);
provider.name = ModelProviderInfo::create_openai_provider(/*base_url*/ None).name;
let harness = websocket_harness_with_provider_options(
provider, /*runtime_metrics_enabled*/ true,
/*concurrent_reasoning_summaries_enabled*/ true,
)
.await;
let mut client_session = harness.client.new_session();
let prompt = prompt_with_input(vec![message_item("hello")]);
let responses_metadata = prewarm_metadata(&harness, /*turn_id*/ None);
@@ -422,6 +428,10 @@ async fn responses_websocket_request_prewarm_reuses_connection() {
assert_eq!(warmup["type"].as_str(), Some("response.create"));
assert_eq!(warmup["generate"].as_bool(), Some(false));
assert_eq!(
warmup["stream_options"]["reasoning_summary_delivery"].as_str(),
Some("sequential_cutoff")
);
assert_eq!(warmup["tools"], serde_json::json!([]));
let warmup_turn_metadata: serde_json::Value = serde_json::from_str(
warmup["client_metadata"]["x-codex-turn-metadata"]
@@ -436,6 +446,10 @@ async fn responses_websocket_request_prewarm_reuses_connection() {
assert_eq!(follow_up["type"].as_str(), Some("response.create"));
assert_eq!(follow_up["previous_response_id"].as_str(), Some("warm-1"));
assert_eq!(follow_up["input"], serde_json::json!([]));
assert_eq!(
follow_up["stream_options"]["reasoning_summary_delivery"].as_str(),
Some("sequential_cutoff")
);
server.shutdown().await;
}
@@ -990,8 +1004,11 @@ async fn responses_websocket_v2_incremental_requests_are_reused_across_turns() {
// use OpenAI provider to check metadata logic
let mut provider = websocket_provider(&server);
provider.name = ModelProviderInfo::create_openai_provider(/*base_url*/ None).name;
let harness =
websocket_harness_with_provider_options(provider, /*runtime_metrics_enabled*/ false).await;
let harness = websocket_harness_with_provider_options(
provider, /*runtime_metrics_enabled*/ false,
/*concurrent_reasoning_summaries_enabled*/ false,
)
.await;
// Turn one: initiate
let prompt_one = prompt_with_input(vec![message_item("hello")]);
@@ -2195,13 +2212,18 @@ async fn websocket_harness_with_options(
server: &WebSocketTestServer,
runtime_metrics_enabled: bool,
) -> WebsocketTestHarness {
websocket_harness_with_provider_options(websocket_provider(server), runtime_metrics_enabled)
.await
websocket_harness_with_provider_options(
websocket_provider(server),
runtime_metrics_enabled,
/*concurrent_reasoning_summaries_enabled*/ false,
)
.await
}
async fn websocket_harness_with_provider_options(
provider: ModelProviderInfo,
runtime_metrics_enabled: bool,
concurrent_reasoning_summaries_enabled: bool,
) -> WebsocketTestHarness {
let codex_home = TempDir::new().unwrap();
let mut config = load_default_config_for_test(&codex_home).await;
@@ -2212,6 +2234,12 @@ async fn websocket_harness_with_provider_options(
.enable(Feature::RuntimeMetrics)
.expect("test config should allow feature update");
}
if concurrent_reasoning_summaries_enabled {
config
.features
.enable(Feature::ConcurrentReasoningSummaries)
.expect("test config should allow feature update");
}
let config = Arc::new(config);
let model_info = codex_core::test_support::construct_model_info_offline(MODEL, &config);
let thread_id = ThreadId::new();
@@ -2251,6 +2279,10 @@ async fn websocket_harness_with_provider_options(
runtime_metrics_enabled,
/*beta_features_header*/ None,
/*item_ids_enabled*/ config.features.enabled(Feature::ItemIds),
/*concurrent_reasoning_summaries_enabled*/
config
.features
.enabled(Feature::ConcurrentReasoningSummaries),
/*attestation_provider*/ None,
config.http_client_factory(),
);
codex-rs/core/tests/suite/exec_policy.rs
+80
View File
@@ -1,6 +1,7 @@
#![allow(clippy::unwrap_used)]
use anyhow::Result;
use codex_config::test_support::CloudConfigBundleFixture;
use codex_features::Feature;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::ModeKind;
@@ -17,6 +18,7 @@ use core_test_support::responses::ev_response_created;
use core_test_support::responses::mount_sse_once;
use core_test_support::responses::sse;
use core_test_support::responses::start_mock_server;
use core_test_support::skip_if_target_windows;
use core_test_support::test_codex::local_selections;
use core_test_support::test_codex::test_codex;
use core_test_support::test_codex::turn_permission_fields;
@@ -255,6 +257,84 @@ async fn execpolicy_blocks_shell_invocation() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn malformed_custom_rules_preserve_managed_forbidden_prefix() -> Result<()> {
skip_if_target_windows!(
Ok(()),
"managed prefix fixture uses POSIX executable semantics"
);
let mut builder = test_codex()
.with_cloud_config_bundle(
CloudConfigBundleFixture::loader_with_enterprise_requirement(
r#"
[rules]
prefix_rules = [
{ pattern = [{ token = "echo" }], decision = "forbidden" },
]
"#,
),
)
.with_config(|config| {
config
.features
.enable(Feature::UnifiedExec)
.expect("test config should allow feature update");
let policy_path = config.codex_home.join("rules").join("broken.rules");
fs::create_dir_all(
policy_path
.parent()
.expect("policy directory must have a parent"),
)
.expect("create policy directory");
fs::write(policy_path, "prefix_rule(").expect("write malformed policy file");
});
let server = start_mock_server().await;
let test = builder.build_with_auto_env(&server).await?;
let call_id = "managed-shell-forbidden";
let args = json!({
"cmd": "echo blocked",
"yield_time_ms": 1_000,
});
mount_sse_once(
&server,
sse(vec![
ev_response_created("resp-managed-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-managed-1"),
]),
)
.await;
let results_mock = mount_sse_once(
&server,
sse(vec![
ev_assistant_message("msg-managed-1", "done"),
ev_completed("resp-managed-2"),
]),
)
.await;
test.submit_turn_with_approval_and_permission_profile(
"run shell command",
AskForApproval::Never,
PermissionProfile::Disabled,
)
.await?;
let output_item = results_mock.single_request().function_call_output(call_id);
let output = output_item
.get("output")
.and_then(Value::as_str)
.expect("function call output should include a string output payload");
assert!(
output.contains("policy forbids commands starting with `echo`"),
"unexpected output: {output}"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn shell_command_empty_script_with_collaboration_mode_does_not_panic() -> Result<()> {
let server = start_mock_server().await;
codex-rs/core/tests/suite/items.rs
+93
View File
@@ -1,6 +1,7 @@
#![cfg(not(target_os = "windows"))]
use anyhow::Ok;
use codex_features::Feature;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Settings;
@@ -1134,6 +1135,98 @@ async fn reasoning_content_delta_has_item_metadata() -> anyhow::Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn sequential_cutoff_renders_done_summaries_for_active_reasoning_item() -> anyhow::Result<()>
{
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let TestCodex { codex, .. } = test_codex()
.with_config(|config| {
config
.features
.enable(Feature::ConcurrentReasoningSummaries)
.expect("test config should allow feature update");
})
.build(&server)
.await?;
let summary_done = |item_id: &str, summary_index: i64, text: &str| {
serde_json::json!({
"type": "response.reasoning_summary_text.done",
"item_id": item_id,
"summary_index": summary_index,
"text": text,
})
};
mount_sse_once(
&server,
sse(vec![
ev_response_created("resp-1"),
ev_reasoning_item_added("reasoning-1", &[""]),
// Sequential cutoff renders atomic done events, not partial legacy events.
ev_reasoning_summary_text_delta("partial"),
summary_done("reasoning-1", 0, "step one"),
summary_done("reasoning-1", 1, "step two"),
ev_reasoning_item("reasoning-1", &["step one", "step two"], &[]),
ev_message_item_added("message-1", ""),
// Drop a stale completion rather than attaching it to the active message.
summary_done("reasoning-1", 2, "late step"),
ev_output_text_delta("Done"),
ev_assistant_message("message-1", "Done"),
ev_completed("resp-1"),
]),
)
.await;
codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: "reason through it".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
responsesapi_client_metadata: None,
additional_context: Default::default(),
thread_settings: Default::default(),
})
.await?;
let reasoning_item = wait_for_event_match(&codex, |ev| match ev {
EventMsg::ItemStarted(ItemStartedEvent {
item: TurnItem::Reasoning(item),
..
}) => Some(item.clone()),
_ => None,
})
.await;
let mut summary_deltas = Vec::new();
let mut summary_sections = Vec::new();
loop {
match wait_for_event(&codex, |_| true).await {
EventMsg::ReasoningContentDelta(event) => {
summary_deltas.push((event.item_id, event.delta, event.summary_index));
}
EventMsg::AgentReasoningSectionBreak(event) => {
summary_sections.push((event.item_id, event.summary_index));
}
EventMsg::TurnComplete(_) => break,
_ => {}
}
}
assert_eq!(
summary_deltas,
vec![
(reasoning_item.id.clone(), "step one".to_string(), 0),
(reasoning_item.id, "step two".to_string(), 1),
]
);
assert_eq!(summary_sections, vec![("reasoning-1".to_string(), 1)]);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reasoning_raw_content_delta_respects_flag() -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
codex-rs/exec-server/Cargo.toml
+1
View File
@@ -74,6 +74,7 @@ rustls = { workspace = true }
serial_test = { workspace = true }
tempfile = { workspace = true }
test-case = "3.3.1"
tokio = { workspace = true, features = ["test-util"] }
tracing-opentelemetry = { workspace = true }
tracing-subscriber = { workspace = true }
wiremock = { workspace = true }
codex-rs/exec-server/src/noise_relay/harness_tests.rs
+3 -3
View File
@@ -31,7 +31,7 @@ use crate::noise_channel::PendingResponderHandshake;
const ENVIRONMENT_ID: &str = "environment-1";
const EXECUTOR_REGISTRATION_ID: &str = "registration-1";
#[tokio::test]
#[tokio::test(start_paused = true)]
async fn fragmented_writes_yield_to_keepalive_and_queued_pong() -> Result<()> {
let (connection, mut control, mut outbound_rx) = connected_controlled_harness().await?;
@@ -48,7 +48,7 @@ async fn fragmented_writes_yield_to_keepalive_and_queued_pong() -> Result<()> {
.await?;
control.wait_for_blocked_write(/*expected*/ 1).await?;
tokio::time::sleep(WEBSOCKET_KEEPALIVE_INTERVAL + Duration::from_millis(10)).await;
tokio::time::advance(WEBSOCKET_KEEPALIVE_INTERVAL + Duration::from_millis(10)).await;
control.grant_writes(/*count*/ 1);
let first_data = read_outbound_data(&mut outbound_rx).await?;
assert_eq!(first_data.seq, 0);
@@ -64,7 +64,7 @@ async fn fragmented_writes_yield_to_keepalive_and_queued_pong() -> Result<()> {
control.wait_for_blocked_write(/*expected*/ 3).await?;
control.send_inbound(Message::Pong(ping_payload))?;
tokio::time::sleep(WEBSOCKET_KEEPALIVE_INTERVAL + Duration::from_millis(10)).await;
tokio::time::advance(WEBSOCKET_KEEPALIVE_INTERVAL + Duration::from_millis(10)).await;
control.grant_writes(/*count*/ 1);
let second_data = read_outbound_data(&mut outbound_rx).await?;
assert_eq!(second_data.seq, 1);
codex-rs/features/src/lib.rs
+8
View File
@@ -204,6 +204,8 @@ pub enum Feature {
ResizeAllImages,
/// Generate Responses API item IDs for client-created history items.
ItemIds,
/// Request sequential cutoff reasoning summary delivery.
ConcurrentReasoningSummaries,
/// Allow prompting and installing missing MCP dependencies.
SkillMcpDependencyInstall,
/// Removed compatibility flag for deleted skill env var dependency prompting.
@@ -1184,6 +1186,12 @@ pub const FEATURES: &[FeatureSpec] = &[
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::ConcurrentReasoningSummaries,
key: "concurrent_reasoning_summaries",
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::SkillMcpDependencyInstall,
key: "skill_mcp_dependency_install",
codex-rs/memories/write/src/runtime.rs
+1
View File
@@ -261,6 +261,7 @@ impl MemoryStartupContext {
config.features.enabled(Feature::RuntimeMetrics),
/*beta_features_header*/ None,
config.features.enabled(Feature::ItemIds),
/*concurrent_reasoning_summaries_enabled*/ false,
/*attestation_provider*/ None,
config.http_client_factory(),
);
codex-rs/otel/src/events/session_telemetry.rs
+1
View File
@@ -1195,6 +1195,7 @@ impl SessionTelemetry {
ResponseEvent::OutputTextDelta(_) => "text_delta".into(),
ResponseEvent::ToolCallInputDelta { .. } => "tool_input_delta".into(),
ResponseEvent::ReasoningSummaryDelta { .. } => "reasoning_summary_delta".into(),
ResponseEvent::ReasoningSummaryDone { .. } => "reasoning_summary_done".into(),
ResponseEvent::ReasoningContentDelta { .. } => "reasoning_content_delta".into(),
ResponseEvent::ReasoningSummaryPartAdded { .. } => {
"reasoning_summary_part_added".into()
codex-rs/protocol/src/items.rs
+11
View File
@@ -13,6 +13,7 @@ use crate::parse_command::ParsedCommand;
use crate::protocol::AgentStatus;
use crate::protocol::CollabAgentRef;
use crate::protocol::ExecCommandSource;
use crate::protocol::ExecCommandStatus;
use crate::protocol::FileChange;
use crate::protocol::PatchApplyStatus;
use crate::protocol::SubAgentActivityKind;
@@ -137,6 +138,16 @@ pub enum CommandExecutionStatus {
Declined,
}
impl From<ExecCommandStatus> for CommandExecutionStatus {
fn from(value: ExecCommandStatus) -> Self {
match value {
ExecCommandStatus::Completed => Self::Completed,
ExecCommandStatus::Failed => Self::Failed,
ExecCommandStatus::Declined => Self::Declined,
}
}
}
#[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema, PartialEq)]
pub struct CommandExecutionItem {
pub id: String,
codex-rs/protocol/src/legacy_events.rs
+304
View File
@@ -1,10 +1,20 @@
use crate::ThreadId;
use crate::dynamic_tools::DynamicToolCallRequest;
use crate::items::AgentMessageContent;
use crate::items::AgentMessageItem;
use crate::items::CollabAgentTool;
use crate::items::CollabAgentToolCallItem;
use crate::items::CollabAgentToolCallStatus;
use crate::items::CommandExecutionItem;
use crate::items::CommandExecutionStatus;
use crate::items::ContextCompactionItem;
use crate::items::DynamicToolCallItem;
use crate::items::DynamicToolCallStatus;
use crate::items::FileChangeItem;
use crate::items::ImageGenerationItem;
use crate::items::McpToolCallItem;
use crate::items::ReasoningItem;
use crate::items::SubAgentActivityItem;
use crate::items::TurnItem;
use crate::items::UserMessageItem;
use crate::items::WebSearchItem;
@@ -12,8 +22,24 @@ use crate::protocol::AgentMessageContentDeltaEvent;
use crate::protocol::AgentMessageEvent;
use crate::protocol::AgentReasoningEvent;
use crate::protocol::AgentReasoningRawContentEvent;
use crate::protocol::AgentStatus;
use crate::protocol::CollabAgentInteractionBeginEvent;
use crate::protocol::CollabAgentInteractionEndEvent;
use crate::protocol::CollabAgentSpawnBeginEvent;
use crate::protocol::CollabAgentSpawnEndEvent;
use crate::protocol::CollabAgentStatusEntry;
use crate::protocol::CollabCloseBeginEvent;
use crate::protocol::CollabCloseEndEvent;
use crate::protocol::CollabResumeBeginEvent;
use crate::protocol::CollabResumeEndEvent;
use crate::protocol::CollabWaitingBeginEvent;
use crate::protocol::CollabWaitingEndEvent;
use crate::protocol::ContextCompactedEvent;
use crate::protocol::DynamicToolCallResponseEvent;
use crate::protocol::EventMsg;
use crate::protocol::ExecCommandBeginEvent;
use crate::protocol::ExecCommandEndEvent;
use crate::protocol::ExecCommandStatus;
use crate::protocol::ImageGenerationBeginEvent;
use crate::protocol::ImageGenerationEndEvent;
use crate::protocol::ItemCompletedEvent;
@@ -26,6 +52,7 @@ use crate::protocol::PatchApplyEndEvent;
use crate::protocol::PatchApplyStatus;
use crate::protocol::ReasoningContentDeltaEvent;
use crate::protocol::ReasoningRawContentDeltaEvent;
use crate::protocol::SubAgentActivityEvent;
use crate::protocol::UserMessageEvent;
use crate::protocol::ViewImageToolCallEvent;
use crate::protocol::WebSearchBeginEvent;
@@ -97,6 +124,258 @@ impl ReasoningItem {
}
}
impl CommandExecutionItem {
pub(crate) fn as_legacy_begin_event(&self, turn_id: String, started_at_ms: i64) -> EventMsg {
EventMsg::ExecCommandBegin(ExecCommandBeginEvent {
call_id: self.id.clone(),
process_id: self.process_id.clone(),
turn_id,
started_at_ms,
command: self.command.clone(),
cwd: self.cwd.clone(),
parsed_cmd: self.parsed_cmd.clone(),
source: self.source,
interaction_input: self.interaction_input.clone(),
})
}
pub(crate) fn as_legacy_end_event(
&self,
turn_id: String,
completed_at_ms: i64,
) -> Option<EventMsg> {
let status = match self.status {
CommandExecutionStatus::InProgress => return None,
CommandExecutionStatus::Completed => ExecCommandStatus::Completed,
CommandExecutionStatus::Failed => ExecCommandStatus::Failed,
CommandExecutionStatus::Declined => ExecCommandStatus::Declined,
};
Some(EventMsg::ExecCommandEnd(ExecCommandEndEvent {
call_id: self.id.clone(),
process_id: self.process_id.clone(),
turn_id,
completed_at_ms,
command: self.command.clone(),
cwd: self.cwd.clone(),
parsed_cmd: self.parsed_cmd.clone(),
source: self.source,
interaction_input: self.interaction_input.clone(),
stdout: self.stdout.clone().unwrap_or_default(),
stderr: self.stderr.clone().unwrap_or_default(),
aggregated_output: self.aggregated_output.clone().unwrap_or_default(),
exit_code: self.exit_code.unwrap_or_default(),
duration: self.duration.unwrap_or_default(),
formatted_output: self.formatted_output.clone().unwrap_or_default(),
status,
}))
}
}
impl DynamicToolCallItem {
pub(crate) fn as_legacy_request_event(&self, turn_id: String, started_at_ms: i64) -> EventMsg {
EventMsg::DynamicToolCallRequest(DynamicToolCallRequest {
call_id: self.id.clone(),
turn_id,
started_at_ms,
namespace: self.namespace.clone(),
tool: self.tool.clone(),
arguments: self.arguments.clone(),
})
}
pub(crate) fn as_legacy_response_event(
&self,
turn_id: String,
completed_at_ms: i64,
) -> Option<EventMsg> {
if matches!(self.status, DynamicToolCallStatus::InProgress) {
return None;
}
Some(EventMsg::DynamicToolCallResponse(
DynamicToolCallResponseEvent {
call_id: self.id.clone(),
turn_id,
completed_at_ms,
namespace: self.namespace.clone(),
tool: self.tool.clone(),
arguments: self.arguments.clone(),
content_items: self.content_items.clone().unwrap_or_default(),
success: self.success.unwrap_or(false),
error: self.error.clone(),
duration: self.duration.unwrap_or_default(),
},
))
}
}
impl CollabAgentToolCallItem {
pub(crate) fn as_legacy_begin_event(&self, started_at_ms: i64) -> Option<EventMsg> {
let receiver_thread_id = self.receiver_thread_ids.first().copied();
match self.tool {
CollabAgentTool::SpawnAgent => Some(EventMsg::CollabAgentSpawnBegin(
CollabAgentSpawnBeginEvent {
call_id: self.id.clone(),
started_at_ms,
sender_thread_id: self.sender_thread_id,
prompt: self.prompt.clone().unwrap_or_default(),
model: self.model.clone().unwrap_or_default(),
reasoning_effort: self.reasoning_effort.clone().unwrap_or_default(),
},
)),
CollabAgentTool::SendInput => receiver_thread_id.map(|receiver_thread_id| {
EventMsg::CollabAgentInteractionBegin(CollabAgentInteractionBeginEvent {
call_id: self.id.clone(),
started_at_ms,
sender_thread_id: self.sender_thread_id,
receiver_thread_id,
prompt: self.prompt.clone().unwrap_or_default(),
})
}),
CollabAgentTool::ResumeAgent => receiver_thread_id.map(|receiver_thread_id| {
let (receiver_agent_nickname, receiver_agent_role) =
self.receiver_agent_identity(receiver_thread_id);
EventMsg::CollabResumeBegin(CollabResumeBeginEvent {
call_id: self.id.clone(),
started_at_ms,
sender_thread_id: self.sender_thread_id,
receiver_thread_id,
receiver_agent_nickname,
receiver_agent_role,
})
}),
CollabAgentTool::Wait => Some(EventMsg::CollabWaitingBegin(CollabWaitingBeginEvent {
started_at_ms,
sender_thread_id: self.sender_thread_id,
receiver_thread_ids: self.receiver_thread_ids.clone(),
receiver_agents: self.receiver_agents.clone(),
call_id: self.id.clone(),
})),
CollabAgentTool::CloseAgent => receiver_thread_id.map(|receiver_thread_id| {
EventMsg::CollabCloseBegin(CollabCloseBeginEvent {
call_id: self.id.clone(),
started_at_ms,
sender_thread_id: self.sender_thread_id,
receiver_thread_id,
})
}),
}
}
pub(crate) fn as_legacy_end_event(&self, completed_at_ms: i64) -> Option<EventMsg> {
if matches!(self.status, CollabAgentToolCallStatus::InProgress) {
return None;
}
let receiver_thread_id = self.receiver_thread_ids.first().copied();
match self.tool {
CollabAgentTool::SpawnAgent => {
let (new_agent_nickname, new_agent_role) = receiver_thread_id
.map(|thread_id| self.receiver_agent_identity(thread_id))
.unwrap_or_default();
Some(EventMsg::CollabAgentSpawnEnd(CollabAgentSpawnEndEvent {
call_id: self.id.clone(),
completed_at_ms,
sender_thread_id: self.sender_thread_id,
new_thread_id: receiver_thread_id,
new_agent_nickname,
new_agent_role,
prompt: self.prompt.clone().unwrap_or_default(),
model: self.model.clone().unwrap_or_default(),
reasoning_effort: self.reasoning_effort.clone().unwrap_or_default(),
status: receiver_thread_id
.map(|thread_id| self.agent_status(thread_id))
.unwrap_or(AgentStatus::NotFound),
}))
}
CollabAgentTool::SendInput => receiver_thread_id.map(|receiver_thread_id| {
let (receiver_agent_nickname, receiver_agent_role) =
self.receiver_agent_identity(receiver_thread_id);
EventMsg::CollabAgentInteractionEnd(CollabAgentInteractionEndEvent {
call_id: self.id.clone(),
completed_at_ms,
sender_thread_id: self.sender_thread_id,
receiver_thread_id,
receiver_agent_nickname,
receiver_agent_role,
prompt: self.prompt.clone().unwrap_or_default(),
status: self.agent_status(receiver_thread_id),
})
}),
CollabAgentTool::ResumeAgent => receiver_thread_id.map(|receiver_thread_id| {
let (receiver_agent_nickname, receiver_agent_role) =
self.receiver_agent_identity(receiver_thread_id);
EventMsg::CollabResumeEnd(CollabResumeEndEvent {
call_id: self.id.clone(),
completed_at_ms,
sender_thread_id: self.sender_thread_id,
receiver_thread_id,
receiver_agent_nickname,
receiver_agent_role,
status: self.agent_status(receiver_thread_id),
})
}),
CollabAgentTool::Wait => Some(EventMsg::CollabWaitingEnd(CollabWaitingEndEvent {
sender_thread_id: self.sender_thread_id,
call_id: self.id.clone(),
completed_at_ms,
agent_statuses: self
.receiver_agents
.iter()
.map(|agent| CollabAgentStatusEntry {
thread_id: agent.thread_id,
agent_nickname: agent.agent_nickname.clone(),
agent_role: agent.agent_role.clone(),
status: self.agent_status(agent.thread_id),
})
.collect(),
statuses: self.agents_states.clone(),
})),
CollabAgentTool::CloseAgent => receiver_thread_id.map(|receiver_thread_id| {
let (receiver_agent_nickname, receiver_agent_role) =
self.receiver_agent_identity(receiver_thread_id);
EventMsg::CollabCloseEnd(CollabCloseEndEvent {
call_id: self.id.clone(),
completed_at_ms,
sender_thread_id: self.sender_thread_id,
receiver_thread_id,
receiver_agent_nickname,
receiver_agent_role,
status: self.agent_status(receiver_thread_id),
})
}),
}
}
fn receiver_agent_identity(&self, thread_id: ThreadId) -> (Option<String>, Option<String>) {
let receiver_agent = self
.receiver_agents
.iter()
.find(|agent| agent.thread_id == thread_id);
(
receiver_agent.and_then(|agent| agent.agent_nickname.clone()),
receiver_agent.and_then(|agent| agent.agent_role.clone()),
)
}
fn agent_status(&self, thread_id: ThreadId) -> AgentStatus {
self.agents_states
.get(&thread_id)
.cloned()
.unwrap_or(AgentStatus::NotFound)
}
}
impl SubAgentActivityItem {
pub(crate) fn as_legacy_event(&self, occurred_at_ms: i64) -> EventMsg {
EventMsg::SubAgentActivity(SubAgentActivityEvent {
event_id: self.id.clone(),
occurred_at_ms,
agent_thread_id: self.agent_thread_id,
agent_path: self.agent_path.clone(),
kind: self.kind,
})
}
}
impl WebSearchItem {
pub fn as_legacy_event(&self) -> EventMsg {
EventMsg::WebSearchEnd(WebSearchEndEvent {
@@ -234,6 +513,16 @@ impl HasLegacyEvent for ItemStartedEvent {
}
TurnItem::FileChange(item) => vec![item.as_legacy_begin_event(self.turn_id.clone())],
TurnItem::McpToolCall(item) => vec![item.as_legacy_begin_event()],
TurnItem::CommandExecution(item) => {
vec![item.as_legacy_begin_event(self.turn_id.clone(), self.started_at_ms)]
}
TurnItem::DynamicToolCall(item) => {
vec![item.as_legacy_request_event(self.turn_id.clone(), self.started_at_ms)]
}
TurnItem::CollabAgentToolCall(item) => item
.as_legacy_begin_event(self.started_at_ms)
.into_iter()
.collect(),
_ => Vec::new(),
}
}
@@ -246,6 +535,21 @@ impl HasLegacyEvent for ItemCompletedEvent {
.as_legacy_end_event(self.turn_id.clone())
.into_iter()
.collect(),
TurnItem::CommandExecution(item) => item
.as_legacy_end_event(self.turn_id.clone(), self.completed_at_ms)
.into_iter()
.collect(),
TurnItem::DynamicToolCall(item) => item
.as_legacy_response_event(self.turn_id.clone(), self.completed_at_ms)
.into_iter()
.collect(),
TurnItem::CollabAgentToolCall(item) => item
.as_legacy_end_event(self.completed_at_ms)
.into_iter()
.collect(),
TurnItem::SubAgentActivity(item) => {
vec![item.as_legacy_event(self.completed_at_ms)]
}
_ => self.item.as_legacy_events(show_raw_agent_reasoning),
}
}
codex-rs/protocol/src/protocol.rs
+137
View File
@@ -4309,6 +4309,10 @@ pub struct CollabResumeEndEvent {
#[cfg(test)]
mod tests {
use super::*;
use crate::items::CommandExecutionItem;
use crate::items::CommandExecutionStatus;
use crate::items::DynamicToolCallItem;
use crate::items::DynamicToolCallStatus;
use crate::items::FileChangeItem;
use crate::items::ImageGenerationItem;
use crate::items::McpToolCallItem;
@@ -5290,6 +5294,139 @@ mod tests {
}
}
#[test]
fn command_execution_item_lifecycle_emits_legacy_exec_events() {
let cwd = PathUri::from_abs_path(&test_path_buf("/tmp").abs());
let started = ItemStartedEvent {
thread_id: ThreadId::new(),
turn_id: "turn-1".into(),
started_at_ms: 10,
item: TurnItem::CommandExecution(CommandExecutionItem {
id: "exec-1".into(),
process_id: Some("pid-1".into()),
command: vec!["echo".into(), "done".into()],
cwd: cwd.clone(),
parsed_cmd: vec![ParsedCommand::Unknown {
cmd: "echo done".into(),
}],
source: ExecCommandSource::Agent,
interaction_input: None,
status: CommandExecutionStatus::InProgress,
stdout: None,
stderr: None,
aggregated_output: None,
exit_code: None,
duration: None,
formatted_output: None,
}),
};
let completed = ItemCompletedEvent {
thread_id: ThreadId::new(),
turn_id: "turn-1".into(),
completed_at_ms: 20,
item: TurnItem::CommandExecution(CommandExecutionItem {
id: "exec-1".into(),
process_id: Some("pid-1".into()),
command: vec!["echo".into(), "done".into()],
cwd,
parsed_cmd: vec![ParsedCommand::Unknown {
cmd: "echo done".into(),
}],
source: ExecCommandSource::Agent,
interaction_input: None,
status: CommandExecutionStatus::Completed,
stdout: Some("done\n".into()),
stderr: Some(String::new()),
aggregated_output: Some("done\n".into()),
exit_code: Some(0),
duration: Some(Duration::from_millis(5)),
formatted_output: Some("done\n".into()),
}),
};
assert!(matches!(
started.as_legacy_events(/*show_raw_agent_reasoning*/ false).as_slice(),
[EventMsg::ExecCommandBegin(ExecCommandBeginEvent {
call_id,
turn_id,
started_at_ms: 10,
..
})] if call_id == "exec-1" && turn_id == "turn-1"
));
assert!(matches!(
completed
.as_legacy_events(/*show_raw_agent_reasoning*/ false)
.as_slice(),
[EventMsg::ExecCommandEnd(ExecCommandEndEvent {
call_id,
turn_id,
completed_at_ms: 20,
aggregated_output,
..
})] if call_id == "exec-1" && turn_id == "turn-1" && aggregated_output == "done\n"
));
}
#[test]
fn dynamic_tool_call_item_lifecycle_emits_legacy_dynamic_tool_events() {
let started = ItemStartedEvent {
thread_id: ThreadId::new(),
turn_id: "turn-1".into(),
started_at_ms: 10,
item: TurnItem::DynamicToolCall(DynamicToolCallItem {
id: "dynamic-1".into(),
namespace: Some("apps".into()),
tool: "lookup".into(),
arguments: json!({"id": "123"}),
status: DynamicToolCallStatus::InProgress,
content_items: None,
success: None,
error: None,
duration: None,
}),
};
let completed = ItemCompletedEvent {
thread_id: ThreadId::new(),
turn_id: "turn-1".into(),
completed_at_ms: 20,
item: TurnItem::DynamicToolCall(DynamicToolCallItem {
id: "dynamic-1".into(),
namespace: Some("apps".into()),
tool: "lookup".into(),
arguments: json!({"id": "123"}),
status: DynamicToolCallStatus::Completed,
content_items: Some(vec![DynamicToolCallOutputContentItem::InputText {
text: "ok".into(),
}]),
success: Some(true),
error: None,
duration: Some(Duration::from_millis(5)),
}),
};
assert!(matches!(
started.as_legacy_events(/*show_raw_agent_reasoning*/ false).as_slice(),
[EventMsg::DynamicToolCallRequest(DynamicToolCallRequest {
call_id,
turn_id,
started_at_ms: 10,
..
})] if call_id == "dynamic-1" && turn_id == "turn-1"
));
assert!(matches!(
completed
.as_legacy_events(/*show_raw_agent_reasoning*/ false)
.as_slice(),
[EventMsg::DynamicToolCallResponse(DynamicToolCallResponseEvent {
call_id,
turn_id,
completed_at_ms: 20,
success: true,
..
})] if call_id == "dynamic-1" && turn_id == "turn-1"
));
}
#[test]
fn item_started_event_requires_started_at_ms() {
let mut value = serde_json::to_value(ItemStartedEvent {