codex: tighten turn metadata staging

Co-authored-by: Codex <noreply@openai.com>

codex-rs/app-server/src/codex_message_processor.rs

@@ -5974,13 +5974,8 @@ impl CodexMessageProcessor {
             .into_iter()
             .map(V2UserInput::into_core)
             .collect();
-
-        if let Some(metadata) = params.metadata.clone()
-            && let Err(error) = Self::set_next_turn_metadata(thread.as_ref(), metadata).await
-        {
-            self.outgoing.send_error(request_id, error).await;
-            return;
-        }
+        let next_turn_metadata = params.metadata.clone();
+        let clear_next_turn_metadata_on_error = next_turn_metadata.is_some();
 
         let has_any_overrides = params.cwd.is_some()
             || params.approval_policy.is_some()
@@ -6018,6 +6013,13 @@ impl CodexMessageProcessor {
                 .await;
         }
 
+        if let Some(metadata) = next_turn_metadata
+            && let Err(error) = Self::set_next_turn_metadata(thread.as_ref(), metadata).await
+        {
+            self.outgoing.send_error(request_id, error).await;
+            return;
+        }
+
         // Start the turn by submitting the user input. Return its submission id as turn_id.
         let turn_id = self
             .submit_core_op(
@@ -6043,6 +6045,15 @@ impl CodexMessageProcessor {
                 self.outgoing.send_response(request_id, response).await;
             }
             Err(err) => {
+                if clear_next_turn_metadata_on_error
+                    && let Err(clear_error) =
+                        Self::set_next_turn_metadata(thread.as_ref(), BTreeMap::new()).await
+                {
+                    warn!(
+                        error = %clear_error.message,
+                        "failed to clear next turn metadata after turn/start submission error"
+                    );
+                }
                 let error = JSONRPCErrorError {
                     code: INTERNAL_ERROR_CODE,
                     message: format!("failed to start turn: {err}"),

codex-rs/core/src/codex_tests.rs

@@ -2278,6 +2278,46 @@ async fn session_configuration_apply_rederives_legacy_file_system_policy_on_cwd_
     );
 }
 
+#[tokio::test]
+async fn new_turn_with_sub_id_consumes_next_turn_metadata_once() {
+    let (session, _) = make_session_and_context().await;
+    let next_turn_metadata = std::collections::BTreeMap::from([(
+        "parentTurnId".to_string(),
+        "parent-turn-123".to_string(),
+    )]);
+
+    session
+        .update_settings(SessionSettingsUpdate {
+            next_turn_metadata: Some(next_turn_metadata.clone()),
+            ..Default::default()
+        })
+        .await
+        .expect("test setup should allow updating next turn metadata");
+
+    let first_turn_context = session
+        .new_turn_with_sub_id("turn-1".to_string(), SessionSettingsUpdate::default())
+        .await
+        .expect("first turn should be created");
+    assert_eq!(
+        first_turn_context.turn_metadata_state.metadata(),
+        &next_turn_metadata
+    );
+    let state = session.state.lock().await;
+    assert!(state.session_configuration.next_turn_metadata.is_empty());
+    drop(state);
+
+    let second_turn_context = session
+        .new_turn_with_sub_id("turn-2".to_string(), SessionSettingsUpdate::default())
+        .await
+        .expect("second turn should be created");
+    assert!(
+        second_turn_context
+            .turn_metadata_state
+            .metadata()
+            .is_empty()
+    );
+}
+
 #[tokio::test]
 async fn session_new_fails_when_zsh_fork_enabled_without_zsh_path() {
     let codex_home = tempfile::tempdir().expect("create temp dir");