agent/codex
1
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

app-server: reject direct input to multi-agent v2 sub-agents (#27173)

Browse Source 
## Why

Multi-agent v2 sub-agents are owned and coordinated by their parent
agent. Allowing an app-server client to start or steer turns on a
spawned child bypasses the multi-agent messaging path and creates a
second, conflicting source of work for that sub-agent.

## What changed

- Reject direct `turn/start` and `turn/steer` requests targeting
multi-agent v2 thread-spawn sub-agents.
- Identify these targets using both the thread's resolved multi-agent
version and its `SubAgentSource::ThreadSpawn` session source, leaving
root threads, v1 agents, and other sub-agent types unchanged.
- Return a consistent invalid-request error before validating or
applying the submitted input.

## Testing

- Added an app-server integration test that spawns a real multi-agent v2
child and verifies that direct `turn/start` and `turn/steer` requests
are rejected.
This commit is contained in:
jif
2026-06-09 19:40:40 +02:00
committed by GitHub
Unverified
parent 8a299fb704
commit 1026e9de1b
2 changed files with 166 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors/turn_processor.rs
+36 -6
View File
@@ -1,6 +1,12 @@
use super::*;
use codex_protocol::protocol::AdditionalContextEntry as CoreAdditionalContextEntry;
use codex_protocol::protocol::AdditionalContextKind as CoreAdditionalContextKind;
use codex_protocol::protocol::MultiAgentVersion;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
const DIRECT_INPUT_TO_MULTI_AGENT_V2_SUBAGENT_ERROR: &str =
"direct app-server input is not allowed for multi-agent v2 sub-agents";
#[derive(Clone)]
pub(crate) struct TurnRequestProcessor {
@@ -236,6 +242,26 @@ impl TurnRequestProcessor {
Ok((thread_id, thread))
}
async fn ensure_direct_input_allowed(
&self,
request_id: &ConnectionRequestId,
thread: &CodexThread,
) -> Result<(), JSONRPCErrorError> {
if thread.multi_agent_version() == Some(MultiAgentVersion::V2)
&& matches!(
thread.config_snapshot().await.session_source,
SessionSource::SubAgent(SubAgentSource::ThreadSpawn { .. })
)
{
let error = invalid_request(DIRECT_INPUT_TO_MULTI_AGENT_V2_SUBAGENT_ERROR);
self.track_error_response(request_id, &error, /*error_type*/ None);
return Err(error);
}
Ok(())
}
fn normalize_collaboration_mode(
&self,
mut collaboration_mode: CollaborationMode,
@@ -370,6 +396,14 @@ impl TurnRequestProcessor {
app_server_client_name: Option<String>,
app_server_client_version: Option<String>,
) -> Result<TurnStartResponse, JSONRPCErrorError> {
let (thread_id, thread) =
self.load_thread(&params.thread_id)
.await
.inspect_err(|error| {
self.track_error_response(&request_id, error, /*error_type*/ None);
})?;
self.ensure_direct_input_allowed(&request_id, thread.as_ref())
.await?;
if let Err(error) = Self::validate_v2_input_limit(&params.input) {
self.track_error_response(
&request_id,
@@ -378,12 +412,6 @@ impl TurnRequestProcessor {
);
return Err(error);
}
let (thread_id, thread) =
self.load_thread(&params.thread_id)
.await
.inspect_err(|error| {
self.track_error_response(&request_id, error, /*error_type*/ None);
})?;
Self::set_app_server_client_info(
thread.as_ref(),
app_server_client_name,
@@ -762,6 +790,8 @@ impl TurnRequestProcessor {
.inspect_err(|error| {
self.track_error_response(request_id, error, /*error_type*/ None);
})?;
self.ensure_direct_input_allowed(request_id, thread.as_ref())
.await?;
if params.expected_turn_id.is_empty() {
return Err(invalid_request("expectedTurnId must not be empty"));
codex-rs/app-server/tests/suite/v2/turn_start.rs
+130
View File
@@ -41,6 +41,7 @@ use codex_app_server_protocol::PatchChangeKind;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::ServerRequestResolvedNotification;
use codex_app_server_protocol::SubAgentActivityKind;
use codex_app_server_protocol::TextElement;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadSource;
@@ -53,6 +54,7 @@ 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::TurnSteerParams;
use codex_app_server_protocol::UserInput as V2UserInput;
use codex_app_server_protocol::WarningNotification;
use codex_config::config_toml::ConfigToml;
@@ -3363,6 +3365,134 @@ async fn turn_start_emits_spawn_agent_item_with_model_metadata_v2() -> Result<()
Ok(())
}
#[tokio::test]
async fn direct_input_to_multi_agent_v2_subagent_is_rejected() -> Result<()> {
const CHILD_PROMPT: &str = "child: do work";
const PARENT_PROMPT: &str = "spawn a child and continue";
const SPAWN_CALL_ID: &str = "spawn-call-direct-input-rejection";
const ERROR_MESSAGE: &str =
"direct app-server input is not allowed for multi-agent v2 sub-agents";
let server = responses::start_mock_server().await;
let spawn_args = serde_json::to_string(&json!({
"message": CHILD_PROMPT,
"task_name": "worker",
}))?;
let _parent_turn = responses::mount_sse_once_match(
&server,
|req: &wiremock::Request| body_contains(req, PARENT_PROMPT),
responses::sse(vec![
responses::ev_response_created("resp-parent-1"),
responses::ev_function_call(SPAWN_CALL_ID, "spawn_agent", &spawn_args),
responses::ev_completed("resp-parent-1"),
]),
)
.await;
let codex_home = TempDir::new()?;
create_config_toml(
codex_home.path(),
&server.uri(),
"never",
&BTreeMap::from([(Feature::MultiAgentV2, true)]),
)?;
write_models_cache(codex_home.path())?;
let mut mcp = TestAppServer::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let thread_req = mcp
.send_thread_start_request(ThreadStartParams {
model: Some("gpt-5.3-codex".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::<ThreadStartResponse>(thread_resp)?;
let turn_req = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id,
input: vec![V2UserInput::Text {
text: PARENT_PROMPT.to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let turn_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(turn_req)),
)
.await??;
let _: TurnStartResponse = to_response(turn_resp)?;
let child_thread_id = timeout(DEFAULT_READ_TIMEOUT, async {
loop {
let completed_notif = mcp
.read_stream_until_notification_message("item/completed")
.await?;
let completed: ItemCompletedNotification =
serde_json::from_value(completed_notif.params.expect("item/completed params"))?;
if let ThreadItem::SubAgentActivity {
id,
kind: SubAgentActivityKind::Started,
agent_thread_id,
..
} = completed.item
&& id == SPAWN_CALL_ID
{
return Ok::<String, anyhow::Error>(agent_thread_id);
}
}
})
.await??;
let direct_turn_req = mcp
.send_turn_start_request(TurnStartParams {
thread_id: child_thread_id.clone(),
input: vec![V2UserInput::Text {
text: "direct app-server turn".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let direct_turn_error: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(direct_turn_req)),
)
.await??;
assert_eq!(direct_turn_error.error.code, INVALID_REQUEST_ERROR_CODE);
assert_eq!(direct_turn_error.error.message, ERROR_MESSAGE);
let direct_steer_req = mcp
.send_turn_steer_request(TurnSteerParams {
thread_id: child_thread_id,
client_user_message_id: None,
input: vec![V2UserInput::Text {
text: "direct app-server steer".to_string(),
text_elements: Vec::new(),
}],
responsesapi_client_metadata: None,
additional_context: None,
expected_turn_id: "any-active-turn".to_string(),
})
.await?;
let direct_steer_error: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(direct_steer_req)),
)
.await??;
assert_eq!(direct_steer_error.error.code, INVALID_REQUEST_ERROR_CODE);
assert_eq!(direct_steer_error.error.message, ERROR_MESSAGE);
Ok(())
}
#[tokio::test]
async fn turn_start_emits_spawn_agent_item_with_effective_role_model_metadata_v2() -> Result<()> {
skip_if_no_network!(Ok(()));