diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 3db9081733..3e66da15d4 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -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": [ diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index 8eb9806c3b..f31ed222ef 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -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": [ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index a078efead8..aec7f7e97a 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 5f037c463e..cb9385ed2f 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/QueuedSubmission.ts b/codex-rs/app-server-protocol/schema/typescript/v2/QueuedSubmission.ts index f364b8c0de..70c66280db 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/QueuedSubmission.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/QueuedSubmission.ts @@ -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, clientUserMessageId: string, }; +export type QueuedSubmission = { id: string, input: Array, clientUserMessageId: string, +/** + * Metadata retained with this input, including across reconnect and restoration. + */ +responsesapiClientMetadata: { [key in string]?: string } | null, }; diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index 2a241ef175..8c7e5f5666 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -939,6 +939,8 @@ pub struct QueuedSubmission { pub id: String, pub input: Vec, pub client_user_message_id: String, + /// Metadata retained with this input, including across reconnect and restoration. + pub responsesapi_client_metadata: Option>, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] @@ -948,6 +950,9 @@ pub struct ThreadQueueAddParams { pub thread_id: String, pub input: Vec, pub client_user_message_id: String, + /// ResponsesAPI metadata captured from the client submitting this input. + #[ts(optional = nullable)] + pub responsesapi_client_metadata: Option>, } #[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, + /// Non-empty metadata replaces the input's metadata; omission or an empty map preserves it. + #[ts(optional = nullable)] + pub responsesapi_client_metadata: Option>, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] diff --git a/codex-rs/app-server/src/request_processors/thread_queue_processor.rs b/codex-rs/app-server/src/request_processors/thread_queue_processor.rs index a2f5b5ad68..e7a4de7624 100644 --- a/codex-rs/app-server/src/request_processors/thread_queue_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_queue_processor.rs @@ -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 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}"), diff --git a/codex-rs/ext/queue/src/service.rs b/codex-rs/ext/queue/src/service.rs index 06aeb63b4c..12671a0ef7 100644 --- a/codex-rs/ext/queue/src/service.rs +++ b/codex-rs/ext/queue/src/service.rs @@ -42,6 +42,7 @@ use uuid::Uuid; pub struct QueuedItem { pub id: String, pub input: TurnInput, + pub responsesapi_client_metadata: Option>, } #[derive(Debug, Error)] @@ -266,12 +267,22 @@ impl QueuedItemService { &self, thread_id: ThreadId, input: TurnInput, + ) -> Result { + 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>, ) -> Result { 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, 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>, + ) -> Result, 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::(&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 { - Ok(QueuedItem { - id: record.id, - input: serde_json::from_str::(&record.payload)?, - }) +fn queued_input_payload( + input: &TurnInput, + metadata: Option<&HashMap>, +) -> Result { + 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 for QueuedItem { + type Error = QueueServiceError; + + fn try_from(record: QueuedUserSubmissionRecord) -> Result { + 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, + }) + } } diff --git a/codex-rs/ext/queue/tests/queue_service.rs b/codex-rs/ext/queue/tests/queue_service.rs index 7730d4a6cd..e2a4cd43b7 100644 --- a/codex-rs/ext/queue/tests/queue_service.rs +++ b/codex-rs/ext/queue/tests/queue_service.rs @@ -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 = 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::(&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(()) } diff --git a/codex-rs/tui/src/session_queue_commands.rs b/codex-rs/tui/src/session_queue_commands.rs index 2abf0724dd..dcddb74998 100644 --- a/codex-rs/tui/src/session_queue_commands.rs +++ b/codex-rs/tui/src/session_queue_commands.rs @@ -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(), diff --git a/sdk/python/src/openai_codex/generated/v2_all.py b/sdk/python/src/openai_codex/generated/v2_all.py index 807a4d3a96..2b0ed292c8 100644 --- a/sdk/python/src/openai_codex/generated/v2_all.py +++ b/sdk/python/src/openai_codex/generated/v2_all.py @@ -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):