Preserve client metadata through queued submissions (#59826)

## Why Queued inputs previously stored and dispatched without the submitting client's Responses API metadata, losing attribution such as `client_type` when the queued turn started. ## What changed - Accept `responsesapiClientMetadata` when adding or updating queued submissions and expose it on returned queue items. - Persist metadata with the input and forward it when starting a selected item or dispatching an idle queue, including after restoration. - Preserve existing metadata when an update omits it or supplies an empty map; replace it when a non-empty map is supplied. - Keep stored inputs readable by older queue workers and update generated protocol artifacts. ## Testing Extend queue service and app-server tests to cover metadata round trips, preservation and replacement on edits, compatibility with older readers, and forwarding during selected, idle, and restored dispatch. GitOrigin-RevId: 508043b1d3d0298c9af2be8e05b6999de46596a1
2026-11-06 20:20:34 +00:00 · 2026-11-06 05:10:44 +00:00
parent 5fab289359
commit a037e65ead
13 changed files with 254 additions and 46 deletions
@@ -19244,6 +19244,16 @@
              "$ref": "#/definitions/v2/UserInput"
            },
            "type": "array"
+          },
+          "responsesapiClientMetadata": {
+            "additionalProperties": {
+              "type": "string"
+            },
+            "description": "Metadata retained with this input, including across reconnect and restoration.",
+            "type": [
+              "object",
+              "null"
+            ]
          }
        },
        "required": [
@@ -15205,6 +15205,16 @@
            "$ref": "#/definitions/UserInput"
          },
          "type": "array"
+        },
+        "responsesapiClientMetadata": {
+          "additionalProperties": {
+            "type": "string"
+          },
+          "description": "Metadata retained with this input, including across reconnect and restoration.",
+          "type": [
+            "object",
+            "null"
+          ]
        }
      },
      "required": [
@@ -3,4 +3,8 @@
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
 import type { UserInput } from "./UserInput";

-export type QueuedSubmission = { id: string, input: Array<UserInput>, clientUserMessageId: string, };
+export type QueuedSubmission = { id: string, input: Array<UserInput>, clientUserMessageId: string,
+/**
+ * Metadata retained with this input, including across reconnect and restoration.
+ */
+responsesapiClientMetadata: { [key in string]?: string } | null, };
@@ -939,6 +939,8 @@ pub struct QueuedSubmission {
    pub id: String,
    pub input: Vec<UserInput>,
    pub client_user_message_id: String,
+    /// Metadata retained with this input, including across reconnect and restoration.
+    pub responsesapi_client_metadata: Option<HashMap<String, String>>,
 }

 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
@@ -948,6 +950,9 @@ pub struct ThreadQueueAddParams {
    pub thread_id: String,
    pub input: Vec<UserInput>,
    pub client_user_message_id: String,
+    /// ResponsesAPI metadata captured from the client submitting this input.
+    #[ts(optional = nullable)]
+    pub responsesapi_client_metadata: Option<HashMap<String, String>>,
 }

 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
@@ -986,6 +991,9 @@ pub struct ThreadQueueUpdateParams {
    pub thread_id: String,
    pub queued_submission_id: String,
    pub input: Vec<UserInput>,
+    /// Non-empty metadata replaces the input's metadata; omission or an empty map preserves it.
+    #[ts(optional = nullable)]
+    pub responsesapi_client_metadata: Option<HashMap<String, String>>,
 }

 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
@@ -83,9 +83,10 @@ impl ThreadQueueRequestProcessor {
        ensure_direct_input_allowed(loaded_thread.as_deref(), &source)?;
        let queued_item = self
            .service()?
-            .enqueue(
+            .enqueue_with_metadata(
                thread_id,
                submission_into_turn_input(params.input, Some(params.client_user_message_id)),
+                params.responsesapi_client_metadata,
            )
            .await
            .map_err(queue_error)?;
@@ -140,10 +141,11 @@ impl ThreadQueueRequestProcessor {
        ensure_direct_input_allowed(loaded_thread.as_deref(), &source)?;
        let queued_item = self
            .service()?
-            .update(
+            .update_with_metadata(
                thread_id,
                params.queued_submission_id.clone(),
                submission_into_turn_input(params.input, /*client_user_message_id*/ None),
+                params.responsesapi_client_metadata,
            )
            .await
            .map_err(queue_error)?
@@ -345,6 +347,7 @@ fn api_queued_submission(value: QueuedItem) -> Result<QueuedSubmission, JSONRPCE
    };
    Ok(QueuedSubmission {
        id: value.id,
+        responsesapi_client_metadata: value.responsesapi_client_metadata,
        input: content.into_iter().map(Into::into).collect(),
        client_user_message_id: client_id
            .ok_or_else(|| internal_error("queued submission is missing its client message id"))?,
@@ -105,6 +105,7 @@ base_url = "{}/v1"
        .request(|request_id| ClientRequest::ThreadQueueAdd {
            request_id,
            params: ThreadQueueAddParams {
+                responsesapi_client_metadata: None,
                thread_id: started.thread.id.clone(),
                input: input.clone(),
                client_user_message_id: "queued-message".into(),
@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::time::Duration;

 use anyhow::Context;
@@ -136,10 +137,18 @@ async fn queue_crud_preserves_identity_order_and_notifications() -> Result<()> {
        &mut app,
        ThreadQueueAddParams {
            client_user_message_id: "first-client-message".to_string(),
+            responsesapi_client_metadata: Some(HashMap::from([(
+                "client_type".to_string(),
+                "macos_app".to_string(),
+            )])),
            ..submission(&thread_id, "first")
        },
    )
    .await?;
+    assert_eq!(
+        first.responsesapi_client_metadata.as_ref().unwrap()["client_type"],
+        "macos_app"
+    );
    let second = queue_item(&mut app, submission(&thread_id, "second")).await?;
    let first_change: ThreadQueueChangedNotification =
        timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??;
@@ -162,6 +171,10 @@ async fn queue_crud_preserves_identity_order_and_notifications() -> Result<()> {
        .request(|request_id| ClientRequest::ThreadQueueUpdate {
            request_id,
            params: ThreadQueueUpdateParams {
+                responsesapi_client_metadata: Some(HashMap::from([(
+                    "client_type".to_string(),
+                    "ios_app".to_string(),
+                )])),
                thread_id: thread_id.clone(),
                queued_submission_id: first.id.clone(),
                input: vec![text("first edited")],
@@ -174,6 +187,14 @@ async fn queue_crud_preserves_identity_order_and_notifications() -> Result<()> {
        first.client_user_message_id
    );
    assert_eq!(updated.queued_submission.input, vec![text("first edited")]);
+    assert_eq!(
+        updated
+            .queued_submission
+            .responsesapi_client_metadata
+            .as_ref()
+            .unwrap()["client_type"],
+        "ios_app"
+    );
    let update_change: ThreadQueueChangedNotification =
        timeout(READ_TIMEOUT, app.read_notification("thread/queue/changed")).await??;
    assert_eq!(
@@ -271,6 +292,7 @@ async fn queue_list_returns_ordered_pages_and_lightweight_notifications() -> Res
    let queued = queue_item(
        &mut app,
        ThreadQueueAddParams {
+            responsesapi_client_metadata: None,
            thread_id: thread_id.clone(),
            input: vec![text(&"x".repeat(64 * 1024))],
            client_user_message_id: "oversized-snapshot".to_string(),
@@ -375,6 +397,10 @@ async fn idle_queue_dispatch_preserves_client_id() -> Result<()> {
        .thread
        .id;
    let queued_submission = ThreadQueueAddParams {
+        responsesapi_client_metadata: Some(std::collections::HashMap::from([(
+            "client_type".to_string(),
+            "ios_app".to_string(),
+        )])),
        thread_id: thread_id.clone(),
        input: vec![text("durable queued message")],
        client_user_message_id: "stable-queued-client-id".to_string(),
@@ -417,6 +443,7 @@ async fn idle_queue_dispatch_preserves_client_id() -> Result<()> {
        .to_str()
        .context("queued turn metadata header is not valid ASCII")?;
    let metadata: Value = serde_json::from_str(metadata_header)?;
+    assert_eq!(metadata["client_type"], "ios_app");
    assert_eq!(metadata["thread_id"].as_str(), Some(thread_id.as_str()));
    assert_eq!(
        metadata["turn_id"].as_str(),
@@ -1056,6 +1083,7 @@ async fn decline_approval(app: &mut TestAppServer, request_id: RequestId) -> Res

 fn submission(thread_id: &str, value: &str) -> ThreadQueueAddParams {
    ThreadQueueAddParams {
+        responsesapi_client_metadata: None,
        thread_id: thread_id.to_string(),
        input: vec![text(value)],
        client_user_message_id: format!("queued-{value}"),
@@ -42,6 +42,7 @@ use uuid::Uuid;
 pub struct QueuedItem {
    pub id: String,
    pub input: TurnInput,
+    pub responsesapi_client_metadata: Option<HashMap<String, String>>,
 }

 #[derive(Debug, Error)]
@@ -266,12 +267,22 @@ impl QueuedItemService {
        &self,
        thread_id: ThreadId,
        input: TurnInput,
+    ) -> Result<QueuedItem, QueueServiceError> {
+        self.enqueue_with_metadata(thread_id, input, /*responsesapi_client_metadata*/ None)
+            .await
+    }
+
+    pub async fn enqueue_with_metadata(
+        &self,
+        thread_id: ThreadId,
+        input: TurnInput,
+        responsesapi_client_metadata: Option<HashMap<String, String>>,
    ) -> Result<QueuedItem, QueueServiceError> {
        let input = prepare_queued_user_input(input).await?;
-        let payload = serde_json::to_string(&input)?;
+        let payload = queued_input_payload(&input, responsesapi_client_metadata.as_ref())?;
        let item = {
            let _dispatch_guard = self.dispatch_guard(thread_id).await;
-            let item = queued_item_from_record(self.queue.enqueue(thread_id, payload).await?)?;
+            let item = QueuedItem::try_from(self.queue.enqueue(thread_id, payload).await?)?;
            self.emit_changed(thread_id);
            item
        };
@@ -294,7 +305,7 @@ impl QueuedItemService {
            .list_page(thread_id, offset, limit)
            .await?
            .into_iter()
-            .map(queued_item_from_record)
+            .map(QueuedItem::try_from)
            .collect()
    }

@@ -302,29 +313,50 @@ impl QueuedItemService {
        &self,
        thread_id: ThreadId,
        queued_item_id: String,
-        mut input: TurnInput,
+        input: TurnInput,
    ) -> Result<Option<QueuedItem>, QueueServiceError> {
+        self.update_with_metadata(
+            thread_id,
+            queued_item_id,
+            input,
+            /*responsesapi_client_metadata*/ None,
+        )
+        .await
+    }
+
+    /// Omitted or empty metadata preserves the input's attribution, not the current controller's.
+    pub async fn update_with_metadata(
+        &self,
+        thread_id: ThreadId,
+        queued_item_id: String,
+        mut input: TurnInput,
+        mut responsesapi_client_metadata: Option<HashMap<String, String>>,
+    ) -> Result<Option<QueuedItem>, QueueServiceError> {
+        responsesapi_client_metadata =
+            responsesapi_client_metadata.filter(|metadata| !metadata.is_empty());
        let _dispatch_guard = self.dispatch_guard(thread_id).await;
+        let previous = self
+            .list(thread_id)
+            .await?
+            .into_iter()
+            .find(|item| item.id == queued_item_id);
        if let TurnInput::UserInput { client_id, .. } = &mut input {
-            *client_id = self
-                .list(thread_id)
-                .await?
-                .into_iter()
-                .find_map(|item| match item {
-                    QueuedItem {
-                        id,
-                        input: TurnInput::UserInput { client_id, .. },
-                    } if id == queued_item_id => client_id,
-                    _ => None,
-                });
+            *client_id = previous.as_ref().and_then(|item| match &item.input {
+                TurnInput::UserInput { client_id, .. } => client_id.clone(),
+                _ => None,
+            });
+        }
+        if responsesapi_client_metadata.is_none() {
+            responsesapi_client_metadata =
+                previous.and_then(|item| item.responsesapi_client_metadata);
        }
        let input = prepare_queued_user_input(input).await?;
-        let payload = serde_json::to_string(&input)?;
+        let payload = queued_input_payload(&input, responsesapi_client_metadata.as_ref())?;
        let item = self
            .queue
            .update(thread_id, queued_item_id, payload)
            .await?
-            .map(queued_item_from_record)
+            .map(QueuedItem::try_from)
            .transpose()?;
        if item.is_some() {
            self.emit_changed(thread_id);
@@ -389,12 +421,15 @@ impl QueuedItemService {
            return Err(QueueServiceError::InvalidInput);
        };
        let submission = thread
-            .start_turn_if_idle(TurnInputRequest::new(input).with_trace(trace).on_start(
-                TurnStartOptions {
-                    turn_trigger: Some("queue".to_string()),
-                    ..Default::default()
-                },
-            ))
+            .start_turn_if_idle(
+                TurnInputRequest::new(input)
+                    .with_responses_metadata(item.responsesapi_client_metadata)
+                    .with_trace(trace)
+                    .on_start(TurnStartOptions {
+                        turn_trigger: Some("queue".to_string()),
+                        ..Default::default()
+                    }),
+            )
            .await?;
        if matches!(submission, StartIfIdleSubmission::Started { .. }) {
            self.delete_locked(thread_id, queued_item_id).await?;
@@ -422,25 +457,29 @@ impl QueuedItemService {
            };
            let queued_item_id = record.id.clone();

-            let input = match serde_json::from_str::<TurnInput>(&record.payload) {
-                Ok(input) => input,
+            let item = match QueuedItem::try_from(record) {
+                Ok(item) => item,
                Err(error) => {
                    tracing::warn!(%queued_item_id, %error, "discarding invalid queued item");
                    self.delete_locked(thread_id, queued_item_id).await?;
                    continue;
                }
            };
-            if !matches!(input, TurnInput::UserInput { .. }) {
+            if !matches!(item.input, TurnInput::UserInput { .. }) {
                tracing::warn!(%queued_item_id, "discarding non-user queued input");
                self.delete_locked(thread_id, queued_item_id).await?;
                continue;
            }

            match thread
-                .start_turn_if_idle(TurnInputRequest::new(input).on_start(TurnStartOptions {
-                    turn_trigger: Some("queue".to_string()),
-                    ..Default::default()
-                }))
+                .start_turn_if_idle(
+                    TurnInputRequest::new(item.input)
+                        .with_responses_metadata(item.responsesapi_client_metadata)
+                        .on_start(TurnStartOptions {
+                            turn_trigger: Some("queue".to_string()),
+                            ..Default::default()
+                        }),
+                )
                .await
            {
                Ok(StartIfIdleSubmission::Started { .. }) => {
@@ -566,11 +605,34 @@ where
    }
 }

-fn queued_item_from_record(
-    record: QueuedUserSubmissionRecord,
-) -> Result<QueuedItem, QueueServiceError> {
-    Ok(QueuedItem {
-        id: record.id,
-        input: serde_json::from_str::<TurnInput>(&record.payload)?,
-    })
+fn queued_input_payload(
+    input: &TurnInput,
+    metadata: Option<&HashMap<String, String>>,
+) -> Result<String, QueueServiceError> {
+    let Some(metadata) = metadata else {
+        return Ok(serde_json::to_string(input)?);
+    };
+    let mut payload = serde_json::to_value(input)?;
+    // Keep TurnInput's wire shape so older readers can still execute the input.
+    payload["UserInput"]["responsesapi_client_metadata"] = serde_json::to_value(metadata)?;
+    Ok(serde_json::to_string(&payload)?)
+}
+
+impl TryFrom<QueuedUserSubmissionRecord> for QueuedItem {
+    type Error = QueueServiceError;
+
+    fn try_from(record: QueuedUserSubmissionRecord) -> Result<Self, Self::Error> {
+        let mut payload: serde_json::Value = serde_json::from_str(&record.payload)?;
+        let responsesapi_client_metadata = payload
+            .get_mut("UserInput")
+            .and_then(serde_json::Value::as_object_mut)
+            .and_then(|input| input.remove("responsesapi_client_metadata"))
+            .map(serde_json::from_value)
+            .transpose()?;
+        Ok(QueuedItem {
+            id: record.id,
+            input: serde_json::from_value(payload)?,
+            responsesapi_client_metadata,
+        })
+    }
 }
@@ -1,4 +1,5 @@
 use codex_protocol::models::AudioReference;
+use std::collections::HashMap;
 use std::path::Path;
 use std::sync::Arc;
 use std::sync::Mutex;
@@ -316,11 +317,28 @@ async fn queued_input_and_unique_event_ids_round_trip() -> anyhow::Result<()> {
    let (queue, _home) = test_queue().await?;
    let sink = Arc::new(RecordingEventSink::default());
    let event_sink: Arc<dyn ExtensionEventSink> = sink.clone();
-    let service = QueuedItemService::new(queue, Weak::new(), event_sink);
+    let service = QueuedItemService::new(Arc::clone(&queue), Weak::new(), event_sink);
    let thread_id = ThreadId::new();
    let input = structured_user_input("structured message");

-    let first = service.enqueue(thread_id, input.clone()).await?;
+    let first = service
+        .enqueue_with_metadata(
+            thread_id,
+            input.clone(),
+            Some(HashMap::from([(
+                "client_type".to_string(),
+                "ios_app".to_string(),
+            )])),
+        )
+        .await?;
+    // Old workers still read TurnInput directly from the stored payload.
+    let records = queue
+        .list_page(thread_id, /*offset*/ 0, /*limit*/ 1)
+        .await?;
+    assert_eq!(
+        input,
+        serde_json::from_str::<TurnInput>(&records[0].payload)?
+    );
    let second = service.enqueue(thread_id, user_input("next")).await?;
    let TurnInput::UserInput {
        client_id: Some(generated_client_id),
@@ -362,7 +380,14 @@ async fn editing_reordering_and_deleting_preserve_queue_identity() -> anyhow::Re
    let service = QueuedItemService::new(queue, Weak::new(), Arc::new(NoopExtensionEventSink));
    let thread_id = ThreadId::new();
    let first = service
-        .enqueue(thread_id, structured_user_input("first"))
+        .enqueue_with_metadata(
+            thread_id,
+            structured_user_input("first"),
+            Some(HashMap::from([(
+                "client_type".to_string(),
+                "desktop_app".to_string(),
+            )])),
+        )
        .await?;
    let second = service.enqueue(thread_id, user_input("second")).await?;
    let edited = service
@@ -374,6 +399,21 @@ async fn editing_reordering_and_deleting_preserve_queue_identity() -> anyhow::Re
        anyhow::bail!("edited queue item does not contain user input");
    };
    assert_eq!(Some("stable-client-message"), client_id.as_deref());
+    assert_eq!(
+        first.responsesapi_client_metadata,
+        edited.responsesapi_client_metadata
+    );
+    let mobile_metadata = HashMap::from([("client_type".to_string(), "ios_app".to_string())]);
+    let edited = service
+        .update_with_metadata(
+            thread_id,
+            first.id.clone(),
+            user_input("edited on phone"),
+            Some(mobile_metadata.clone()),
+        )
+        .await?
+        .context("queued item missing")?;
+    assert_eq!(Some(mobile_metadata), edited.responsesapi_client_metadata);
    assert!(matches!(
        service.reorder(thread_id, vec![first.id.clone()]).await,
        Err(QueueServiceError::Storage(
@@ -403,7 +443,14 @@ async fn starting_a_selected_item_preserves_the_remaining_queue() -> anyhow::Res
    let service = QueuedItemService::new(queue, Weak::new(), Arc::new(NoopExtensionEventSink));
    let first = service.enqueue(thread_id, user_input("first")).await?;
    let second = service
-        .enqueue(thread_id, structured_user_input("second"))
+        .enqueue_with_metadata(
+            thread_id,
+            structured_user_input("second"),
+            Some(HashMap::from([(
+                "client_type".to_string(),
+                "ios_app".to_string(),
+            )])),
+        )
        .await?;

    let submission = service
@@ -432,6 +479,13 @@ async fn starting_a_selected_item_preserves_the_remaining_queue() -> anyhow::Res
            .last()
            .map(String::as_str)
    );
+    let body = response.single_request().body_json();
+    let metadata: serde_json::Value = serde_json::from_str(
+        body["client_metadata"]["x-codex-turn-metadata"]
+            .as_str()
+            .context("missing turn metadata")?,
+    )?;
+    assert_eq!(metadata["client_type"], "ios_app");
    Ok(())
 }

@@ -1107,7 +1161,8 @@ async fn invalid_head_is_skipped_and_a_live_user_turn_is_accepted() -> anyhow::R
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn resumed_idle_dispatches_input_without_a_loaded_manager() -> anyhow::Result<()> {
    let server = start_mock_server().await;
-    responses::mount_sse_once(&server, responses::sse_completed("resumed-turn")).await;
+    let response =
+        responses::mount_sse_once(&server, responses::sse_completed("resumed-turn")).await;
    let test = test_codex().build_with_auto_env(&server).await?;
    let thread_id = test.session_configured.thread_id;
    let queue = loaded_thread_queue(&test)?;
@@ -1116,7 +1171,14 @@ async fn resumed_idle_dispatches_input_without_a_loaded_manager() -> anyhow::Res
        Weak::new(),
        Arc::new(NoopExtensionEventSink),
    )
-    .enqueue(thread_id, user_input("queued while unloaded"))
+    .enqueue_with_metadata(
+        thread_id,
+        user_input("queued while unloaded"),
+        Some(HashMap::from([(
+            "client_type".to_string(),
+            "ios_app".to_string(),
+        )])),
+    )
    .await?;
    let service = QueuedItemService::new(
        queue,
@@ -1125,5 +1187,17 @@ async fn resumed_idle_dispatches_input_without_a_loaded_manager() -> anyhow::Res
    );
    emit_idle(&service, thread_id).await;
    assert!(service.list(thread_id).await?.is_empty());
+    wait_for_event_match(test.codex.as_ref(), |event| match event {
+        EventMsg::TurnComplete(_) => Some(()),
+        _ => None,
+    })
+    .await;
+    let body = response.single_request().body_json();
+    let metadata: serde_json::Value = serde_json::from_str(
+        body["client_metadata"]["x-codex-turn-metadata"]
+            .as_str()
+            .context("missing turn metadata")?,
+    )?;
+    assert_eq!(metadata["client_type"], "ios_app");
    Ok(())
 }
@@ -127,6 +127,7 @@ pub(super) async fn run_session_queue_action_with_app_server(
        .request_typed(ClientRequest::ThreadQueueAdd {
            request_id,
            params: ThreadQueueAddParams {
+                responsesapi_client_metadata: None,
                thread_id: thread_id.to_string(),
                input: vec![UserInput::Text {
                    text: message.to_string(),
@@ -9677,6 +9677,13 @@ class QueuedSubmission(BaseModel):
    client_user_message_id: Annotated[str, Field(alias="clientUserMessageId")]
    id: str
    input: list[UserInput]
+    responsesapi_client_metadata: Annotated[
+        dict[str, Any] | None,
+        Field(
+            alias="responsesapiClientMetadata",
+            description="Metadata retained with this input, including across reconnect and restoration.",
+        ),
+    ] = None


 class RateLimitResetCredit(BaseModel):