diff --git a/codex-rs/app-server-protocol/src/protocol/v2/turn.rs b/codex-rs/app-server-protocol/src/protocol/v2/turn.rs index 7e20de4ac..0b05408be 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/turn.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/turn.rs @@ -96,6 +96,8 @@ pub struct TurnStartParams { #[ts(optional = nullable)] pub responsesapi_client_metadata: Option>, /// Optional client-provided context fragments keyed by an opaque source identifier. + // App-server request validation only accepts `untrusted` here. `application` is reserved for + // trusted server-owned context handled before client request submission. #[experimental("turn/start.additionalContext")] #[ts(optional = nullable)] pub additional_context: Option>, @@ -192,6 +194,8 @@ pub struct TurnSteerParams { #[ts(optional = nullable)] pub responsesapi_client_metadata: Option>, /// Optional client-provided context fragments keyed by an opaque source identifier. + // App-server request validation only accepts `untrusted` here. `application` is reserved for + // trusted server-owned context handled before client request submission. #[experimental("turn/steer.additionalContext")] #[ts(optional = nullable)] pub additional_context: Option>, diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 8a90c4e4c..cc22df7c9 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -186,9 +186,9 @@ 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 Codewith generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Pass `modelProvider` with `model` when switching to a provider whose model slugs are not provider-prefixed. Experimental `additionalContext` injects keyed context fragments before the turn input; each entry declares its trust `kind` and may include `source` provenance for local app-store records such as OpenProjects projects, OpenTodos tasks, Conversations threads, or Mementos notes. Experimental `runtimeWorkspaceRoots` replaces the thread-scoped runtime workspace roots used to materialize `:workspace_roots`; paths must be absolute. Prefer experimental `permissions` profile selection by id for permission overrides; the legacy `sandboxPolicy` field is still accepted but cannot be combined with `permissions`. 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 Codewith generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Pass `modelProvider` with `model` when switching to a provider whose model slugs are not provider-prefixed. Experimental `additionalContext` injects keyed context fragments before the turn input; normal app-server clients may only send `kind: "untrusted"` and may include `source` provenance for local app-store records such as OpenProjects projects, OpenTodos tasks, Conversations threads, or Mementos notes. `kind: "application"` is reserved for trusted server-owned context and is rejected by `turn/start`. Experimental `runtimeWorkspaceRoots` replaces the thread-scoped runtime workspace roots used to materialize `:workspace_roots`; paths must be absolute. Prefer experimental `permissions` profile selection by id for permission overrides; the legacy `sandboxPolicy` field is still accepted but cannot be combined with `permissions`. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode". - `thread/inject_items` — append raw Responses API items to a loaded thread’s model-visible history without starting a user turn; returns `{}` on success. -- `turn/steer` — add user input to an already in-flight regular turn without starting a new turn; returns the active `turnId` that accepted the input. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Like `turn/start`, experimental `additionalContext` may refresh keyed context for the active turn. Review and manual compaction turns reject `turn/steer`. +- `turn/steer` — add user input to an already in-flight regular turn without starting a new turn; returns the active `turnId` that accepted the input. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Experimental `additionalContext` has the same trust and size limits as `turn/start` and may refresh keyed untrusted context for the active turn. Review and manual compaction turns reject `turn/steer`. - `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); pass `outputModality: "text"` or `outputModality: "audio"` to choose model output, returns `{}` and streams `thread/realtime/*` notifications. Omit `transport` for the websocket transport, or pass `{ "type": "webrtc", "sdp": "..." }` to create a WebRTC session from a browser-generated SDP offer; the remote answer SDP is emitted as `thread/realtime/sdp`. - `thread/realtime/appendAudio` — append an input audio chunk to the active realtime session (experimental); returns `{}`. @@ -1323,7 +1323,12 @@ You can optionally specify config overrides on the new turn. If specified, these ### Example: Start a turn with sourced context -Use experimental `additionalContext` when a host app wants Codewith to see local app-store context without adding a visible user message. The map key is the stable dedupe key for that fragment. `kind` controls trust: use `"untrusted"` for project/task/conversation/memento content unless it was generated by the trusted host application. Optional `source` metadata identifies the local record so clients can render, edit, refresh, or remove the context source without granting the agent hidden write access to that store. +Use experimental `additionalContext` when a host app wants Codewith to see local app-store context without adding a visible user message. The map key is the stable dedupe key for that fragment. Normal app-server clients may only send entries with `kind: "untrusted"`; these are rendered as external user context. `kind: "application"` is reserved for trusted server-owned context and is rejected on client `turn/start` and `turn/steer` requests. Optional `source` metadata identifies the local record so clients can render, edit, refresh, or remove the context source without granting the agent hidden write access to that store. + +The request is rejected with `invalid params` before model submission if +`additionalContext` exceeds 16 entries, a key exceeds 128 characters, a value exceeds 65,536 +bytes, or all keys, values, and source metadata exceed 131,072 bytes. Limit errors include +`input_error_code: "input_too_large"` in the JSON-RPC error data. ```json { "method": "turn/start", "id": 31, "params": { @@ -1507,6 +1512,9 @@ Use `thread/backgroundTerminals/clean` to terminate all running background termi Use `turn/steer` to append additional user input to the currently active regular turn. This does not emit `turn/started` and does not accept thread settings overrides. +`additionalContext`, when provided, follows the same `untrusted`-only trust boundary and size +limits documented for `turn/start`. + ```json { "method": "turn/steer", "id": 32, "params": { "threadId": "thr_123", diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index 67e50d2aa..08c56370a 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -2,6 +2,11 @@ use super::*; use codex_protocol::protocol::AdditionalContextEntry as CoreAdditionalContextEntry; use codex_protocol::protocol::AdditionalContextKind as CoreAdditionalContextKind; +const MAX_ADDITIONAL_CONTEXT_ENTRIES: usize = 16; +const MAX_ADDITIONAL_CONTEXT_KEY_CHARS: usize = 128; +const MAX_ADDITIONAL_CONTEXT_VALUE_BYTES: usize = 64 * 1024; +const MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES: usize = 128 * 1024; + #[derive(Clone)] pub(crate) struct TurnRequestProcessor { auth_manager: Arc, @@ -20,32 +25,142 @@ pub(crate) struct TurnRequestProcessor { local_active_owner_id: String, } -fn map_additional_context( +fn additional_context_input_too_large_error( + message: String, + data: serde_json::Value, +) -> JSONRPCErrorError { + let mut error = invalid_params(message); + error.data = Some(data); + error +} + +fn reject_application_additional_context_error() -> JSONRPCErrorError { + invalid_params( + "additionalContext kind `application` is reserved for trusted server-owned context and \ + cannot be supplied by turn/start or turn/steer clients", + ) +} + +fn additional_context_error_type(error: &JSONRPCErrorError) -> Option { + let is_too_large = error.data.as_ref().is_some_and(|data| { + data.get("input_error_code") + .and_then(serde_json::Value::as_str) + == Some(INPUT_TOO_LARGE_ERROR_CODE) + }); + if is_too_large { + Some(AnalyticsJsonRpcError::Input(InputError::TooLarge)) + } else { + None + } +} + +fn validate_and_map_additional_context( additional_context: Option>, -) -> BTreeMap { - additional_context - .unwrap_or_default() +) -> Result, JSONRPCErrorError> { + let Some(additional_context) = additional_context else { + return Ok(BTreeMap::new()); + }; + + let entry_count = additional_context.len(); + if entry_count > MAX_ADDITIONAL_CONTEXT_ENTRIES { + return Err(additional_context_input_too_large_error( + format!( + "additionalContext must not contain more than \ + {MAX_ADDITIONAL_CONTEXT_ENTRIES} entries." + ), + serde_json::json!({ + "input_error_code": INPUT_TOO_LARGE_ERROR_CODE, + "input_field": "additionalContext", + "limit_name": "max_entries", + "max_entries": MAX_ADDITIONAL_CONTEXT_ENTRIES, + "actual_entries": entry_count, + }), + )); + } + + if additional_context + .values() + .any(|entry| matches!(entry.kind, AdditionalContextKind::Application)) + { + return Err(reject_application_additional_context_error()); + } + + let mut total_bytes = 0usize; + for (key, entry) in &additional_context { + let key_chars = key.chars().count(); + if key_chars > MAX_ADDITIONAL_CONTEXT_KEY_CHARS { + return Err(additional_context_input_too_large_error( + format!( + "additionalContext keys must not exceed \ + {MAX_ADDITIONAL_CONTEXT_KEY_CHARS} characters." + ), + serde_json::json!({ + "input_error_code": INPUT_TOO_LARGE_ERROR_CODE, + "input_field": "additionalContext", + "limit_name": "max_key_chars", + "max_chars": MAX_ADDITIONAL_CONTEXT_KEY_CHARS, + "actual_chars": key_chars, + }), + )); + } + + let value_bytes = entry.value.len(); + if value_bytes > MAX_ADDITIONAL_CONTEXT_VALUE_BYTES { + return Err(additional_context_input_too_large_error( + format!( + "additionalContext values must not exceed \ + {MAX_ADDITIONAL_CONTEXT_VALUE_BYTES} bytes." + ), + serde_json::json!({ + "input_error_code": INPUT_TOO_LARGE_ERROR_CODE, + "input_field": "additionalContext", + "limit_name": "max_value_bytes", + "max_bytes": MAX_ADDITIONAL_CONTEXT_VALUE_BYTES, + "actual_bytes": value_bytes, + }), + )); + } + + total_bytes = total_bytes + .saturating_add(key.len()) + .saturating_add(value_bytes); + if let Some(source) = &entry.source { + total_bytes = total_bytes + .saturating_add(source.namespace.len()) + .saturating_add(source.id.len()) + .saturating_add(source.record_type.as_deref().map(str::len).unwrap_or(0)) + .saturating_add(source.label.as_deref().map(str::len).unwrap_or(0)); + } + } + + if total_bytes > MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES { + return Err(additional_context_input_too_large_error( + format!( + "additionalContext must not exceed \ + {MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES} aggregate bytes." + ), + serde_json::json!({ + "input_error_code": INPUT_TOO_LARGE_ERROR_CODE, + "input_field": "additionalContext", + "limit_name": "max_total_bytes", + "max_bytes": MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES, + "actual_bytes": total_bytes, + }), + )); + } + + Ok(additional_context .into_iter() .map(|(key, entry)| { - let AdditionalContextEntry { - value, - kind, - source: _, - } = entry; ( key, CoreAdditionalContextEntry { - value, - kind: match kind { - AdditionalContextKind::Untrusted => CoreAdditionalContextKind::Untrusted, - AdditionalContextKind::Application => { - CoreAdditionalContextKind::Application - } - }, + value: entry.value, + kind: CoreAdditionalContextKind::Untrusted, }, ) }) - .collect() + .collect()) } fn existing_dir_is_inside(path: &Path, root: &Path) -> std::io::Result { @@ -430,7 +545,18 @@ impl TurnRequestProcessor { .map(V2UserInput::into_core) .collect(); let client_user_message_id = params.client_user_message_id; - let additional_context = map_additional_context(params.additional_context); + let additional_context = + match validate_and_map_additional_context(params.additional_context) { + Ok(additional_context) => additional_context, + Err(error) => { + self.track_error_response( + &request_id, + &error, + additional_context_error_type(&error), + ); + return Err(error); + } + }; let turn_has_input = !mapped_items.is_empty(); let cwd = resolve_request_cwd(params.cwd)?; let thread_settings = self @@ -872,7 +998,18 @@ impl TurnRequestProcessor { .into_iter() .map(V2UserInput::into_core) .collect(); - let additional_context = map_additional_context(params.additional_context); + let additional_context = + match validate_and_map_additional_context(params.additional_context) { + Ok(additional_context) => additional_context, + Err(error) => { + self.track_error_response( + request_id, + &error, + additional_context_error_type(&error), + ); + return Err(error); + } + }; let turn_id = thread .steer_input( diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index f7e2ddebd..8a29b0fc2 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -20,6 +20,7 @@ use codex_app_server::INPUT_TOO_LARGE_ERROR_CODE; use codex_app_server::INVALID_PARAMS_ERROR_CODE; use codex_app_server_protocol::AdditionalContextEntry; use codex_app_server_protocol::AdditionalContextKind; +use codex_app_server_protocol::AdditionalContextSource; use codex_app_server_protocol::ByteRange; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::CollabAgentStatus; @@ -84,6 +85,7 @@ use std::path::Path; use tempfile::TempDir; use tokio::time::timeout; use wiremock::Mock; +use wiremock::MockServer; use wiremock::ResponseTemplate; use wiremock::matchers::method; use wiremock::matchers::path; @@ -98,6 +100,10 @@ const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs const TEST_ORIGINATOR: &str = "codex_vscode"; const LOCAL_PRAGMATIC_TEMPLATE: &str = "You are a deeply pragmatic, effective software engineer."; const INVALID_REQUEST_ERROR_CODE: i64 = -32600; +const MAX_ADDITIONAL_CONTEXT_ENTRIES: usize = 16; +const MAX_ADDITIONAL_CONTEXT_KEY_CHARS: usize = 128; +const MAX_ADDITIONAL_CONTEXT_VALUE_BYTES: usize = 64 * 1024; +const MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES: usize = 128 * 1024; const TINY_PNG_BYTES: &[u8] = &[ 137, 80, 78, 71, 13, 10, 26, 10, 0, 0, 0, 13, 73, 72, 68, 82, 0, 0, 0, 1, 0, 0, 0, 1, 8, 6, 0, 0, 0, 31, 21, 196, 137, 0, 0, 0, 11, 73, 68, 65, 84, 120, 156, 99, 96, 0, 2, 0, 0, 5, 0, 1, @@ -110,6 +116,96 @@ fn body_contains(req: &wiremock::Request, text: &str) -> bool { .is_some_and(|body| body.contains(text)) } +async fn start_mock_thread(mcp: &mut TestAppServer) -> Result { + let thread_req = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let thread_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_req)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response::(thread_resp)?; + Ok(thread.id) +} + +async fn expect_turn_start_additional_context_error( + mcp: &mut TestAppServer, + thread_id: &str, + additional_context: HashMap, +) -> Result { + let turn_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread_id.to_string(), + client_user_message_id: None, + input: vec![V2UserInput::Text { + text: "inspect tab".to_string(), + text_elements: Vec::new(), + }], + additional_context: Some(additional_context), + ..Default::default() + }) + .await?; + let err = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(turn_req)), + ) + .await??; + + let turn_started = tokio::time::timeout( + std::time::Duration::from_millis(250), + mcp.read_stream_until_notification_message("turn/started"), + ) + .await; + assert!( + turn_started.is_err(), + "did not expect a turn/started notification for rejected additionalContext" + ); + + Ok(err) +} + +fn untrusted_context_entry(value: impl Into) -> AdditionalContextEntry { + AdditionalContextEntry { + value: value.into(), + kind: AdditionalContextKind::Untrusted, + source: None, + } +} + +fn assert_additional_context_too_large_error( + err: JSONRPCError, + message: &str, + limit_name: &str, +) -> Value { + assert_eq!(err.error.code, INVALID_PARAMS_ERROR_CODE); + assert_eq!(err.error.message, message); + let Some(data) = err.error.data else { + panic!("expected structured error data"); + }; + assert_eq!(data["input_error_code"], INPUT_TOO_LARGE_ERROR_CODE); + assert_eq!(data["input_field"], "additionalContext"); + assert_eq!(data["limit_name"], limit_name); + data +} + +async fn assert_no_responses_requests(server: &MockServer) -> Result<()> { + let requests = server + .received_requests() + .await + .context("failed to fetch received requests")?; + assert!( + requests + .iter() + .all(|request| !request.url.path().ends_with("/responses")), + "rejected turn/start should not send a model request" + ); + Ok(()) +} + #[tokio::test] async fn turn_start_unprefixed_xiaomi_model_routes_to_xiaomi_chat_provider() -> Result<()> { let server = create_mock_responses_server_sequence_unchecked(vec![ @@ -500,6 +596,174 @@ async fn turn_start_additional_context_flows_to_model_input() -> Result<()> { body.to_string() .contains("source value") ); + let body_text = body.to_string(); + assert!( + !body_text.contains("source value"), + "untrusted additionalContext must not be rendered as application developer context" + ); + + Ok(()) +} + +#[tokio::test] +async fn turn_start_additional_context_rejects_application_and_limits() -> Result<()> { + let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await; + + let codex_home = TempDir::new()?; + create_config_toml( + codex_home.path(), + &server.uri(), + "never", + &BTreeMap::default(), + )?; + + let mut mcp = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + let thread_id = start_mock_thread(&mut mcp).await?; + + let application_err = expect_turn_start_additional_context_error( + &mut mcp, + &thread_id, + HashMap::from([( + "custom_source".to_string(), + AdditionalContextEntry { + value: "source value".to_string(), + kind: AdditionalContextKind::Application, + source: None, + }, + )]), + ) + .await?; + assert_eq!(application_err.error.code, INVALID_PARAMS_ERROR_CODE); + assert_eq!( + application_err.error.message, + "additionalContext kind `application` is reserved for trusted server-owned context and \ + cannot be supplied by turn/start or turn/steer clients" + ); + assert!( + application_err.error.data.is_none(), + "application kind rejection should not be reported as an input-too-large error" + ); + + let too_many_entries = (0..=MAX_ADDITIONAL_CONTEXT_ENTRIES) + .map(|index| { + ( + format!("source_{index}"), + untrusted_context_entry(format!("value {index}")), + ) + }) + .collect::>(); + let too_many_err = + expect_turn_start_additional_context_error(&mut mcp, &thread_id, too_many_entries).await?; + let data = assert_additional_context_too_large_error( + too_many_err, + &format!( + "additionalContext must not contain more than \ + {MAX_ADDITIONAL_CONTEXT_ENTRIES} entries." + ), + "max_entries", + ); + assert_eq!(data["max_entries"], MAX_ADDITIONAL_CONTEXT_ENTRIES); + assert_eq!(data["actual_entries"], MAX_ADDITIONAL_CONTEXT_ENTRIES + 1); + + let long_key = "k".repeat(MAX_ADDITIONAL_CONTEXT_KEY_CHARS + 1); + let long_key_err = expect_turn_start_additional_context_error( + &mut mcp, + &thread_id, + HashMap::from([(long_key, untrusted_context_entry("value"))]), + ) + .await?; + let data = assert_additional_context_too_large_error( + long_key_err, + &format!( + "additionalContext keys must not exceed \ + {MAX_ADDITIONAL_CONTEXT_KEY_CHARS} characters." + ), + "max_key_chars", + ); + assert_eq!(data["max_chars"], MAX_ADDITIONAL_CONTEXT_KEY_CHARS); + assert_eq!(data["actual_chars"], MAX_ADDITIONAL_CONTEXT_KEY_CHARS + 1); + + let oversized_value = "x".repeat(MAX_ADDITIONAL_CONTEXT_VALUE_BYTES + 1); + let oversized_value_err = expect_turn_start_additional_context_error( + &mut mcp, + &thread_id, + HashMap::from([( + "source".to_string(), + untrusted_context_entry(oversized_value), + )]), + ) + .await?; + let data = assert_additional_context_too_large_error( + oversized_value_err, + &format!( + "additionalContext values must not exceed \ + {MAX_ADDITIONAL_CONTEXT_VALUE_BYTES} bytes." + ), + "max_value_bytes", + ); + assert_eq!(data["max_bytes"], MAX_ADDITIONAL_CONTEXT_VALUE_BYTES); + assert_eq!(data["actual_bytes"], MAX_ADDITIONAL_CONTEXT_VALUE_BYTES + 1); + + let aggregate_context = (0..3) + .map(|index| { + ( + format!("source_{index}"), + untrusted_context_entry("x".repeat(44 * 1024)), + ) + }) + .collect::>(); + let aggregate_err = + expect_turn_start_additional_context_error(&mut mcp, &thread_id, aggregate_context).await?; + let data = assert_additional_context_too_large_error( + aggregate_err, + &format!( + "additionalContext must not exceed \ + {MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES} aggregate bytes." + ), + "max_total_bytes", + ); + assert_eq!(data["max_bytes"], MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES); + assert!( + data["actual_bytes"] + .as_u64() + .is_some_and(|actual| actual > MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES as u64) + ); + + let source_metadata_err = expect_turn_start_additional_context_error( + &mut mcp, + &thread_id, + HashMap::from([( + "source".to_string(), + AdditionalContextEntry { + value: "value".to_string(), + kind: AdditionalContextKind::Untrusted, + source: Some(AdditionalContextSource { + namespace: "open-todos".to_string(), + id: "task".to_string(), + record_type: Some("task".to_string()), + label: Some("x".repeat(MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES)), + }), + }, + )]), + ) + .await?; + let data = assert_additional_context_too_large_error( + source_metadata_err, + &format!( + "additionalContext must not exceed \ + {MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES} aggregate bytes." + ), + "max_total_bytes", + ); + assert_eq!(data["max_bytes"], MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES); + assert!( + data["actual_bytes"] + .as_u64() + .is_some_and(|actual| actual > MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES as u64) + ); + + assert_no_responses_requests(&server).await?; Ok(()) } diff --git a/codex-rs/app-server/tests/suite/v2/turn_steer.rs b/codex-rs/app-server/tests/suite/v2/turn_steer.rs index 702dfb441..659c70fd7 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_steer.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_steer.rs @@ -12,6 +12,7 @@ use codex_app_server::INPUT_TOO_LARGE_ERROR_CODE; use codex_app_server::INVALID_PARAMS_ERROR_CODE; use codex_app_server_protocol::AdditionalContextEntry; use codex_app_server_protocol::AdditionalContextKind; +use codex_app_server_protocol::AdditionalContextSource; use codex_app_server_protocol::ItemStartedNotification; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCNotification; @@ -35,6 +36,61 @@ use super::analytics::mount_analytics_capture; use super::analytics::wait_for_analytics_event; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +const MAX_ADDITIONAL_CONTEXT_ENTRIES: usize = 16; +const MAX_ADDITIONAL_CONTEXT_KEY_CHARS: usize = 128; +const MAX_ADDITIONAL_CONTEXT_VALUE_BYTES: usize = 64 * 1024; +const MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES: usize = 128 * 1024; + +fn untrusted_context_entry(value: impl Into) -> AdditionalContextEntry { + AdditionalContextEntry { + value: value.into(), + kind: AdditionalContextKind::Untrusted, + source: None, + } +} + +async fn expect_turn_steer_additional_context_error( + mcp: &mut TestAppServer, + thread_id: &str, + turn_id: &str, + additional_context: HashMap, +) -> Result { + let steer_req = mcp + .send_turn_steer_request(TurnSteerParams { + thread_id: thread_id.to_string(), + client_user_message_id: None, + input: vec![V2UserInput::Text { + text: "steer".to_string(), + text_elements: Vec::new(), + }], + responsesapi_client_metadata: None, + additional_context: Some(additional_context), + expected_turn_id: turn_id.to_string(), + }) + .await?; + let err = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(steer_req)), + ) + .await??; + Ok(err) +} + +fn assert_additional_context_too_large_error( + err: JSONRPCError, + message: &str, + limit_name: &str, +) -> Value { + assert_eq!(err.error.code, INVALID_PARAMS_ERROR_CODE); + assert_eq!(err.error.message, message); + let Some(data) = err.error.data else { + panic!("expected structured error data"); + }; + assert_eq!(data["input_error_code"], INPUT_TOO_LARGE_ERROR_CODE); + assert_eq!(data["input_field"], "additionalContext"); + assert_eq!(data["limit_name"], limit_name); + data +} #[tokio::test] async fn turn_steer_requires_active_turn() -> Result<()> { @@ -216,6 +272,257 @@ async fn turn_steer_rejects_oversized_text_input() -> Result<()> { Ok(()) } +#[tokio::test] +async fn turn_steer_rejects_additional_context_application_and_limits() -> Result<()> { + #[cfg(target_os = "windows")] + let shell_command = vec![ + "powershell".to_string(), + "-Command".to_string(), + "Start-Sleep -Seconds 10".to_string(), + ]; + #[cfg(not(target_os = "windows"))] + let shell_command = vec!["sleep".to_string(), "10".to_string()]; + + let tmp = TempDir::new()?; + let codex_home = tmp.path().join("codex_home"); + std::fs::create_dir(&codex_home)?; + let working_directory = tmp.path().join("workdir"); + std::fs::create_dir(&working_directory)?; + + let server = + create_mock_responses_server_sequence_unchecked(vec![create_shell_command_sse_response( + shell_command.clone(), + Some(&working_directory), + Some(10_000), + "call_sleep", + )?]) + .await; + write_mock_responses_config_toml_with_chatgpt_base_url( + &codex_home, + &server.uri(), + &server.uri(), + )?; + mount_analytics_capture(&server, &codex_home).await?; + + let mut mcp = TestAppServer::new_without_managed_config(&codex_home).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let thread_req = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let thread_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_req)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response::(thread_resp)?; + + let turn_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + client_user_message_id: None, + input: vec![V2UserInput::Text { + text: "run sleep".to_string(), + text_elements: Vec::new(), + }], + cwd: Some(working_directory.clone()), + ..Default::default() + }) + .await?; + let turn_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_req)), + ) + .await??; + let TurnStartResponse { turn } = to_response::(turn_resp)?; + + let _task_started: JSONRPCNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/started"), + ) + .await??; + + let application_err = expect_turn_steer_additional_context_error( + &mut mcp, + &thread.id, + &turn.id, + HashMap::from([( + "custom_source".to_string(), + AdditionalContextEntry { + value: "source value".to_string(), + kind: AdditionalContextKind::Application, + source: None, + }, + )]), + ) + .await?; + assert_eq!(application_err.error.code, INVALID_PARAMS_ERROR_CODE); + assert_eq!( + application_err.error.message, + "additionalContext kind `application` is reserved for trusted server-owned context and \ + cannot be supplied by turn/start or turn/steer clients" + ); + assert!( + application_err.error.data.is_none(), + "application kind rejection should not be reported as an input-too-large error" + ); + + let too_many_entries = (0..=MAX_ADDITIONAL_CONTEXT_ENTRIES) + .map(|index| { + ( + format!("source_{index}"), + untrusted_context_entry(format!("value {index}")), + ) + }) + .collect::>(); + let too_many_err = expect_turn_steer_additional_context_error( + &mut mcp, + &thread.id, + &turn.id, + too_many_entries, + ) + .await?; + let data = assert_additional_context_too_large_error( + too_many_err, + &format!( + "additionalContext must not contain more than \ + {MAX_ADDITIONAL_CONTEXT_ENTRIES} entries." + ), + "max_entries", + ); + assert_eq!(data["max_entries"], MAX_ADDITIONAL_CONTEXT_ENTRIES); + assert_eq!(data["actual_entries"], MAX_ADDITIONAL_CONTEXT_ENTRIES + 1); + + let long_key = "k".repeat(MAX_ADDITIONAL_CONTEXT_KEY_CHARS + 1); + let long_key_err = expect_turn_steer_additional_context_error( + &mut mcp, + &thread.id, + &turn.id, + HashMap::from([(long_key, untrusted_context_entry("value"))]), + ) + .await?; + let data = assert_additional_context_too_large_error( + long_key_err, + &format!( + "additionalContext keys must not exceed \ + {MAX_ADDITIONAL_CONTEXT_KEY_CHARS} characters." + ), + "max_key_chars", + ); + assert_eq!(data["max_chars"], MAX_ADDITIONAL_CONTEXT_KEY_CHARS); + assert_eq!(data["actual_chars"], MAX_ADDITIONAL_CONTEXT_KEY_CHARS + 1); + + let oversized_value = "x".repeat(MAX_ADDITIONAL_CONTEXT_VALUE_BYTES + 1); + let oversized_value_err = expect_turn_steer_additional_context_error( + &mut mcp, + &thread.id, + &turn.id, + HashMap::from([( + "source".to_string(), + untrusted_context_entry(oversized_value), + )]), + ) + .await?; + let data = assert_additional_context_too_large_error( + oversized_value_err, + &format!( + "additionalContext values must not exceed \ + {MAX_ADDITIONAL_CONTEXT_VALUE_BYTES} bytes." + ), + "max_value_bytes", + ); + assert_eq!(data["max_bytes"], MAX_ADDITIONAL_CONTEXT_VALUE_BYTES); + assert_eq!(data["actual_bytes"], MAX_ADDITIONAL_CONTEXT_VALUE_BYTES + 1); + + let aggregate_context = (0..3) + .map(|index| { + ( + format!("source_{index}"), + untrusted_context_entry("x".repeat(44 * 1024)), + ) + }) + .collect::>(); + let aggregate_err = expect_turn_steer_additional_context_error( + &mut mcp, + &thread.id, + &turn.id, + aggregate_context, + ) + .await?; + let data = assert_additional_context_too_large_error( + aggregate_err, + &format!( + "additionalContext must not exceed \ + {MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES} aggregate bytes." + ), + "max_total_bytes", + ); + assert_eq!(data["max_bytes"], MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES); + assert!( + data["actual_bytes"] + .as_u64() + .is_some_and(|actual| actual > MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES as u64) + ); + + let source_metadata_err = expect_turn_steer_additional_context_error( + &mut mcp, + &thread.id, + &turn.id, + HashMap::from([( + "source".to_string(), + AdditionalContextEntry { + value: "value".to_string(), + kind: AdditionalContextKind::Untrusted, + source: Some(AdditionalContextSource { + namespace: "open-todos".to_string(), + id: "task".to_string(), + record_type: Some("task".to_string()), + label: Some("x".repeat(MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES)), + }), + }, + )]), + ) + .await?; + let data = assert_additional_context_too_large_error( + source_metadata_err, + &format!( + "additionalContext must not exceed \ + {MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES} aggregate bytes." + ), + "max_total_bytes", + ); + assert_eq!(data["max_bytes"], MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES); + assert!( + data["actual_bytes"] + .as_u64() + .is_some_and(|actual| actual > MAX_ADDITIONAL_CONTEXT_TOTAL_BYTES as u64) + ); + + let requests = server + .received_requests() + .await + .context("failed to fetch received requests")?; + let response_requests = requests + .iter() + .filter(|request| request.url.path().ends_with("/responses")) + .collect::>(); + for request in response_requests { + let body = String::from_utf8_lossy(&request.body); + assert!( + !body.contains("steer") && !body.contains("source value"), + "rejected turn/steer additionalContext should not be submitted to the model" + ); + } + + mcp.interrupt_turn_and_wait_for_aborted(thread.id, turn.id, DEFAULT_READ_TIMEOUT) + .await?; + + Ok(()) +} + #[tokio::test] async fn turn_steer_returns_active_turn_id() -> Result<()> { #[cfg(target_os = "windows")]