Support standalone tool outputs in turn/start (#41002)

## What changed

- Add an optional `toolOutput` to `turn/start` for starting or steering a turn with a named function-call output instead of user input.
- Emit standalone outputs as `functionCallOutput` thread items, persist them in durable history, and include them in resumed threads.
- Validate that tool outputs have a name, are not combined with nonempty `input`, satisfy the text-size limit, and use supported image URLs.

## Testing

- Cover standalone tool-output notifications, model input, and resumed history in both legacy and paginated history modes.
- Cover the text-size limit for structured tool output.

GitOrigin-RevId: a1eeec7b9b7de67b1a1c5e829a54552b85ec118b

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

@@ -4,17 +4,27 @@ use app_test_support::MockResponsesConfig;
 use app_test_support::TestAppServer;
 use codex_app_server_protocol::AdditionalContextEntry;
 use codex_app_server_protocol::AdditionalContextKind;
+use codex_app_server_protocol::ItemCompletedNotification;
+use codex_app_server_protocol::ItemStartedNotification;
+use codex_app_server_protocol::ThreadHistoryMode;
 use codex_app_server_protocol::ThreadInjectItemsParams;
 use codex_app_server_protocol::ThreadInjectItemsResponse;
+use codex_app_server_protocol::ThreadItem;
+use codex_app_server_protocol::ThreadResumeParams;
+use codex_app_server_protocol::ThreadResumeResponse;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
 use codex_app_server_protocol::TurnStartParams;
 use codex_app_server_protocol::TurnStartResponse;
+use codex_app_server_protocol::TurnStartedNotification;
+use codex_app_server_protocol::TurnStatus;
+use codex_app_server_protocol::TurnToolOutput;
 use codex_app_server_protocol::UserInput as V2UserInput;
 use codex_core::RolloutRecorder;
 use codex_features::Feature;
 use codex_protocol::ThreadId;
 use codex_protocol::models::ContentItem;
+use codex_protocol::models::FunctionCallOutputBody;
 use codex_protocol::models::ResponseItem;
 use codex_rollout::InitialHistory;
 use codex_rollout::RolloutItem;
@@ -27,12 +37,17 @@ use serde_json::Value;
 use serde_json::json;
 use std::collections::HashMap;
 use tempfile::TempDir;
+use test_case::test_case;
 use tokio::time::timeout;
 
 const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
 
+#[test_case(ThreadHistoryMode::Legacy; "legacy")]
+#[test_case(ThreadHistoryMode::Paginated; "paginated")]
 #[tokio::test]
-async fn thread_inject_items_adds_raw_response_items_to_thread_history() -> Result<()> {
+async fn thread_inject_items_adds_raw_response_items_to_thread_history(
+    history_mode: ThreadHistoryMode,
+) -> Result<()> {
     let server = responses::start_mock_server().await;
     let body = responses::sse(vec![
         responses::ev_response_created("resp-1"),
@@ -61,6 +76,7 @@ async fn thread_inject_items_adds_raw_response_items_to_thread_history() -> Resu
     let thread_req = mcp
         .send_thread_start_request_with_auto_env(ThreadStartParams {
             model: Some("mock-model".to_string()),
+            history_mode: Some(history_mode),
             ..Default::default()
         })
         .await?;
@@ -110,6 +126,8 @@ async fn thread_inject_items_adds_raw_response_items_to_thread_history() -> Resu
         .await?;
     let _response: ThreadInjectItemsResponse =
         timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(inject_req)).await??;
+    assert!(response_mock.requests().is_empty());
+
     assert_eq!(
         state_db
             .get_thread_memory_mode(ThreadId::from_string(&thread.id)?)
@@ -266,6 +284,79 @@ async fn thread_inject_items_adds_raw_response_items_to_thread_history() -> Resu
         "named unpaired tool output should be sent in the next model request"
     );
 
+    let delegated_mock = responses::mount_sse_once(
+        &server,
+        responses::sse(vec![responses::ev_completed("resp-delegated")]),
+    )
+    .await;
+    mcp.clear_message_buffer();
+    let delegated_req = mcp
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id.clone(),
+            input: Vec::new(),
+            tool_output: Some(Box::new(TurnToolOutput {
+                name: "send_message_to_thread".to_string(),
+                namespace: Some("codex_app".to_string()),
+                output: FunctionCallOutputBody::Text("Start a delegated turn.".to_string()),
+            })),
+            ..Default::default()
+        })
+        .await?;
+    let TurnStartResponse { turn } =
+        timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(delegated_req)).await??;
+    let turn_started: TurnStartedNotification =
+        timeout(DEFAULT_READ_TIMEOUT, mcp.read_notification("turn/started")).await??;
+    let item_started: ItemStartedNotification =
+        timeout(DEFAULT_READ_TIMEOUT, mcp.read_notification("item/started")).await??;
+    let item_completed: ItemCompletedNotification = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_notification("item/completed"),
+    )
+    .await??;
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("turn/completed"),
+    )
+    .await??;
+    assert_eq!(turn.id, turn_started.turn.id);
+    assert_eq!(turn_started.turn.status, TurnStatus::InProgress);
+    assert_eq!(item_completed.turn_id, turn_started.turn.id);
+    assert_eq!(item_started.item, item_completed.item);
+    assert!(matches!(
+        &item_completed.item,
+        ThreadItem::FunctionCallOutput {
+            name,
+            namespace,
+            output: FunctionCallOutputBody::Text(output),
+            ..
+        } if name == "send_message_to_thread"
+            && namespace.as_deref() == Some("codex_app")
+            && output == "Start a delegated turn."
+    ));
+    assert!(
+        delegated_mock
+            .single_request()
+            .input()
+            .into_iter()
+            .any(|item| item["type"] == "function_call_output"
+                && item["output"] == "Start a delegated turn."),
+        "the delegated turn must preserve the original model-visible tool output"
+    );
+    let resume_req = mcp
+        .send_thread_resume_request(ThreadResumeParams {
+            thread_id: thread.id,
+            ..Default::default()
+        })
+        .await?;
+    let ThreadResumeResponse { thread, .. } =
+        timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(resume_req)).await??;
+    assert_eq!(thread.turns.len(), 2);
+    assert!(thread.turns.iter().any(|turn| {
+        turn.id == turn_started.turn.id
+            && turn.status == TurnStatus::Completed
+            && turn.items.contains(&item_completed.item)
+    }));
+
     Ok(())
 }
 

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

@@ -1473,15 +1473,15 @@ async fn turn_start_rejects_combined_oversized_text_input() -> Result<()> {
 
     let turn_req = mcp
         .send_turn_start_request(TurnStartParams {
-            thread_id: thread.id,
+            thread_id: thread.id.clone(),
             client_user_message_id: None,
             input: vec![
                 V2UserInput::Text {
-                    text: first,
+                    text: first.clone(),
                     text_elements: Vec::new(),
                 },
                 V2UserInput::Text {
-                    text: second,
+                    text: second.clone(),
                     text_elements: Vec::new(),
                 },
             ],
@@ -1504,6 +1504,30 @@ async fn turn_start_rejects_combined_oversized_text_input() -> Result<()> {
     assert_eq!(data["max_chars"], MAX_USER_INPUT_TEXT_CHARS);
     assert_eq!(data["actual_chars"], actual_chars);
 
+    let tool_req = mcp
+        .send_raw_request(
+            "turn/start",
+            Some(json!({
+                "threadId": thread.id,
+                "input": [],
+                "toolOutput": {
+                    "name": "send_message_to_thread",
+                    "output": [
+                        { "type": "input_text", "text": first },
+                        { "type": "input_text", "text": second },
+                    ],
+                },
+            })),
+        )
+        .await?;
+    let err: JSONRPCError = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_error_message(RequestId::Integer(tool_req)),
+    )
+    .await??;
+    assert_eq!(err.error.code, INVALID_PARAMS_ERROR_CODE);
+    assert_eq!(err.error.data, Some(data));
+
     let turn_started = tokio::time::timeout(
         std::time::Duration::from_millis(250),
         mcp.read_stream_until_notification_message("turn/started"),
@@ -3003,6 +3027,7 @@ async fn turn_start_explicit_local_environment_updates_legacy_cwd_between_turns(
                     text_elements: Vec::new(),
                 }],
                 turn_trigger: None,
+                tool_output: None,
                 responsesapi_client_metadata: None,
                 additional_context: None,
                 cwd: Some(first_cwd.clone()),
@@ -3054,6 +3079,7 @@ async fn turn_start_explicit_local_environment_updates_legacy_cwd_between_turns(
                     text_elements: Vec::new(),
                 }],
                 turn_trigger: None,
+                tool_output: None,
                 responsesapi_client_metadata: None,
                 additional_context: None,
                 cwd: None,