codex: seed turn metadata from app-server turn start

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

codex-rs/app-server/README.md

@@ -69,7 +69,7 @@ Use the thread APIs to create, list, or archive conversations. Drive a conversat
 - Initialize once per connection: Immediately after opening a transport connection, send an `initialize` request with your client metadata, then emit an `initialized` notification. Any other request on that connection before this handshake gets rejected.
 - Start (or resume) a thread: Call `thread/start` to open a fresh conversation. The response returns the thread object and you’ll also get a `thread/started` notification. If you’re continuing an existing conversation, call `thread/resume` with its ID instead. If you want to branch from an existing conversation, call `thread/fork` to create a new thread id with copied history. Like `thread/start`, `thread/fork` also accepts `ephemeral: true` for an in-memory temporary thread.
   The returned `thread.ephemeral` flag tells you whether the session is intentionally in-memory only; when it is `true`, `thread.path` is `null`.
-- Begin a turn: To send user input, call `turn/start` with the target `threadId` and the user's input. Optional fields let you override model, cwd, sandbox policy, approval policy, approvals reviewer, etc. This immediately returns the new turn object. The app-server emits `turn/started` when that turn actually begins running.
+- Begin a turn: To send user input, call `turn/start` with the target `threadId` and the user's input. Optional fields let you override model, cwd, sandbox policy, approval policy, approvals reviewer, etc. `turn/start.metadata` also lets callers attach arbitrary string metadata to that turn. This immediately returns the new turn object. The app-server emits `turn/started` when that turn actually begins running.
 - Stream events: After `turn/start`, keep reading JSON-RPC notifications on stdout. You’ll see `item/started`, `item/completed`, deltas like `item/agentMessage/delta`, tool progress, etc. These represent streaming model output plus any side effects (commands, tool calls, reasoning notes).
 - Finish the turn: When the model is done (or the turn is interrupted via making the `turn/interrupt` call), the server sends `turn/completed` with the final turn state and token usage.
 
@@ -139,7 +139,7 @@ Example with notification opt-out:
 - `thread/shellCommand` — run a user-initiated `!` shell command against a thread; this runs unsandboxed with full access rather than inheriting the thread sandbox policy. Returns `{}` immediately while progress streams through standard turn/item notifications and any active turn receives the formatted output in its message stream.
 - `thread/backgroundTerminals/clean` — terminate all running background terminals for a thread (experimental; requires `capabilities.experimentalApi`); returns `{}` when the cleanup request is accepted.
 - `thread/rollback` — drop the last N turns from the agent’s in-memory context and persist a rollback marker in the rollout so future resumes see the pruned history; returns the updated `thread` (with `turns` populated) on success.
-- `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode".
+- `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. `metadata` accepts arbitrary string key/value pairs for turn-scoped observability and request forwarding. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode".
 - `turn/steer` — add user input to an already in-flight turn without starting a new turn; returns the active `turnId` that accepted the input.
 - `turn/interrupt` — request cancellation of an in-flight turn by `(thread_id, turn_id)`; success is an empty `{}` response and the turn finishes with `status: "interrupted"`.
 - `thread/realtime/start` — start a thread-scoped realtime session (experimental); returns `{}` and streams `thread/realtime/*` notifications.

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

@@ -6018,6 +6018,7 @@ impl CodexMessageProcessor {
                 Op::UserInput {
                     items: mapped_items,
                     final_output_json_schema: params.output_schema,
+                    metadata: params.metadata,
                 },
             )
             .await;

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

@@ -857,6 +857,176 @@ mod tests {
             .expect("in-process runtime should shutdown cleanly");
     }
 
+    #[tokio::test]
+    async fn in_process_start_forwards_turn_metadata_headers_on_turn_requests() {
+        let server = create_mock_responses_server_repeating_assistant("Done").await;
+        let codex_home = TempDir::new().expect("tempdir should create");
+        let mut client = start(InProcessStartArgs {
+            arg0_paths: Arg0DispatchPaths::default(),
+            config: Arc::new(
+                ConfigBuilder::default()
+                    .codex_home(codex_home.path().to_path_buf())
+                    .build()
+                    .await
+                    .expect("test config should load"),
+            ),
+            cli_overrides: vec![
+                (
+                    "model_provider".to_string(),
+                    TomlValue::String("mock_provider".to_string()),
+                ),
+                (
+                    "model_providers.mock_provider.name".to_string(),
+                    TomlValue::String("Mock provider for test".to_string()),
+                ),
+                (
+                    "model_providers.mock_provider.base_url".to_string(),
+                    TomlValue::String(format!("{}/v1", server.uri())),
+                ),
+                (
+                    "model_providers.mock_provider.wire_api".to_string(),
+                    TomlValue::String("responses".to_string()),
+                ),
+                (
+                    "model_providers.mock_provider.request_max_retries".to_string(),
+                    TomlValue::Integer(0),
+                ),
+                (
+                    "model_providers.mock_provider.stream_max_retries".to_string(),
+                    TomlValue::Integer(0),
+                ),
+            ],
+            loader_overrides: LoaderOverrides::default(),
+            cloud_requirements: CloudRequirementsLoader::default(),
+            auth_manager: None,
+            thread_manager: None,
+            feedback: CodexFeedback::new(),
+            config_warnings: Vec::new(),
+            session_source: SessionSource::Cli,
+            enable_codex_api_key_env: false,
+            initialize: InitializeParams {
+                client_info: ClientInfo {
+                    name: "codex-in-process-test".to_string(),
+                    title: None,
+                    version: "0.0.0".to_string(),
+                },
+                capabilities: None,
+            },
+            channel_capacity: DEFAULT_IN_PROCESS_CHANNEL_CAPACITY,
+        })
+        .await
+        .expect("in-process runtime should start");
+
+        let thread_start_response: ThreadStartResponse = serde_json::from_value(
+            client
+                .request(ClientRequest::ThreadStart {
+                    request_id: RequestId::Integer(5),
+                    params: ThreadStartParams {
+                        model: Some("mock-model".to_string()),
+                        ephemeral: Some(true),
+                        ..ThreadStartParams::default()
+                    },
+                })
+                .await
+                .expect("thread/start request should work")
+                .expect("thread/start should succeed"),
+        )
+        .expect("thread/start response should parse");
+
+        let _turn_start_response: TurnStartResponse = serde_json::from_value(
+            client
+                .request(ClientRequest::TurnStart {
+                    request_id: RequestId::Integer(6),
+                    params: TurnStartParams {
+                        thread_id: thread_start_response.thread.id,
+                        input: vec![UserInput::Text {
+                            text: "Hello".to_string(),
+                            text_elements: Vec::new(),
+                        }],
+                        metadata: Some(std::collections::BTreeMap::from([
+                            (
+                                "parentConversationId".to_string(),
+                                "parent-conversation-123".to_string(),
+                            ),
+                            (
+                                "parentMessageId".to_string(),
+                                "parent-message-123".to_string(),
+                            ),
+                            ("parentTurnId".to_string(), "parent-turn-123".to_string()),
+                        ])),
+                        ..TurnStartParams::default()
+                    },
+                })
+                .await
+                .expect("turn/start request should work")
+                .expect("turn/start should succeed"),
+        )
+        .expect("turn/start response should parse");
+
+        let turn_completed = loop {
+            let event = timeout(std::time::Duration::from_secs(10), client.next_event())
+                .await
+                .expect("timed out waiting for turn completion");
+            match event {
+                Some(InProcessServerEvent::ServerNotification(
+                    ServerNotification::TurnCompleted(notification),
+                )) => break notification,
+                Some(_) => continue,
+                None => panic!("in-process runtime exited before turn completion"),
+            }
+        };
+        assert_eq!(
+            turn_completed.turn.status,
+            TurnStatus::Completed,
+            "turn error: {:?}",
+            turn_completed.turn.error
+        );
+        assert_eq!(turn_completed.turn.error, None);
+
+        let requests = timeout(std::time::Duration::from_secs(5), async {
+            loop {
+                let requests = server
+                    .received_requests()
+                    .await
+                    .expect("received requests should be available");
+                if !requests.is_empty() {
+                    break requests;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("timed out waiting for outbound requests");
+        assert!(!requests.is_empty());
+        for request in requests {
+            assert_eq!(
+                request
+                    .headers
+                    .get("x-openai-parent-conversation-id")
+                    .and_then(|value| value.to_str().ok()),
+                Some("parent-conversation-123")
+            );
+            assert_eq!(
+                request
+                    .headers
+                    .get("x-openai-parent-message-id")
+                    .and_then(|value| value.to_str().ok()),
+                Some("parent-message-123")
+            );
+            assert_eq!(
+                request
+                    .headers
+                    .get("x-openai-parent-turn-id")
+                    .and_then(|value| value.to_str().ok()),
+                Some("parent-turn-123")
+            );
+        }
+
+        client
+            .shutdown()
+            .await
+            .expect("in-process runtime should shutdown cleanly");
+    }
     #[test]
     fn guaranteed_delivery_helpers_cover_terminal_notifications() {
         assert!(server_notification_requires_delivery(

codex-rs/app-server/src/message_processor/tracing_tests.rs

@@ -601,6 +601,7 @@ async fn turn_start_jsonrpc_span_parents_core_turn_spans() -> Result<()> {
                     personality: None,
                     output_schema: None,
                     collaboration_mode: None,
+                    metadata: None,
                 },
             },
             Some(remote_trace),

codex-rs/app-server/tests/suite/v2/turn_start.rs

@@ -1395,6 +1395,7 @@ async fn turn_start_updates_sandbox_and_cwd_between_turns_v2() -> Result<()> {
             personality: None,
             output_schema: None,
             collaboration_mode: None,
+            metadata: None,
         })
         .await?;
     timeout(
@@ -1428,6 +1429,7 @@ async fn turn_start_updates_sandbox_and_cwd_between_turns_v2() -> Result<()> {
             personality: None,
             output_schema: None,
             collaboration_mode: None,
+            metadata: None,
         })
         .await?;
     timeout(