From ab3b4d26d4739ad824e31d0b5d96284d84630947 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Mon, 10 Aug 2026 23:34:58 +0000 Subject: [PATCH] Make submission operations move-only (#37901) ## What changed - Remove `Clone` from `Submission` and remove `Clone` and `PartialEq` from `Op`. - Consume operations directly in the submission loop instead of cloning them. - Limit thread-manager test capture to the operation variants under test and update assertions to compare their relevant fields. GitOrigin-RevId: 781cc631391b8583d94dad5ca45f589856c0b21a --- codex-rs/core/src/agent/control_tests.rs | 63 +++++++++++++++---- codex-rs/core/src/codex_delegate_tests.rs | 61 ++++++++++-------- codex-rs/core/src/session/handlers.rs | 6 +- codex-rs/core/src/session/tests.rs | 6 +- codex-rs/core/src/thread_manager.rs | 39 +++++++++++- .../src/tools/handlers/multi_agents_tests.rs | 46 +++++++------- codex-rs/protocol/src/protocol.rs | 4 +- 7 files changed, 155 insertions(+), 70 deletions(-) diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index c03e3a63cb..0cbd809f0d 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -92,6 +92,45 @@ fn text_input(text: &str) -> Vec { }] } +fn captured_op_matches(actual: &(ThreadId, Op), expected: &(ThreadId, Op)) -> bool { + if actual.0 != expected.0 { + return false; + } + match (&actual.1, &expected.1) { + ( + Op::UserInput { + items: actual_items, + final_output_json_schema: actual_schema, + responsesapi_client_metadata: actual_metadata, + additional_context: actual_context, + thread_settings: actual_settings, + }, + Op::UserInput { + items: expected_items, + final_output_json_schema: expected_schema, + responsesapi_client_metadata: expected_metadata, + additional_context: expected_context, + thread_settings: expected_settings, + }, + ) => { + actual_items == expected_items + && actual_schema == expected_schema + && actual_metadata == expected_metadata + && actual_context == expected_context + && actual_settings == expected_settings + } + ( + Op::InterAgentCommunication { + communication: actual, + }, + Op::InterAgentCommunication { + communication: expected, + }, + ) => actual == expected, + _ => false, + } +} + fn assistant_message(text: &str, phase: Option) -> ResponseItem { ResponseItem::Message { id: None, @@ -580,8 +619,8 @@ async fn send_input_submits_user_message() { .manager .captured_ops() .into_iter() - .find(|entry| *entry == expected); - assert_eq!(captured, Some(expected)); + .find(|entry| captured_op_matches(entry, &expected)); + assert!(captured.is_some()); } #[tokio::test] @@ -618,8 +657,8 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig .manager .captured_ops() .into_iter() - .find(|entry| *entry == expected); - assert_eq!(captured, Some(expected)); + .find(|entry| captured_op_matches(entry, &expected)); + assert!(captured.is_some()); timeout(Duration::from_secs(5), async { loop { @@ -779,8 +818,8 @@ async fn ensure_v2_agent_loaded_reloads_registered_unloaded_agent() { .manager .captured_ops() .into_iter() - .find(|entry| *entry == expected); - assert_eq!(captured, Some(expected)); + .find(|entry| captured_op_matches(entry, &expected)); + assert!(captured.is_some()); } #[tokio::test] @@ -938,8 +977,8 @@ async fn spawn_agent_creates_thread_and_sends_prompt() { .manager .captured_ops() .into_iter() - .find(|entry| *entry == expected); - assert_eq!(captured, Some(expected)); + .find(|entry| captured_op_matches(entry, &expected)); + assert!(captured.is_some()); } #[tokio::test] @@ -1539,8 +1578,8 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() { .manager .captured_ops() .into_iter() - .find(|entry| *entry == expected); - assert_eq!(captured, Some(expected)); + .find(|entry| captured_op_matches(entry, &expected)); + assert!(captured.is_some()); let _ = harness .control @@ -2902,8 +2941,8 @@ async fn multi_agent_v2_completion_queues_message_for_direct_parent() { .manager .captured_ops() .into_iter() - .find(|entry| *entry == expected); - if captured == Some(expected.clone()) { + .find(|entry| captured_op_matches(entry, &expected)); + if captured.is_some() { break; } sleep(Duration::from_millis(10)).await; diff --git a/codex-rs/core/src/codex_delegate_tests.rs b/codex-rs/core/src/codex_delegate_tests.rs index 2ee551d84e..8a817934e8 100644 --- a/codex-rs/core/src/codex_delegate_tests.rs +++ b/codex-rs/core/src/codex_delegate_tests.rs @@ -177,17 +177,25 @@ async fn forward_ops_preserves_submission_trace_context() { }), parent_turn_id: Some("parent-turn".to_string()), }; - tx_ops.send(submission.clone()).await.unwrap(); + tx_ops.send(submission).await.unwrap(); drop(tx_ops); let forwarded = timeout(Duration::from_secs(1), rx_sub.recv()) .await .expect("forward_ops hung") .expect("forwarded submission missing"); - assert_eq!(submission.id, forwarded.id); - assert_eq!(submission.op, forwarded.op); - assert_eq!(submission.trace, forwarded.trace); - assert_eq!(submission.parent_turn_id, forwarded.parent_turn_id); + assert_eq!("sub-1", forwarded.id); + assert!(matches!(forwarded.op, Op::Interrupt)); + assert_eq!( + forwarded.trace, + Some(codex_protocol::protocol::W3cTraceContext { + traceparent: Some( + "00-1234567890abcdef1234567890abcdef-1234567890abcdef-01".to_string(), + ), + tracestate: Some("vendor=state".to_string()), + }) + ); + assert_eq!(Some("parent-turn".to_string()), forwarded.parent_turn_id); timeout(Duration::from_secs(1), forward) .await @@ -321,13 +329,11 @@ async fn handle_request_permissions_uses_tool_call_id_for_round_trip() { .await .expect("request_permissions response timed out") .expect("request_permissions response missing"); - assert_eq!( - submission.op, - Op::RequestPermissionsResponse { - id: call_id, - response: expected_response, - } - ); + let Op::RequestPermissionsResponse { id, response } = submission.op else { + panic!("expected request permissions response"); + }; + assert_eq!(id, call_id); + assert_eq!(response, expected_response); } #[tokio::test] @@ -415,13 +421,11 @@ async fn handle_request_user_input_preserves_non_blocking_flag_for_round_trip() .await .expect("request_user_input response timed out") .expect("request_user_input response missing"); - assert_eq!( - submission.op, - Op::UserInputAnswer { - id: child_event_id, - response: expected_response, - } - ); + let Op::UserInputAnswer { id, response } = submission.op else { + panic!("expected user input answer"); + }; + assert_eq!(id, child_event_id); + assert_eq!(response, expected_response); } #[tokio::test] @@ -538,14 +542,17 @@ async fn handle_exec_approval_uses_call_id_for_guardian_review_and_approval_id_f .await .expect("exec approval response timed out") .expect("exec approval response missing"); - assert_eq!( - submission.op, - Op::ExecApproval { - id: "callback-approval-1".to_string(), - turn_id: Some("child-turn-1".to_string()), - decision: ReviewDecision::Abort, - } - ); + let Op::ExecApproval { + id, + turn_id, + decision, + } = submission.op + else { + panic!("expected exec approval"); + }; + assert_eq!(id, "callback-approval-1"); + assert_eq!(turn_id, Some("child-turn-1".to_string())); + assert_eq!(decision, ReviewDecision::Abort); } #[tokio::test] diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 3308d2dd95..4f6b0f9c22 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -714,7 +714,7 @@ pub(super) async fn submission_loop( debug!(?sub, "Submission"); let dispatch_span = submission_dispatch_span(&sub); let should_exit = async { - match sub.op.clone() { + match sub.op { Op::Interrupt => { interrupt(&sess).await; false @@ -758,11 +758,11 @@ pub(super) async fn submission_loop( realtime_conversation_list_voices(&sess, sub.id.clone()).await; false } - Op::UserInput { .. } => { + op @ Op::UserInput { .. } => { user_input_or_turn( &sess, sub.id.clone(), - sub.op, + op, sub.client_user_message_id, sub.parent_turn_id, ) diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 9aade4d1b2..fc305616f1 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -7675,7 +7675,7 @@ async fn shutdown_and_wait_allows_multiple_waiters() { let (_tx_event, rx_event) = async_channel::unbounded(); let session_loop_handle = tokio::spawn(async move { let shutdown: Submission = rx_sub.recv().await.expect("shutdown submission"); - assert_eq!(shutdown.op, Op::Shutdown); + assert!(matches!(shutdown.op, Op::Shutdown)); tokio::time::sleep(StdDuration::from_millis(50)).await; }); let io = Arc::new(SessionIo { @@ -7766,7 +7766,7 @@ async fn shutdown_and_wait_shuts_down_cached_guardian_subagent() { .recv() .await .expect("child shutdown submission"); - assert_eq!(shutdown.op, Op::Shutdown); + assert!(matches!(shutdown.op, Op::Shutdown)); child_shutdown_tx .send(()) .expect("child shutdown signal should be delivered"); @@ -7851,7 +7851,7 @@ async fn shutdown_and_wait_shuts_down_tracked_ephemeral_guardian_review() { .recv() .await .expect("child shutdown submission"); - assert_eq!(shutdown.op, Op::Shutdown); + assert!(matches!(shutdown.op, Op::Shutdown)); child_shutdown_tx .send(()) .expect("child shutdown signal should be delivered"); diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 5a53c6fe39..197256133d 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -110,6 +110,32 @@ type CapturedOps = Vec<(ThreadId, Op)>; type SharedCapturedOps = Arc>; pub(crate) type ThreadIdGenerator = Arc ThreadId + Send + Sync>; +// `Op` is intentionally not `Clone`. Thread-manager tests only snapshot the +// small subset of ops they inspect. +fn capture_test_op(op: &Op) -> Option { + match op { + Op::Interrupt => Some(Op::Interrupt), + Op::UserInput { + items, + final_output_json_schema, + responsesapi_client_metadata, + additional_context, + thread_settings, + } => Some(Op::UserInput { + items: items.clone(), + final_output_json_schema: final_output_json_schema.clone(), + responsesapi_client_metadata: responsesapi_client_metadata.clone(), + additional_context: additional_context.clone(), + thread_settings: thread_settings.clone(), + }), + Op::InterAgentCommunication { communication } => Some(Op::InterAgentCommunication { + communication: communication.clone(), + }), + Op::Shutdown => Some(Op::Shutdown), + _ => None, + } +} + pub(crate) fn default_thread_id_generator() -> ThreadIdGenerator { Arc::new(ThreadId::new) } @@ -1240,7 +1266,15 @@ impl ThreadManager { self.state .ops_log .as_ref() - .and_then(|ops_log| ops_log.lock().ok().map(|log| log.clone())) + .and_then(|ops_log| { + ops_log.lock().ok().map(|log| { + log.iter() + .filter_map(|(thread_id, op)| { + capture_test_op(op).map(|op| (*thread_id, op)) + }) + .collect() + }) + }) .unwrap_or_default() } } @@ -1344,8 +1378,9 @@ impl ThreadManagerState { let thread = self.get_thread(thread_id).await?; if let Some(ops_log) = &self.ops_log && let Ok(mut log) = ops_log.lock() + && let Some(captured_op) = capture_test_op(&op) { - log.push((thread_id, op.clone())); + log.push((thread_id, captured_op)); } thread .io diff --git a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs index 7a42cd4f3a..f8ba561dfc 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -2621,27 +2621,31 @@ async fn send_input_accepts_structured_items() { .await .expect("send_input should succeed"); - let expected = Op::UserInput { - items: vec![ - UserInput::Mention { - name: "drive".to_string(), - path: "app://google_drive".to_string(), - }, - UserInput::Text { - text: "read the folder".to_string(), - text_elements: Vec::new(), - }, - ], - final_output_json_schema: None, - responsesapi_client_metadata: None, - additional_context: Default::default(), - thread_settings: Default::default(), - }; - let captured = manager - .captured_ops() - .into_iter() - .find(|(id, op)| *id == agent_id && *op == expected); - assert_eq!(captured, Some((agent_id, expected))); + let expected_items = vec![ + UserInput::Mention { + name: "drive".to_string(), + path: "app://google_drive".to_string(), + }, + UserInput::Text { + text: "read the folder".to_string(), + text_elements: Vec::new(), + }, + ]; + assert!(manager.captured_ops().iter().any(|(id, op)| { + *id == agent_id + && matches!( + op, + Op::UserInput { + items, + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context, + thread_settings, + } if items == &expected_items + && additional_context.is_empty() + && thread_settings == &Default::default() + ) + })); let _ = thread .thread diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 4398896b3f..a16738aa61 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -171,7 +171,7 @@ impl GitSha { } /// Submission Queue Entry - requests from user -#[derive(Debug, Clone)] +#[derive(Debug)] pub struct Submission { /// Unique id for this Submission to correlate with Events pub id: String, @@ -524,7 +524,7 @@ pub struct AdditionalContextEntry { } /// Submission operation -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug)] #[allow(clippy::large_enum_variant)] #[non_exhaustive] pub enum Op {