openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-05 20:10:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Accept agent follow-ups through native turn input delivery (#58829)

Browse Source 
## Why

`start_or_steer_turn` rejected agent messages, preventing callers from submitting follow-up work through the turn input API.

## What changed

Route `TurnInput::InterAgentCommunication` through native agent message delivery and return `TurnInputSubmission::Accepted`, without implying an immediate turn ID. These submissions request work regardless of `trigger_turn`, retain the recipient's settings, and forward turn-start options.

Wrap plaintext messages with agent context while preserving encrypted content. Reject submissions to idle recipients when the host is draining.

## Testing

Add scenario tests and snapshots covering queued-message ordering, encrypted follow-ups, preemption at reasoning boundaries, and follow-ups arriving before or after a final answer. Extend drain-gate coverage to agent submissions with parent-turn lineage.

GitOrigin-RevId: 92d49465f97bb366babbf1b1bc18a1e8a6c6ed1c
This commit is contained in:
Won Park
2026-11-02 07:27:49 +00:00
committed by copyberry
parent a09812981d
commit 05c08d7655
12 changed files with 541 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors/daemon_continuation.rs
+5 -1
View File
@@ -130,7 +130,11 @@ impl ThreadRequestProcessor {
Ok(TurnInputSubmission::NotSubmitted { reason }) => {
tracing::debug!(%thread_id, ?reason, "recovery continuation was not started");
}
Ok(TurnInputSubmission::Steered { .. }) => unreachable!("continuation cannot steer"),
Ok(TurnInputSubmission::Steered { .. } | TurnInputSubmission::Accepted) => {
// continue_turn_if_idle uses start_if_idle, which cannot steer or
// enter the agent-message path that returns Accepted.
unreachable!("continuation must start or decline")
}
Err(err) => tracing::warn!(%thread_id, %err, "recovery continuation failed"),
}
}
codex-rs/app-server/src/request_processors/turn_processor.rs
+5
View File
@@ -715,6 +715,11 @@ impl TurnRequestProcessor {
error
})?;
let (turn_id, root_turn_id, started) = match submission {
TurnInputSubmission::Accepted => {
return Err(internal_error(
"user input unexpectedly returned agent acceptance".to_string(),
));
}
TurnInputSubmission::Started {
turn_id,
root_turn_id,
codex-rs/core/src/agent/control.rs
+1
View File
@@ -148,6 +148,7 @@ impl LocalAgentControl {
.await
{
Ok(TurnInputSubmission::Started { turn_id, .. }) => Ok(turn_id),
Ok(TurnInputSubmission::Accepted) => unreachable!("user input returns a turn outcome"),
Ok(TurnInputSubmission::Steered { .. }) => {
// MAv1 exposes an opaque `submission_id` to the model. The legacy
// `Op::UserInput` path returned a fresh ID for every steer, while the
codex-rs/core/src/codex_thread.rs
+16 -9
View File
@@ -477,9 +477,16 @@ impl CodexThread {
/// Submits turn input without requiring the caller to inspect thread state.
///
/// The result describes whether Core started a turn, steered an active
/// turn, or declined it without recording or enqueueing the input.
/// User input and named standalone function-call outputs are accepted.
/// User input reports whether Core started or steered a turn. Agent messages
/// report acceptance into native delivery, which can wait for a later turn.
/// Declined input is neither recorded nor enqueued.
/// User input, agent messages, and named standalone function-call outputs are accepted.
/// Agent messages contain unformatted plaintext or encrypted content. Core adds
/// the agent-message context wrapper; callers must not add it themselves.
/// Agent messages request work here regardless of their `trigger_turn` flag.
/// Agent messages retain the recipient's settings. Only the message and `start`
/// options are passed to native delivery; `thread_settings`, `turn_extension_init`,
/// `additional_context`, and `responsesapi_client_metadata` are ignored.
pub async fn start_or_steer_turn(
&self,
request: impl Into<WithTurnExtensionData<TurnInputRequest>>,
@@ -510,8 +517,8 @@ impl CodexThread {
TurnInputSubmission::NotSubmitted { reason } => {
Ok(StartIfIdleSubmission::NotSubmitted { reason })
}
TurnInputSubmission::Steered { .. } => {
unreachable!("start-if-idle submission cannot steer")
TurnInputSubmission::Steered { .. } | TurnInputSubmission::Accepted => {
unreachable!("start-if-idle submission must start or decline")
}
}
}
@@ -578,8 +585,8 @@ impl CodexThread {
TurnInputSubmission::NotSubmitted { reason } => {
Ok(StartIfIdleSubmission::NotSubmitted { reason })
}
TurnInputSubmission::Steered { .. } => {
unreachable!("recovered turn submission cannot steer")
TurnInputSubmission::Steered { .. } | TurnInputSubmission::Accepted => {
unreachable!("recovered turn submission must start or decline")
}
}
}
@@ -645,8 +652,8 @@ impl CodexThread {
TurnInputSubmission::NotSubmitted { reason } => {
Ok(SteerSubmission::NotSubmitted { reason })
}
TurnInputSubmission::Started { .. } => {
unreachable!("steer-only submission cannot start a turn")
TurnInputSubmission::Started { .. } | TurnInputSubmission::Accepted => {
unreachable!("steer-only submission must steer or decline")
}
}
}
codex-rs/core/src/session/turn_input.rs
+27 -7
View File
@@ -60,6 +60,9 @@ use std::collections::BTreeMap;
use std::sync::Arc;
use uuid::Uuid;
#[path = "turn_input_agent.rs"]
mod agent;
#[cfg(test)]
#[path = "turn_input_tests.rs"]
mod tests;
@@ -244,12 +247,20 @@ pub(super) async fn handle(
{
validate_injected_response_item(item)?;
}
request
.request
.additional_context
.values()
.try_for_each(AdditionalContextEntry::validate)
.map_err(|message| CodexErr::InvalidRequest(message.to_owned()))?;
// Native agent follow-ups do not use the request's additional context.
if !(matches!(mode, TurnInputMode::StartOrSteer)
&& matches!(
request.request.input,
SubmittedTurnInput::InterAgentCommunication(_)
))
{
request
.request
.additional_context
.values()
.try_for_each(AdditionalContextEntry::validate)
.map_err(|message| CodexErr::InvalidRequest(message.to_owned()))?;
}
let result = match mode {
TurnInputMode::StartOrSteer => start_or_steer(session, request, submission_id).await,
TurnInputMode::StartIfIdle => {
@@ -354,6 +365,12 @@ async fn start_or_steer(
request: WithTurnExtensionData<TurnInputRequest>,
submission_id: String,
) -> CodexResult<TurnInputSubmission> {
if matches!(
request.request.input,
SubmittedTurnInput::InterAgentCommunication(_)
) {
return agent::submit(session, request, submission_id).await;
}
let WithTurnExtensionData {
request,
turn_extension_init,
@@ -379,7 +396,7 @@ async fn start_or_steer(
}) => true,
_ => {
return Err(CodexErr::InvalidRequest(
"only user input or standalone function-call outputs can start or steer a turn"
"only user input, agent messages, or standalone function-call outputs can start or steer a turn"
.to_string(),
));
}
@@ -816,6 +833,9 @@ impl Session {
)
.await;
match submission {
Ok(TurnInputSubmission::Accepted) => {
unreachable!("realtime user input returns a turn outcome")
}
Ok(TurnInputSubmission::Started { .. } | TurnInputSubmission::Steered { .. }) => {}
Ok(TurnInputSubmission::NotSubmitted {
reason: NotSubmittedReason::ServerDraining,
codex-rs/core/src/session/turn_input_agent.rs
+61
View File
@@ -0,0 +1,61 @@
//! Admits reply-bearing agent follow-ups through native message delivery.
//! Acceptance means queued by Core; the recipient can consume the input in a later turn.
//! Follow-ups retain the recipient's settings and pass turn-start metadata to native delivery.
use crate::PendingAgentMessage;
use crate::WithTurnExtensionData;
use crate::agent::types::AgentMessage;
use crate::agent::types::MessageDeliveryMode;
use crate::session::handlers;
use crate::session::session::Session;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::turn_input::NotSubmittedReason;
use codex_protocol::turn_input::TurnInput;
use codex_protocol::turn_input::TurnInputRequest;
use codex_protocol::turn_input::TurnInputSubmission;
use std::sync::Arc;
pub(super) async fn submit(
session: &Arc<Session>,
request: WithTurnExtensionData<TurnInputRequest>,
submission_id: String,
) -> CodexResult<TurnInputSubmission> {
let WithTurnExtensionData { request, .. } = request;
let TurnInputRequest { input, start, .. } = request;
let TurnInput::InterAgentCommunication(mut communication) = input else {
unreachable!("agent submission requires an agent message");
};
// CodexThread already checks local agent capacity before submitting this operation.
// Preserve the host's drain gate when accepting work for an idle recipient.
let _admission = if session.active_turn.lock().await.is_none() {
let Some(admission) = session.services.extensions.admit_turn_start() else {
return Ok(TurnInputSubmission::NotSubmitted {
reason: NotSubmittedReason::ServerDraining,
});
};
Some(admission)
} else {
None
};
if communication.encrypted_content.is_none() {
communication.content = AgentMessage::Plaintext(std::mem::take(&mut communication.content))
.into_communication(
communication.author.clone(),
communication.recipient.clone(),
MessageDeliveryMode::TriggerTurn,
)
.content;
}
// Calling start-or-steer requests work.
communication.trigger_turn = true;
let pending = PendingAgentMessage::from(communication);
// Bound the future layout depth of this additional native-delivery path.
Box::pin(handlers::inter_agent_communication(
session,
submission_id,
pending,
start,
))
.await;
Ok(TurnInputSubmission::Accepted)
}
codex-rs/core/tests/suite/scenarios.rs
+3
View File
@@ -1,5 +1,8 @@
//! Multi-turn Astra scenarios snapshot the model-visible request history of shipped features.
#[path = "scenarios_turn_input_agent_message.rs"]
mod turn_input_agent_message;
use super::image_rollout::RecordingFileAttachmentStore;
use super::permissions_messages::PermissionPathScenario;
use codex_attachment_store::AttachmentStoreErrorKind;
codex-rs/core/tests/suite/scenarios_turn_input_agent_message.rs
+308
View File
@@ -0,0 +1,308 @@
//! Follow-ups use native agent-mail sampling and final-answer boundaries.
use codex_core::StartThreadOptions;
use codex_core::TurnInput;
use codex_core::TurnInputRequest;
use codex_core::TurnInputSubmission;
use codex_core::TurnStartOptions;
use codex_features::Feature;
use codex_protocol::AgentPath;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use core_test_support::context_snapshot;
use core_test_support::context_snapshot::ContextSnapshotOptions;
use core_test_support::context_snapshot::SnapshotEntry;
use core_test_support::responses;
use core_test_support::streaming_sse::StreamingSseChunk;
use core_test_support::streaming_sse::start_streaming_sse_server;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use test_case::test_case;
use tokio::sync::oneshot;
#[derive(Clone, Copy, Debug)]
enum FollowupArrival {
BeforeFinalAnswer,
AfterFinalAnswer,
}
#[test_case(FollowupArrival::BeforeFinalAnswer; "before_final_answer")]
#[test_case(FollowupArrival::AfterFinalAnswer; "after_final_answer")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn agent_followup_preserves_native_boundaries(
arrival: FollowupArrival,
) -> anyhow::Result<()> {
let (release_reasoning, reasoning_gate) = oneshot::channel();
let (release_final, final_gate) = oneshot::channel();
let (release_completion, completion_gate) = oneshot::channel();
let mut final_answer = responses::ev_assistant_message("final", "The first task is finished.");
final_answer["item"]["phase"] = json!("final_answer");
let mut final_added = responses::ev_message_item_added("final", "");
final_added["item"]["phase"] = json!("final_answer");
let (streaming, _) = start_streaming_sse_server(vec![
vec![
StreamingSseChunk {
gate: None,
body: responses::sse(vec![
responses::ev_response_created("first"),
responses::ev_reasoning_item_added("reasoning", &["Checking the task"]),
]),
},
StreamingSseChunk {
gate: Some(reasoning_gate),
body: responses::sse(vec![
responses::ev_reasoning_item("reasoning", &["Checking the task"], &[]),
responses::ev_function_call(
"must-not-run",
"update_plan",
r#"{"plan":[{"step":"This should be preempted","status":"in_progress"}]}"#,
),
responses::ev_completed("first"),
]),
},
],
vec![
StreamingSseChunk {
gate: None,
body: responses::sse(vec![responses::ev_response_created("second"), final_added]),
},
StreamingSseChunk {
gate: Some(final_gate),
body: responses::sse(vec![final_answer]),
},
StreamingSseChunk {
gate: Some(completion_gate),
body: responses::sse(vec![responses::ev_completed("second")]),
},
],
vec![StreamingSseChunk {
gate: None,
body: responses::sse(vec![
responses::ev_response_created("third"),
responses::ev_assistant_message("next-final", "The next task is finished."),
responses::ev_completed("third"),
]),
}],
])
.await;
let config_server = responses::start_mock_server().await;
let base_url = format!("{}/v1", streaming.uri());
let test = test_codex()
.with_model("gpt-5.4")
.with_config(move |config| {
config.model_provider.base_url = Some(base_url);
config.update_plan_enabled = true;
config
.features
.enable(Feature::MultiAgentV2)
.expect("enable native multi-agent delivery");
config
.features
.disable(Feature::WakeOnAgentMessages)
.expect("disable queue-only message wakeups");
config
.features
.disable(Feature::DeferMailboxPreemption)
.expect("enable immediate mailbox preemption");
config
.features
.disable(Feature::EnableRequestCompression)
.expect("disable request compression for the mock server");
})
.build_with_auto_env(&config_server)
.await?;
let child_path = AgentPath::try_from("/root/worker").expect("child path");
let child = test
.thread_manager
.start_thread(StartThreadOptions {
session_source: Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: test.session_configured.thread_id,
depth: 1,
agent_path: Some(child_path.clone()),
agent_nickname: None,
agent_role: None,
})),
..test.start_thread_options().await
})
.await?
.thread;
// An idle recipient must see its older queue-only mail before the new task.
child
.submit(Op::InterAgentCommunication {
communication: InterAgentCommunication::new(
AgentPath::root(),
child_path.clone(),
Vec::new(),
"Older queued mail".to_owned(),
/*trigger_turn*/ false,
),
start_options: Default::default(),
})
.await?;
child.submit(Op::RealtimeConversationListVoices).await?;
wait_for_event(&child, |event| {
matches!(event, EventMsg::RealtimeConversationListVoicesResponse(_))
})
.await;
let message = |text: &str| {
TurnInputRequest::new(TurnInput::InterAgentCommunication(
InterAgentCommunication::new(
AgentPath::root(),
child_path.clone(),
Vec::new(),
text.to_owned(),
/*trigger_turn*/ false,
),
))
.on_start(TurnStartOptions {
parent_turn_id: Some("parent-turn".to_owned()),
root_turn_id: Some("root-turn".to_owned()),
initiating_agent_path: Some(AgentPath::root()),
..Default::default()
})
};
assert_eq!(
child
.start_or_steer_turn(message("Find the failing test"))
.await?,
TurnInputSubmission::Accepted
);
wait_for_event(
&child,
|event| matches!(event, EventMsg::ItemStarted(item) if item.item.id() == "reasoning"),
)
.await;
assert_eq!(
child
.start_or_steer_turn(TurnInputRequest::new(TurnInput::InterAgentCommunication(
InterAgentCommunication::new_encrypted(
AgentPath::root(),
child_path.clone(),
Vec::new(),
"opaque-agent-message".to_owned(),
/*trigger_turn*/ false,
),
)))
.await?,
TurnInputSubmission::Accepted
);
release_reasoning.send(()).expect("release reasoning");
// The queued follow-up must preempt the remaining tool call at this reasoning boundary.
wait_for_event(&child, |event| {
assert!(
!matches!(
event,
EventMsg::PlanUpdate(_) | EventMsg::TurnComplete(_) | EventMsg::Error(_)
),
"{event:?}"
);
matches!(event, EventMsg::ItemStarted(item) if item.item.id() == "final")
})
.await;
match arrival {
FollowupArrival::BeforeFinalAnswer => {
assert_eq!(
child
.start_or_steer_turn(message("Now check the fix"))
.await?,
TurnInputSubmission::Accepted
);
release_final.send(()).expect("release final answer");
}
FollowupArrival::AfterFinalAnswer => {
release_final.send(()).expect("release final answer");
wait_for_event(
&child,
|event| matches!(event, EventMsg::ItemCompleted(item) if item.item.id() == "final"),
)
.await;
assert_eq!(
child
.start_or_steer_turn(message("Now check the fix"))
.await?,
TurnInputSubmission::Accepted
);
}
}
release_completion.send(()).expect("release completion");
// Neither arrival may keep the first turn open. The follow-up gets a new turn.
let mut completed_turns = Vec::new();
wait_for_event(&child, |event| {
assert!(
!matches!(event, EventMsg::PlanUpdate(_) | EventMsg::Error(_)),
"{event:?}"
);
if let EventMsg::TurnComplete(done) = event {
completed_turns.push(done.turn_id.clone());
}
completed_turns.len() == 2
})
.await;
assert_ne!(completed_turns[0], completed_turns[1]);
let requests = streaming
.requests()
.await
.iter()
.map(|body| serde_json::from_slice::<Value>(body))
.collect::<serde_json::Result<Vec<_>>>()?;
assert_eq!(requests.len(), 3);
let first_mail = requests[0]["input"]
.as_array()
.expect("input array")
.iter()
.filter(|item| item["type"] == "agent_message")
.map(|item| {
item["content"][0]["text"]
.as_str()
.expect("plaintext message")
})
.collect::<Vec<_>>();
assert_eq!(
first_mail,
vec![
"Older queued mail",
"Message Type: NEW_TASK\nTask name: /root/worker\nSender: /root\nPayload:\nFind the failing test",
]
);
let later_mail = requests[1..]
.iter()
.map(|request| {
request["input"]
.as_array()
.expect("input array")
.iter()
.rev()
.find(|item| item["type"] == "agent_message")
.expect("follow-up message")["content"]
.clone()
})
.collect::<Vec<_>>();
assert_eq!(
later_mail,
vec![
json!([
{"type": "input_text", "text": "Message Type: NEW_TASK\nTask name: /root/worker\nSender: /root\nPayload:\n"},
{"type": "encrypted_content", "encrypted_content": "opaque-agent-message"},
]),
json!([{"type": "input_text", "text": "Message Type: NEW_TASK\nTask name: /root/worker\nSender: /root\nPayload:\nNow check the fix"}]),
]
);
let entries = requests.iter().map(SnapshotEntry::body).collect::<Vec<_>>();
insta::assert_snapshot!(
format!("agent_followup_{arrival:?}"),
context_snapshot::format_context_snapshot(
"Agent follow-ups interrupt at a reasoning boundary, preserve encrypted input, and start a new turn when not consumed before the final answer completes.",
&entries,
&ContextSnapshotOptions::default().rewrite_known_segments(),
)
);
child.shutdown_and_wait().await?;
streaming.shutdown().await;
Ok(())
}
codex-rs/core/tests/suite/snapshots/all__suite__scenarios__turn_input_agent_message__agent_followup_AfterFinalAnswer.snap
+44
View File
@@ -0,0 +1,44 @@
---
source: core/tests/suite/scenarios_turn_input_agent_message.rs
assertion_line: 198
expression: "context_snapshot::format_context_snapshot(\"Agent follow-ups interrupt at a reasoning boundary, preserve encrypted input, and start a new turn when not consumed before the final answer completes.\",\n&entries, &ContextSnapshotOptions::default().rewrite_known_segments(),)"
---
Scenario: Agent follow-ups interrupt at a reasoning boundary, preserve encrypted input, and start a new turn when not consumed before the final answer completes.
## Window 1
-- request 1 (request) --
00:message/developer:
<MODEL_INSTRUCTIONS>
01:message/developer:
<PERMISSIONS_INSTRUCTIONS>
02:message/developer:
You are an agent in a team of agents collaborating to complete a task.
You can spawn sub-agents to handle subtasks, and those sub-agents can spawn their own sub-agents. All ag...nd capable, and have access to the same set of tools. [hash=876a93c54d3f61d6]
You can use `spawn_agent` to create a new agent, `followup_task` to give an existing agent a new task an... `send_message` to pass a message to a running agent. [hash=824c12425c7626dc]
Child agents can also spawn their own sub-agents.
When you provide a response in the final channel, that content is immediately delivered back to your parent agent.
<OMITTED 15 LINES; ~225 TOKENS; hash=fa366829bdb214be>
- All agents use the same current working directory.
- As a result, edits made by one agent are immediately visible to all other agents.
When calling `wait_agent`, prefer longer waits (minutes) to avoid busy polling.
There are 4 available concurrency slots, meaning that up to 4 agents can be active at once, including you.
Full-history forks (`fork_turns` omitted or `"all"`) inherit the parent model and reasoning effort and d...uctions; when doing so, set `fork_turns` to `"none"`. [hash=18ccf7e93ecabab9]
03:message/developer:
<MULTI_AGENT_MODE>
04:message/user:
<ENVIRONMENT_CONTEXT>
05:agent_message
06:agent_message
-- request 2 (request) --
07:reasoning:summary=Checking the task:encrypted=true
08:agent_message
-- request 3 (request) --
09:message/assistant:
The first task is finished.
10:agent_message
codex-rs/core/tests/suite/snapshots/all__suite__scenarios__turn_input_agent_message__agent_followup_BeforeFinalAnswer.snap
+44
View File
@@ -0,0 +1,44 @@
---
source: core/tests/suite/scenarios_turn_input_agent_message.rs
assertion_line: 198
expression: "context_snapshot::format_context_snapshot(\"Agent follow-ups interrupt at a reasoning boundary, preserve encrypted input, and start a new turn when not consumed before the final answer completes.\",\n&entries, &ContextSnapshotOptions::default().rewrite_known_segments(),)"
---
Scenario: Agent follow-ups interrupt at a reasoning boundary, preserve encrypted input, and start a new turn when not consumed before the final answer completes.
## Window 1
-- request 1 (request) --
00:message/developer:
<MODEL_INSTRUCTIONS>
01:message/developer:
<PERMISSIONS_INSTRUCTIONS>
02:message/developer:
You are an agent in a team of agents collaborating to complete a task.
You can spawn sub-agents to handle subtasks, and those sub-agents can spawn their own sub-agents. All ag...nd capable, and have access to the same set of tools. [hash=876a93c54d3f61d6]
You can use `spawn_agent` to create a new agent, `followup_task` to give an existing agent a new task an... `send_message` to pass a message to a running agent. [hash=824c12425c7626dc]
Child agents can also spawn their own sub-agents.
When you provide a response in the final channel, that content is immediately delivered back to your parent agent.
<OMITTED 15 LINES; ~225 TOKENS; hash=fa366829bdb214be>
- All agents use the same current working directory.
- As a result, edits made by one agent are immediately visible to all other agents.
When calling `wait_agent`, prefer longer waits (minutes) to avoid busy polling.
There are 4 available concurrency slots, meaning that up to 4 agents can be active at once, including you.
Full-history forks (`fork_turns` omitted or `"all"`) inherit the parent model and reasoning effort and d...uctions; when doing so, set `fork_turns` to `"none"`. [hash=18ccf7e93ecabab9]
03:message/developer:
<MULTI_AGENT_MODE>
04:message/user:
<ENVIRONMENT_CONTEXT>
05:agent_message
06:agent_message
-- request 2 (request) --
07:reasoning:summary=Checking the task:encrypted=true
08:agent_message
-- request 3 (request) --
09:message/assistant:
The first task is finished.
10:agent_message
codex-rs/core/tests/suite/turn_input_submission.rs
+24
View File
@@ -311,6 +311,30 @@ async fn host_drain_allows_spawned_agent_input_but_not_automatic_work() -> anyho
reason: NotSubmittedReason::ServerDraining
},
);
// A remote agent submission has not already been admitted by this host,
// even when its caller supplies parent lineage.
assert_eq!(
child
.start_or_steer_turn(
TurnInputRequest::new(TurnInput::InterAgentCommunication(
InterAgentCommunication::new(
AgentPath::root(),
AgentPath::try_from("/root/worker").expect("valid agent path"),
Vec::new(),
"agent input while draining".to_string(),
/*trigger_turn*/ true,
),
))
.on_start(TurnStartOptions {
parent_turn_id: Some("parent-turn".to_string()),
..Default::default()
}),
)
.await?,
TurnInputSubmission::NotSubmitted {
reason: NotSubmittedReason::ServerDraining
},
);
assert!(matches!(
child
.start_or_steer_turn(user_message_request("delegated input").on_start(
codex-rs/protocol/src/turn_input.rs
+3
View File
@@ -247,6 +247,9 @@ pub struct TurnStartOptions {
/// resolved causal root, including when steering preserves an existing turn.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum TurnInputSubmission {
/// Core accepted an agent follow-up through native message delivery. The
/// receiving turn may still be running or finishing, so no turn ID is implied.
Accepted,
/// Core started a turn. Persistent thread settings and start options were applied.
Started {
turn_id: String,