agent/codex
2
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

[codex] Store pending response items directly (#24865)

Browse Source
This commit is contained in:
pakrym-oai
2026-05-28 07:23:08 -07:00
committed by GitHub
Unverified
parent e7d156eb08
commit 1c7832ffa3
21 changed files with 183 additions and 460 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control.rs
-16
View File
@@ -704,22 +704,6 @@ impl AgentControl {
result
}
/// Append a prebuilt message to an existing agent thread outside the normal user-input path.
#[cfg(test)]
pub(crate) async fn append_message(
&self,
agent_id: ThreadId,
message: ResponseItem,
) -> CodexResult<String> {
let state = self.upgrade()?;
self.handle_thread_request_result(
agent_id,
&state,
state.append_message(agent_id, message).await,
)
.await
}
pub(crate) async fn send_inter_agent_communication(
&self,
agent_id: ThreadId,
codex-rs/core/src/agent/control_tests.rs
-54
View File
@@ -518,60 +518,6 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig
));
}
#[tokio::test]
async fn append_message_records_assistant_message() {
let harness = AgentControlHarness::new().await;
let (thread_id, thread) = harness.start_thread().await;
let message =
"author: /root\nrecipient: /root/worker\nother_recipients: []\nContent: hello from tests";
let submission_id = harness
.control
.append_message(
thread_id,
ResponseItem::Message {
id: None,
role: "assistant".to_string(),
content: vec![ContentItem::InputText {
text: message.to_string(),
}],
phase: None,
},
)
.await
.expect("append_message should succeed");
assert!(!submission_id.is_empty());
timeout(Duration::from_secs(5), async {
loop {
let history_items = thread
.codex
.session
.clone_history()
.await
.raw_items()
.to_vec();
let recorded = history_items.iter().any(|item| {
matches!(
item,
ResponseItem::Message { role, content, .. }
if role == "assistant"
&& content.iter().any(|content_item| matches!(
content_item,
ContentItem::InputText { text } if text == message
))
)
});
if recorded {
break;
}
sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("assistant message should be recorded");
}
#[tokio::test]
async fn spawn_agent_creates_thread_and_sends_prompt() {
let harness = AgentControlHarness::new().await;
codex-rs/core/src/codex_thread.rs
+11 -66
View File
@@ -260,7 +260,12 @@ impl CodexThread {
&self,
items: Vec<ResponseInputItem>,
) -> Result<(), Vec<ResponseInputItem>> {
self.codex.session.inject_response_items(items).await
let response_items = items.iter().cloned().map(ResponseItem::from).collect();
self.codex
.session
.inject_if_running(response_items)
.await
.map_err(|_| items)
}
pub async fn set_app_server_client_info(
@@ -366,58 +371,16 @@ impl CodexThread {
/// Records a user-role session-prefix message without creating a new user turn boundary.
pub(crate) async fn inject_user_message_without_turn(&self, message: String) {
let message = ResponseItem::Message {
let item = ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText { text: message }],
phase: None,
};
let pending_item = match pending_message_input_item(&message) {
Ok(pending_item) => pending_item,
Err(err) => {
debug_assert!(false, "session-prefix message append should succeed: {err}");
return;
}
};
if self
.codex
self.codex
.session
.inject_response_items(vec![pending_item])
.await
.is_err()
{
let turn_context = self.codex.session.new_default_turn().await;
self.codex
.session
.record_conversation_items(turn_context.as_ref(), &[message])
.await;
}
}
/// Append a prebuilt message to the thread history without treating it as a user turn.
///
/// If the thread already has an active turn, the message is queued as pending input for that
/// turn. Otherwise it is queued at session scope and a regular turn is started so the agent
/// can consume that pending input through the normal turn pipeline.
#[cfg(test)]
pub(crate) async fn append_message(&self, message: ResponseItem) -> CodexResult<String> {
let submission_id = uuid::Uuid::new_v4().to_string();
let pending_item = pending_message_input_item(&message)?;
if let Err(items) = self
.codex
.session
.inject_response_items(vec![pending_item])
.await
{
self.codex
.session
.input_queue
.queue_response_items_for_next_turn(items)
.await;
self.codex.session.maybe_start_turn_for_pending_work().await;
}
Ok(submission_id)
.inject_no_new_turn(vec![item], /*current_turn_context*/ None)
.await;
}
/// Append raw Responses API items to the thread's model-visible history.
@@ -437,7 +400,7 @@ impl CodexThread {
}
self.codex
.session
.record_conversation_items(turn_context.as_ref(), &items)
.inject_no_new_turn(items, Some(turn_context.as_ref()))
.await;
self.codex.session.flush_rollout().await?;
Ok(())
@@ -598,21 +561,3 @@ impl CodexThread {
Ok(*guard)
}
}
fn pending_message_input_item(message: &ResponseItem) -> CodexResult<ResponseInputItem> {
match message {
ResponseItem::Message {
role,
content,
phase,
..
} => Ok(ResponseInputItem::Message {
role: role.clone(),
content: content.clone(),
phase: phase.clone(),
}),
_ => Err(CodexErr::InvalidRequest(
"append_message only supports ResponseItem::Message".to_string(),
)),
}
}
codex-rs/core/src/compact.rs
+1 -1
View File
@@ -568,7 +568,7 @@ async fn drain_to_completed(
};
match event {
Ok(ResponseEvent::OutputItemDone(item)) => {
sess.record_into_history(std::slice::from_ref(&item), turn_context)
sess.record_conversation_items(turn_context, std::slice::from_ref(&item))
.await;
}
Ok(ResponseEvent::ServerReasoningIncluded(included)) => {
codex-rs/core/src/goals.rs
+11 -22
View File
@@ -26,7 +26,7 @@ use codex_otel::GOAL_TOKEN_COUNT_METRIC;
use codex_otel::GOAL_USAGE_LIMITED_METRIC;
use codex_protocol::ThreadId;
use codex_protocol::config_types::ModeKind;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ThreadGoal;
use codex_protocol::protocol::ThreadGoalStatus;
@@ -177,7 +177,7 @@ pub(crate) struct GoalRuntimeState {
struct GoalContinuationCandidate {
goal_id: String,
items: Vec<ResponseInputItem>,
items: Vec<ResponseItem>,
}
impl GoalRuntimeState {
@@ -686,7 +686,7 @@ impl Session {
.await;
if let Some(goal) = goal_for_steering {
let item = goal_context_input_item(objective_updated_prompt(&goal));
if self.inject_response_items(vec![item]).await.is_err() {
if self.inject_if_running(vec![item]).await.is_err() {
tracing::debug!(
"skipping objective-updated goal steering because no turn is active"
);
@@ -1074,7 +1074,7 @@ impl Session {
.await;
if should_steer_budget_limit {
let item = budget_limit_steering_item(&goal);
if self.inject_response_items(vec![item]).await.is_err() {
if self.inject_if_running(vec![item]).await.is_err() {
tracing::debug!("skipping budget-limit goal steering because no turn is active");
}
*self.goal_runtime.budget_limit_reported_goal_id.lock().await = Some(goal_id);
@@ -1332,7 +1332,7 @@ impl Session {
candidate
.items
.into_iter()
.map(TurnInput::ResponseInputItem)
.map(TurnInput::ResponseItem)
.collect(),
)
.await;
@@ -1371,14 +1371,6 @@ impl Session {
tracing::debug!("skipping active goal continuation because a turn is already active");
return None;
}
if self
.input_queue
.has_queued_response_items_for_next_turn()
.await
{
tracing::debug!("skipping active goal continuation because queued input exists");
return None;
}
if self.input_queue.has_trigger_turn_mailbox_items().await {
tracing::debug!(
"skipping active goal continuation because trigger-turn mailbox input is pending"
@@ -1416,10 +1408,6 @@ impl Session {
return None;
}
if self.active_turn.lock().await.is_some()
|| self
.input_queue
.has_queued_response_items_for_next_turn()
.await
|| self.input_queue.has_trigger_turn_mailbox_items().await
{
tracing::debug!("skipping active goal continuation because pending work appeared");
@@ -1604,12 +1592,12 @@ fn escape_xml_text(input: &str) -> String {
.replace('>', "&gt;")
}
fn budget_limit_steering_item(goal: &ThreadGoal) -> ResponseInputItem {
fn budget_limit_steering_item(goal: &ThreadGoal) -> ResponseItem {
goal_context_input_item(budget_limit_prompt(goal))
}
fn goal_context_input_item(prompt: String) -> ResponseInputItem {
GoalContext::new(prompt).into_response_input_item()
fn goal_context_input_item(prompt: String) -> ResponseItem {
ResponseItem::from(GoalContext::new(prompt).into_response_input_item())
}
pub(crate) fn protocol_goal_from_state(goal: codex_state::ThreadGoal) -> ThreadGoal {
@@ -1678,7 +1666,7 @@ mod tests {
use codex_protocol::ThreadId;
use codex_protocol::config_types::ModeKind;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::ThreadGoal;
use codex_protocol::protocol::ThreadGoalStatus;
use codex_protocol::protocol::TokenUsage;
@@ -1807,7 +1795,8 @@ mod tests {
assert_eq!(
item,
ResponseInputItem::Message {
ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText {
text: "<goal_context>\nContinue working.\n</goal_context>".to_string(),
codex-rs/core/src/guardian/review_session.rs
+1 -2
View File
@@ -790,12 +790,11 @@ async fn run_review_on_session(
}
async fn append_guardian_followup_reminder(review_session: &GuardianReviewSession) {
let turn_context = review_session.codex.session.new_default_turn().await;
let reminder: ResponseItem = ContextualUserFragment::into(GuardianFollowupReviewReminder);
review_session
.codex
.session
.record_conversation_items(turn_context.as_ref(), std::slice::from_ref(&reminder))
.inject_no_new_turn(vec![reminder], /*current_turn_context*/ None)
.await;
}
codex-rs/core/src/guardian/tests.rs
+14 -14
View File
@@ -191,7 +191,8 @@ async fn guardian_test_session_and_turn_with_base_url(
async fn seed_guardian_parent_history(session: &Arc<Session>, turn: &Arc<TurnContext>) {
session
.record_into_history(
.record_conversation_items(
turn.as_ref(),
&[
ResponseItem::Message {
id: None,
@@ -225,7 +226,6 @@ async fn seed_guardian_parent_history(session: &Arc<Session>, turn: &Arc<TurnCon
phase: None,
},
],
turn.as_ref(),
)
.await;
}
@@ -362,7 +362,8 @@ async fn build_guardian_prompt_delta_mode_preserves_original_numbering() -> anyh
let (session, turn) = guardian_test_session_and_turn_with_base_url("http://localhost").await;
seed_guardian_parent_history(&session, &turn).await;
session
.record_into_history(
.record_conversation_items(
turn.as_ref(),
&[
ResponseItem::Message {
id: None,
@@ -381,7 +382,6 @@ async fn build_guardian_prompt_delta_mode_preserves_original_numbering() -> anyh
phase: None,
},
],
turn.as_ref(),
)
.await;
@@ -516,7 +516,8 @@ async fn build_guardian_prompt_stale_delta_version_falls_back_to_full_prompt() -
)
.await;
session
.record_into_history(
.record_conversation_items(
turn.as_ref(),
&[
ResponseItem::Message {
id: None,
@@ -535,7 +536,6 @@ async fn build_guardian_prompt_stale_delta_version_falls_back_to_full_prompt() -
phase: None,
},
],
turn.as_ref(),
)
.await;
@@ -1469,7 +1469,8 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow:
)
.await;
session
.record_into_history(
.record_conversation_items(
turn.as_ref(),
&[
ResponseItem::Message {
id: None,
@@ -1488,7 +1489,6 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow:
phase: None,
},
],
turn.as_ref(),
)
.await;
let second_request = GuardianApprovalRequest::Shell {
@@ -1513,7 +1513,8 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow:
)
.await;
session
.record_into_history(
.record_conversation_items(
turn.as_ref(),
&[
ResponseItem::Message {
id: None,
@@ -1532,7 +1533,6 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow:
phase: None,
},
],
turn.as_ref(),
)
.await;
let third_request = GuardianApprovalRequest::Shell {
@@ -2005,7 +2005,8 @@ async fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> a
ReviewDecision::Approved
);
session
.record_into_history(
.record_conversation_items(
turn.as_ref(),
&[
ResponseItem::Message {
id: None,
@@ -2024,7 +2025,6 @@ async fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> a
phase: None,
},
],
turn.as_ref(),
)
.await;
@@ -2072,7 +2072,8 @@ async fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> a
"second guardian request was not observed"
);
session
.record_into_history(
.record_conversation_items(
turn.as_ref(),
&[
ResponseItem::Message {
id: None,
@@ -2091,7 +2092,6 @@ async fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> a
phase: None,
},
],
turn.as_ref(),
)
.await;
codex-rs/core/src/hook_runtime.rs
+3 -4
View File
@@ -522,7 +522,7 @@ pub(crate) async fn inspect_pending_input(
)
.await
}
TurnInput::ResponseInputItem(_) => HookRuntimeOutcome {
TurnInput::ResponseItem(_) => HookRuntimeOutcome {
should_stop: false,
additional_contexts: Vec::new(),
},
@@ -540,9 +540,8 @@ pub(crate) async fn record_pending_input(
sess.record_user_prompt_and_emit_turn_item(turn_context.as_ref(), content.as_slice())
.await;
}
TurnInput::ResponseInputItem(input) => {
let response_item = ResponseItem::from(input);
sess.record_conversation_items(turn_context, std::slice::from_ref(&response_item))
TurnInput::ResponseItem(item) => {
sess.record_conversation_items(turn_context, std::slice::from_ref(&item))
.await;
}
}
codex-rs/core/src/session/handlers.rs
+7 -8
View File
@@ -27,6 +27,7 @@ use crate::tasks::UserShellCommandTask;
use crate::tasks::execute_user_shell_command;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::ErrorEvent;
use codex_protocol::protocol::Event;
@@ -255,7 +256,8 @@ pub(super) async fn user_input_or_turn_inner(
};
let mut task_input = additional_context_input
.into_iter()
.map(TurnInput::ResponseInputItem)
.map(ResponseItem::from)
.map(TurnInput::ResponseItem)
.collect::<Vec<_>>();
if !items.is_empty() {
task_input.push(TurnInput::UserInput(items));
@@ -899,17 +901,14 @@ Do not assume this also authorizes similar operations with different payloads.
Approved action:
{approved_action_json}"#,
);
let items = vec![ResponseInputItem::Message {
let items = vec![ResponseItem::from(ResponseInputItem::Message {
role: "developer".to_string(),
content: vec![ContentItem::InputText { text }],
phase: None,
}];
})];
if let Err(items) = sess.inject_response_items(items).await {
sess.input_queue
.queue_response_items_for_next_turn(items)
.await;
}
sess.inject_no_new_turn(items, /*current_turn_context*/ None)
.await;
}
pub(super) fn submission_dispatch_span(sub: &Submission) -> tracing::Span {
codex-rs/core/src/session/inject.rs
+50
View File
@@ -0,0 +1,50 @@
use super::input_queue::TurnInput;
use super::session::Session;
use super::turn_context::TurnContext;
use codex_protocol::models::ResponseItem;
impl Session {
/// Returns the input if there is no active turn to inject into.
#[expect(
clippy::await_holding_invalid_type,
reason = "active turn checks and turn state updates must remain atomic"
)]
pub async fn inject_if_running(
&self,
input: Vec<ResponseItem>,
) -> Result<(), Vec<ResponseItem>> {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(active_turn) => {
self.input_queue
.extend_pending_input_for_turn_state(
active_turn.turn_state.as_ref(),
input.into_iter().map(TurnInput::ResponseItem).collect(),
)
.await;
Ok(())
}
None => Err(input),
}
}
/// Injects items into active work, or records them without starting a turn.
pub(crate) async fn inject_no_new_turn(
&self,
items: Vec<ResponseItem>,
current_turn_context: Option<&TurnContext>,
) {
let Err(items) = self.inject_if_running(items).await else {
return;
};
let default_turn_context;
let turn_context = match current_turn_context {
Some(turn_context) => turn_context,
None => {
default_turn_context = self.new_default_turn().await;
default_turn_context.as_ref()
}
};
self.record_conversation_items(turn_context, &items).await;
}
}
codex-rs/core/src/session/input_queue.rs
+7 -52
View File
@@ -1,7 +1,7 @@
use crate::state::ActiveTurn;
use crate::state::MailboxDeliveryPhase;
use crate::state::TurnState;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::user_input::UserInput;
use std::collections::VecDeque;
@@ -12,7 +12,7 @@ use tokio::sync::watch;
#[derive(Clone, Debug, PartialEq)]
pub(crate) enum TurnInput {
UserInput(Vec<UserInput>),
ResponseInputItem(ResponseInputItem),
ResponseItem(ResponseItem),
}
/// Turn-local pending input storage owned by the input queue flow.
@@ -25,8 +25,6 @@ pub(crate) struct TurnInputQueue {
pub(crate) struct InputQueue {
mailbox_tx: watch::Sender<()>,
mailbox_pending_mails: Mutex<VecDeque<InterAgentCommunication>>,
idle_pending_input: Mutex<Vec<ResponseInputItem>>,
}
impl InputQueue {
@@ -35,7 +33,6 @@ impl InputQueue {
Self {
mailbox_tx,
mailbox_pending_mails: Mutex::new(VecDeque::new()),
idle_pending_input: Mutex::new(Vec::new()),
}
}
@@ -70,31 +67,15 @@ impl InputQueue {
.any(|mail| mail.trigger_turn)
}
pub(crate) async fn drain_mailbox_input_items(&self) -> Vec<ResponseInputItem> {
pub(crate) async fn drain_mailbox_input_items(&self) -> Vec<ResponseItem> {
self.mailbox_pending_mails
.lock()
.await
.drain(..)
.map(|mail| mail.to_response_input_item())
.map(|mail| ResponseItem::from(mail.to_response_input_item()))
.collect()
}
pub(crate) async fn queue_response_items_for_next_turn(&self, items: Vec<ResponseInputItem>) {
if items.is_empty() {
return;
}
self.idle_pending_input.lock().await.extend(items);
}
pub(crate) async fn take_queued_response_items_for_next_turn(&self) -> Vec<ResponseInputItem> {
std::mem::take(&mut *self.idle_pending_input.lock().await)
}
pub(crate) async fn has_queued_response_items_for_next_turn(&self) -> bool {
!self.idle_pending_input.lock().await.is_empty()
}
pub(crate) async fn turn_state_for_sub_id(
&self,
active_turn: &Mutex<Option<ActiveTurn>>,
@@ -181,32 +162,6 @@ impl InputQueue {
turn_state.lock().await.pending_input.items.split_off(0)
}
#[expect(
clippy::await_holding_invalid_type,
reason = "active turn checks and turn state updates must remain atomic"
)]
pub(crate) async fn inject_response_items(
&self,
active_turn: &Mutex<Option<ActiveTurn>>,
input: Vec<ResponseInputItem>,
) -> Result<(), Vec<ResponseInputItem>> {
let mut active = active_turn.lock().await;
match active.as_mut() {
Some(active_turn) => {
self.extend_pending_input_for_turn_state(
active_turn.turn_state.as_ref(),
input
.into_iter()
.map(TurnInput::ResponseInputItem)
.collect(),
)
.await;
Ok(())
}
None => Err(input),
}
}
#[expect(
clippy::await_holding_invalid_type,
reason = "active turn checks and turn state updates must remain atomic"
@@ -235,7 +190,7 @@ impl InputQueue {
.drain_mailbox_input_items()
.await
.into_iter()
.map(TurnInput::ResponseInputItem);
.map(TurnInput::ResponseItem);
if pending_input.is_empty() {
mailbox_items.collect()
} else {
@@ -345,8 +300,8 @@ mod tests {
assert_eq!(
input_queue.drain_mailbox_input_items().await,
vec![
mail_one.to_response_input_item(),
mail_two.to_response_input_item()
ResponseItem::from(mail_one.to_response_input_item()),
ResponseItem::from(mail_two.to_response_input_item())
]
);
assert!(!input_queue.has_pending_mailbox_items().await);
codex-rs/core/src/session/mod.rs
+18 -67
View File
@@ -196,6 +196,7 @@ use codex_protocol::exec_output::StreamOutput;
mod config_lock;
mod handlers;
mod inject;
mod input_queue;
mod mcp;
mod multi_agents;
@@ -324,7 +325,6 @@ use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::Personality;
use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig;
use codex_protocol::config_types::WindowsSandboxLevel;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;
@@ -1842,27 +1842,11 @@ impl Session {
warn!("execpolicy amendment for {sub_id} had no command prefix");
return;
};
let fragment = ApprovedCommandPrefixSaved::new(prefixes);
let text = fragment.render();
let message: ResponseItem = ContextualUserFragment::into(fragment);
if let Some(turn_context) = self.turn_context_for_sub_id(sub_id).await {
self.record_conversation_items(&turn_context, std::slice::from_ref(&message))
.await;
return;
}
if self
.inject_response_items(vec![ResponseInputItem::Message {
role: "developer".to_string(),
content: vec![ContentItem::InputText { text }],
phase: None,
}])
.await
.is_err()
{
warn!("no active turn found to record execpolicy amendment message for {sub_id}");
}
let message: ResponseItem =
ContextualUserFragment::into(ApprovedCommandPrefixSaved::new(prefixes));
let turn_context = self.turn_context_for_sub_id(sub_id).await;
self.inject_no_new_turn(vec![message], turn_context.as_deref())
.await;
}
pub(crate) async fn persist_network_policy_amendment(
@@ -1939,27 +1923,10 @@ impl Session {
sub_id: &str,
amendment: &NetworkPolicyAmendment,
) {
let fragment = NetworkRuleSaved::new(amendment);
let text = fragment.render();
let message: ResponseItem = ContextualUserFragment::into(fragment);
if let Some(turn_context) = self.turn_context_for_sub_id(sub_id).await {
self.record_conversation_items(&turn_context, std::slice::from_ref(&message))
.await;
return;
}
if self
.inject_response_items(vec![ResponseInputItem::Message {
role: "developer".to_string(),
content: vec![ContentItem::InputText { text }],
phase: None,
}])
.await
.is_err()
{
warn!("no active turn found to record network policy amendment message for {sub_id}");
}
let message: ResponseItem = ContextualUserFragment::into(NetworkRuleSaved::new(amendment));
let turn_context = self.turn_context_for_sub_id(sub_id).await;
self.inject_no_new_turn(vec![message], turn_context.as_deref())
.await;
}
/// Emit an exec approval request event and await the user's decision.
@@ -2510,28 +2477,21 @@ impl Session {
}
}
/// Records input items: always append to conversation history and
/// persist these response items to rollout.
/// Records conversation items: append to history, persist to rollout, and
/// notify clients observing raw response items.
pub(crate) async fn record_conversation_items(
&self,
turn_context: &TurnContext,
items: &[ResponseItem],
) {
self.record_into_history(items, turn_context).await;
{
let mut state = self.state.lock().await;
state.record_items(items.iter(), turn_context.truncation_policy);
}
self.persist_rollout_response_items(items).await;
self.send_raw_response_items(turn_context, items).await;
}
/// Append ResponseItems to the in-memory conversation history only.
pub(crate) async fn record_into_history(
&self,
items: &[ResponseItem],
turn_context: &TurnContext,
) {
let mut state = self.state.lock().await;
state.record_items(items.iter(), turn_context.truncation_policy);
}
async fn maybe_warn_on_server_model_mismatch(
self: &Arc<Self>,
turn_context: &Arc<TurnContext>,
@@ -3192,7 +3152,8 @@ impl Session {
let mut pending_input = additional_context_input
.into_iter()
.map(TurnInput::ResponseInputItem)
.map(ResponseItem::from)
.map(TurnInput::ResponseItem)
.collect::<Vec<_>>();
pending_input.push(TurnInput::UserInput(input));
self.input_queue
@@ -3204,16 +3165,6 @@ impl Session {
Ok(active_turn_id.clone())
}
/// Returns the input if there was no task running to inject into.
pub async fn inject_response_items(
&self,
input: Vec<ResponseInputItem>,
) -> Result<(), Vec<ResponseInputItem>> {
self.input_queue
.inject_response_items(&self.active_turn, input)
.await
}
pub(crate) async fn record_memory_citation_for_turn(&self, sub_id: &str) {
let turn_state = self
.input_queue
codex-rs/core/src/session/tests.rs
+40 -81
View File
@@ -103,7 +103,6 @@ use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Settings;
use codex_protocol::models::BaseInstructions;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::CodexErrorInfo;
@@ -399,6 +398,12 @@ async fn interrupting_regular_turn_waiting_on_startup_prewarm_emits_turn_aborted
sess.abort_all_tasks(TurnAbortReason::Interrupted).await;
let marker_evt = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
.await
.expect("expected turn aborted marker event")
.expect("channel open");
assert!(matches!(marker_evt.msg, EventMsg::RawResponseItem(_)));
let second = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
.await
.expect("expected turn aborted event")
@@ -1773,7 +1778,7 @@ async fn recompute_token_usage_uses_session_base_instructions() {
let item = user_message("hello");
session
.record_into_history(std::slice::from_ref(&item), &turn_context)
.record_conversation_items(&turn_context, std::slice::from_ref(&item))
.await;
let history = session.clone_history().await;
@@ -2627,7 +2632,7 @@ async fn thread_rollback_fails_without_persisted_thread_history() {
let (sess, tc, rx) = make_session_and_context_with_rx().await;
let initial_context = sess.build_initial_context(tc.as_ref()).await;
sess.record_into_history(&initial_context, tc.as_ref())
sess.record_conversation_items(tc.as_ref(), &initial_context)
.await;
handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await;
@@ -3000,7 +3005,7 @@ async fn thread_rollback_fails_when_turn_in_progress() {
let (sess, tc, rx) = make_session_and_context_with_rx().await;
let initial_context = sess.build_initial_context(tc.as_ref()).await;
sess.record_into_history(&initial_context, tc.as_ref())
sess.record_conversation_items(tc.as_ref(), &initial_context)
.await;
*sess.active_turn.lock().await = Some(crate::state::ActiveTurn::default());
@@ -3021,7 +3026,7 @@ async fn thread_rollback_fails_when_num_turns_is_zero() {
let (sess, tc, rx) = make_session_and_context_with_rx().await;
let initial_context = sess.build_initial_context(tc.as_ref()).await;
sess.record_into_history(&initial_context, tc.as_ref())
sess.record_conversation_items(tc.as_ref(), &initial_context)
.await;
handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 0).await;
@@ -7611,7 +7616,7 @@ async fn record_context_updates_and_set_reference_context_item_reinjects_full_co
phase: None,
};
session
.record_into_history(std::slice::from_ref(&compacted_summary), &turn_context)
.record_conversation_items(&turn_context, std::slice::from_ref(&compacted_summary))
.await;
session
.record_context_updates_and_set_reference_context_item(&turn_context)
@@ -8042,7 +8047,7 @@ async fn guardian_helper_review_interrupts_after_three_consecutive_denials() {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[test_log::test]
async fn abort_regular_task_emits_turn_aborted_only() {
async fn abort_regular_task_emits_marker_before_turn_aborted() {
let (sess, tc, rx) = make_session_and_context_with_rx().await;
let input = vec![TurnInput::UserInput(vec![UserInput::Text {
text: "hello".to_string(),
@@ -8060,8 +8065,13 @@ async fn abort_regular_task_emits_turn_aborted_only() {
sess.abort_all_tasks(TurnAbortReason::Interrupted).await;
// Interrupts persist a model-visible `<turn_aborted>` marker into history, but there is no
// separate client-visible event for that marker (only `EventMsg::TurnAborted`).
// Interrupts surface the model-visible `<turn_aborted>` marker before the abort event.
let marker_evt = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
.await
.expect("timeout waiting for marker event")
.expect("event");
assert!(matches!(marker_evt.msg, EventMsg::RawResponseItem(_)));
let evt = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
.await
.expect("timeout waiting for event")
@@ -8075,7 +8085,7 @@ async fn abort_regular_task_emits_turn_aborted_only() {
}
#[tokio::test]
async fn abort_gracefully_emits_turn_aborted_only() {
async fn abort_gracefully_emits_marker_before_turn_aborted() {
let (sess, tc, rx) = make_session_and_context_with_rx().await;
let input = vec![TurnInput::UserInput(vec![UserInput::Text {
text: "hello".to_string(),
@@ -8093,8 +8103,13 @@ async fn abort_gracefully_emits_turn_aborted_only() {
sess.abort_all_tasks(TurnAbortReason::Interrupted).await;
// Even if tasks handle cancellation gracefully, interrupts still result in `TurnAborted`
// being the only client-visible signal.
// Gracefully cancelled tasks surface the model-visible marker before the abort event too.
let marker_evt = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
.await
.expect("timeout waiting for marker event")
.expect("event");
assert!(matches!(marker_evt.msg, EventMsg::RawResponseItem(_)));
let evt = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
.await
.expect("timeout waiting for event")
@@ -8363,66 +8378,11 @@ async fn steer_input_returns_active_turn_id() {
assert!(sess.input_queue.has_pending_input(&sess.active_turn).await);
}
#[tokio::test]
async fn queued_response_items_for_next_turn_move_into_next_active_turn() {
let (sess, tc, _rx) = make_session_and_context_with_rx().await;
let queued_item = ResponseInputItem::Message {
role: "assistant".to_string(),
content: vec![ContentItem::InputText {
text: "queued before wake".to_string(),
}],
phase: None,
};
sess.input_queue
.queue_response_items_for_next_turn(vec![queued_item.clone()])
.await;
sess.spawn_task(
Arc::clone(&tc),
Vec::new(),
NeverEndingTask {
kind: TaskKind::Regular,
listen_to_cancellation_token: false,
},
)
.await;
assert_eq!(
sess.input_queue.get_pending_input(&sess.active_turn).await,
vec![TurnInput::ResponseInputItem(queued_item)]
);
}
#[tokio::test]
async fn idle_interrupt_does_not_wake_queued_next_turn_items() {
let (sess, _tc, _rx) = make_session_and_context_with_rx().await;
let queued_item = ResponseInputItem::Message {
role: "assistant".to_string(),
content: vec![ContentItem::InputText {
text: "queued before interrupt".to_string(),
}],
phase: None,
};
sess.input_queue
.queue_response_items_for_next_turn(vec![queued_item])
.await;
sess.abort_all_tasks(TurnAbortReason::Interrupted).await;
assert!(sess.active_turn.lock().await.is_none());
assert!(
sess.input_queue
.has_queued_response_items_for_next_turn()
.await
);
}
#[tokio::test]
async fn abort_empty_active_turn_preserves_pending_input() {
let (sess, _tc, _rx) = make_session_and_context_with_rx().await;
let pending_item = ResponseInputItem::Message {
let pending_item = ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText {
text: "late pending input".to_string(),
@@ -8437,7 +8397,7 @@ async fn abort_empty_active_turn_preserves_pending_input() {
sess.input_queue
.extend_pending_input_for_turn_state(
turn_state.as_ref(),
vec![TurnInput::ResponseInputItem(pending_item.clone())],
vec![TurnInput::ResponseItem(pending_item.clone())],
)
.await;
@@ -8448,7 +8408,7 @@ async fn abort_empty_active_turn_preserves_pending_input() {
sess.input_queue
.take_pending_input_for_turn_state(turn_state.as_ref())
.await,
vec![TurnInput::ResponseInputItem(pending_item)]
vec![TurnInput::ResponseItem(pending_item)]
);
}
@@ -8874,7 +8834,7 @@ async fn budget_limited_accounting_steers_active_turn_without_aborting() -> anyh
.await?;
let pending_input = sess.input_queue.get_pending_input(&sess.active_turn).await;
let [TurnInput::ResponseInputItem(ResponseInputItem::Message { role, content, .. })] =
let [TurnInput::ResponseItem(ResponseItem::Message { role, content, .. })] =
pending_input.as_slice()
else {
panic!("expected one budget-limit steering message, got {pending_input:#?}");
@@ -9101,7 +9061,7 @@ async fn external_objective_change_steers_active_turn() -> anyhow::Result<()> {
pending_input.iter().any(|item| {
matches!(
item,
TurnInput::ResponseInputItem(ResponseInputItem::Message { role, content, .. })
TurnInput::ResponseItem(ResponseItem::Message { role, content, .. })
if role == "user"
&& content.iter().any(|content| matches!(
content,
@@ -9325,9 +9285,9 @@ async fn queue_only_mailbox_mail_waits_for_next_turn_after_answer_boundary() {
assert_eq!(
sess.input_queue.get_pending_input(&sess.active_turn).await,
vec![TurnInput::ResponseInputItem(
vec![TurnInput::ResponseItem(ResponseItem::from(
communication.to_response_input_item()
)],
))],
);
}
@@ -9412,7 +9372,7 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() {
text: "follow up".to_string(),
text_elements: Vec::new(),
}]),
TurnInput::ResponseInputItem(communication.to_response_input_item()),
TurnInput::ResponseItem(ResponseItem::from(communication.to_response_input_item())),
],
);
}
@@ -9466,7 +9426,7 @@ async fn stale_defer_mailbox_delivery_does_not_override_steered_input() {
text: "follow up".to_string(),
text_elements: Vec::new(),
}]),
TurnInput::ResponseInputItem(communication.to_response_input_item()),
TurnInput::ResponseItem(ResponseItem::from(communication.to_response_input_item())),
],
);
}
@@ -9521,9 +9481,9 @@ async fn tool_calls_reopen_mailbox_delivery_for_current_turn() {
assert!(output.tool_future.is_some());
assert_eq!(
sess.input_queue.get_pending_input(&sess.active_turn).await,
vec![TurnInput::ResponseInputItem(
vec![TurnInput::ResponseItem(ResponseItem::from(
communication.to_response_input_item()
)],
))],
);
}
@@ -9581,8 +9541,7 @@ async fn abort_review_task_emits_exited_then_aborted_and_records_history() {
);
let history = sess.clone_history().await;
// The `<turn_aborted>` marker is silent in the event stream, so verify it is still
// recorded in history for the model.
// Verify the `<turn_aborted>` marker is still recorded in history for the model.
assert!(
history.raw_items().iter().any(|item| {
let ResponseItem::Message { role, content, .. } = item else {
codex-rs/core/src/session/turn.rs
+2 -2
View File
@@ -458,7 +458,7 @@ async fn build_skills_and_plugins(
.iter()
.filter_map(|item| match item {
TurnInput::UserInput(content) => Some(content.as_slice()),
TurnInput::ResponseInputItem(_) => None,
TurnInput::ResponseItem(_) => None,
})
.flatten()
.cloned()
@@ -611,7 +611,7 @@ async fn track_turn_resolved_config_analytics(
.iter()
.filter_map(|item| match item {
TurnInput::UserInput(content) => Some(content.as_slice()),
TurnInput::ResponseInputItem(_) => None,
TurnInput::ResponseItem(_) => None,
})
.flatten()
.filter(|item| {
codex-rs/core/src/tasks/mod.rs
+10 -25
View File
@@ -44,7 +44,6 @@ use codex_otel::TURN_TOKEN_USAGE_METRIC;
use codex_otel::TURN_TOOL_CALL_METRIC;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::TokenUsage;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::TurnAbortedEvent;
@@ -347,11 +346,7 @@ impl Session {
{
warn!("failed to apply goal runtime turn-start event: {err}");
}
let queued_response_items = self
.input_queue
.take_queued_response_items_for_next_turn()
.await;
let mailbox_items = self.input_queue.get_pending_input(&self.active_turn).await;
let pending_items = self.input_queue.get_pending_input(&self.active_turn).await;
let turn_state = {
let mut active = self.active_turn.lock().await;
let turn = active.get_or_insert_with(ActiveTurn::default);
@@ -359,11 +354,6 @@ impl Session {
Arc::clone(&turn.turn_state)
};
turn_state.lock().await.token_usage_at_turn_start = token_usage_at_turn_start.clone();
let mut pending_items = queued_response_items
.into_iter()
.map(TurnInput::ResponseInputItem)
.collect::<Vec<_>>();
pending_items.extend(mailbox_items);
self.input_queue
.extend_pending_input_for_turn_state(turn_state.as_ref(), pending_items)
.await;
@@ -452,8 +442,7 @@ impl Session {
/// Starts a regular turn when the session is idle and pending work is waiting.
///
/// Pending work currently includes queued next-turn items and mailbox mail marked with
/// `trigger_turn`.
/// Pending work currently includes mailbox mail marked with `trigger_turn`.
///
/// This helper generates a fresh sub-id for the synthetic turn before delegating to the
/// explicit-sub-id variant.
@@ -465,18 +454,13 @@ impl Session {
/// Starts a regular turn with the provided sub-id when pending work should wake an idle
/// session.
///
/// The turn is created only when there are queued next-turn items or mailbox mail marked with
/// `trigger_turn`, and only if the session is currently idle.
/// The turn is created only when there is mailbox mail marked with `trigger_turn`, and only
/// if the session is currently idle.
pub(crate) async fn maybe_start_turn_for_pending_work_with_sub_id(
self: &Arc<Self>,
sub_id: String,
) {
if !self
.input_queue
.has_queued_response_items_for_next_turn()
.await
&& !self.input_queue.has_trigger_turn_mailbox_items().await
{
if !self.input_queue.has_trigger_turn_mailbox_items().await {
return;
}
@@ -859,10 +843,11 @@ impl Session {
InterruptedTurnHistoryMarker::from_config(task.turn_context.config.as_ref()),
)
{
self.record_into_history(std::slice::from_ref(&marker), task.turn_context.as_ref())
.await;
self.persist_rollout_items(&[RolloutItem::ResponseItem(marker)])
.await;
self.record_conversation_items(
task.turn_context.as_ref(),
std::slice::from_ref(&marker),
)
.await;
// Ensure the marker is durably visible before emitting TurnAborted: some clients
// synchronously re-read the rollout on receipt of the abort event.
if let Err(err) = self.flush_rollout().await {
codex-rs/core/src/tasks/review.rs
+1 -1
View File
@@ -73,7 +73,7 @@ impl SessionTask for ReviewTask {
for item in input {
match item {
TurnInput::UserInput(mut content) => user_input.append(&mut content),
TurnInput::ResponseInputItem(_) => {}
TurnInput::ResponseItem(_) => {}
}
}
codex-rs/core/src/tasks/user_shell.rs
+3 -28
View File
@@ -38,8 +38,6 @@ use super::SessionTask;
use super::SessionTaskContext;
use crate::session::session::Session;
use codex_protocol::models::PermissionProfile;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
const USER_SHELL_TIMEOUT_MS: u64 = 60 * 60 * 1000; // 1 hour
@@ -361,30 +359,7 @@ async fn persist_user_shell_output(
return;
}
let response_input_item = match output_item {
ResponseItem::Message {
role,
content,
phase,
..
} => ResponseInputItem::Message {
role,
content,
phase,
},
_ => unreachable!("user shell command output record should always be a message"),
};
if let Err(items) = session
.inject_response_items(vec![response_input_item])
.await
{
let response_items = items
.into_iter()
.map(ResponseItem::from)
.collect::<Vec<_>>();
session
.record_conversation_items(turn_context, &response_items)
.await;
}
session
.inject_no_new_turn(vec![output_item], Some(turn_context))
.await;
}
codex-rs/core/src/thread_manager.rs
-13
View File
@@ -33,8 +33,6 @@ use codex_protocol::ThreadId;
use codex_protocol::config_types::CollaborationModeMask;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
#[cfg(test)]
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ModelPreset;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
@@ -1016,17 +1014,6 @@ impl ThreadManagerState {
thread.submit(op).await
}
#[cfg(test)]
/// Append a prebuilt message to a thread by ID outside the normal user-input path.
pub(crate) async fn append_message(
&self,
thread_id: ThreadId,
message: ResponseItem,
) -> CodexResult<String> {
let thread = self.get_thread(thread_id).await?;
thread.append_message(message).await
}
/// Remove a thread from the manager by ID, returning it when present.
pub(crate) async fn remove_thread(&self, thread_id: &ThreadId) -> Option<Arc<CodexThread>> {
self.threads.write().await.remove(thread_id)
codex-rs/core/src/tools/code_mode/mod.rs
+2 -2
View File
@@ -13,7 +13,7 @@ use codex_code_mode::CodeModeTurnHost;
use codex_code_mode::RuntimeResponse;
use codex_protocol::models::FunctionCallOutputContentItem;
use codex_protocol::models::FunctionCallOutputPayload;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
use serde_json::Value as JsonValue;
use tokio_util::sync::CancellationToken;
@@ -134,7 +134,7 @@ impl CodeModeTurnHost for CoreTurnHost {
}
self.exec
.session
.inject_response_items(vec![ResponseInputItem::CustomToolCallOutput {
.inject_if_running(vec![ResponseItem::CustomToolCallOutput {
call_id,
name: Some(PUBLIC_TOOL_NAME.to_string()),
output: FunctionCallOutputPayload::from_text(text),
codex-rs/core/src/tools/handlers/extension_tools.rs
+1 -1
View File
@@ -208,7 +208,7 @@ mod tests {
phase: None,
};
session
.record_into_history(std::slice::from_ref(&history_item), &turn)
.record_conversation_items(&turn, std::slice::from_ref(&history_item))
.await;
let invocation = ToolInvocation {
session: session.into(),
codex-rs/core/src/tools/router_tests.rs
+1 -1
View File
@@ -332,7 +332,7 @@ async fn extension_tool_executors_are_model_visible_and_dispatchable() -> anyhow
phase: None,
};
session
.record_into_history(std::slice::from_ref(&history_item), &turn)
.record_conversation_items(&turn, std::slice::from_ref(&history_item))
.await;
let router = ToolRouter::from_turn_context(