Skip to content

[codex] Store pending response items directly #24865

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pakrym-oai merged 5 commits into from
May 28, 2026
Merged

[codex] Store pending response items directly #24865

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
0a683cc
core: store pending response items directly
pakrym-oai May 28, 2026
49cb5a6
core: remove test-only injection helper from library build
pakrym-oai May 28, 2026
8c80006
core: remove unused idle response input queue
pakrym-oai May 28, 2026
59b96e5
core: remove stale thread manager test import
pakrym-oai May 28, 2026
7a37149
Merge branch 'main' into pakrym/responseitem-pending-input
jif-oai May 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 0 additions & 16 deletions codex-rs/core/src/agent/control.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
54 changes: 0 additions & 54 deletions codex-rs/core/src/agent/control_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
77 changes: 11 additions & 66 deletions codex-rs/core/src/codex_thread.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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
.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
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.
Expand All @@ -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;
Comment on lines 401 to 404

@chatgpt-codex-connector chatgpt-codex-connector Bot May 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Preserve raw injected items immediately

When CodexThread::inject_response_items is called while a turn is active, this now routes through inject_no_new_turn, whose inject_if_running success path only appends the items to the active turn's pending-input queue and returns without recording them. The method still returns Ok after flush_rollout, but the items are not in history/rollout until a later pending-input drain; if the active turn is interrupted or aborted first, abort_all_tasks clears that pending input, so an API call that previously durably appended raw items can silently lose them.

Useful? React with 👍 / 👎.

@pakrym-oai pakrym-oai May 28, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Temporary cost of unification for an advanced-use method.

self.codex.session.flush_rollout().await?;
Ok(())
Expand Down Expand Up @@ -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(),
)),
}
}
2 changes: 1 addition & 1 deletion codex-rs/core/src/compact.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Comment on lines 570 to 572

@chatgpt-codex-connector chatgpt-codex-connector Bot May 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Do not persist unfinished compact outputs

In local compaction, drain_to_completed explicitly treats a stream that closes before response.completed as an error, but this call now persists every OutputItemDone to rollout before that completion is observed. If the compact model emits the summary item and then the stream disconnects before response.completed, the compaction fails/retries without installing replace_compacted_history, yet the unfinished compact output is already durable and will pollute reconstructed history after a restart; the old record_into_history path did not write these provisional items to rollout.

Useful? React with 👍 / 👎.

@pakrym-oai pakrym-oai May 28, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

replace_compacted_history "will" wipe these extra items anyway.

Comment on lines +571 to 572

@chatgpt-codex-connector chatgpt-codex-connector Bot May 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Keep local compaction output transient until completion

For local compaction, this now persists and emits each compaction model OutputItemDone, whereas the previous record_into_history call kept these synthesized summary outputs only in memory until replace_compacted_history runs after response.completed. If the compaction stream emits an output item and then fails or is interrupted before completion, the rollout now contains that partial compaction response as ordinary conversation history, so a resumed session can include a summary the compaction never successfully installed.

Useful? React with 👍 / 👎.

}
Ok(ResponseEvent::ServerReasoningIncluded(included)) => {
Expand Down
33 changes: 11 additions & 22 deletions codex-rs/core/src/goals.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -177,7 +177,7 @@ pub(crate) struct GoalRuntimeState {

struct GoalContinuationCandidate {
goal_id: String,
items: Vec<ResponseInputItem>,
items: Vec<ResponseItem>,
}

impl GoalRuntimeState {
Expand Down Expand Up @@ -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"
);
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -1332,7 +1332,7 @@ impl Session {
candidate
.items
.into_iter()
.map(TurnInput::ResponseInputItem)
.map(TurnInput::ResponseItem)
.collect(),
)
.await;
Expand Down Expand Up @@ -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"
Expand Down Expand Up @@ -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");
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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(),
Expand Down
3 changes: 1 addition & 2 deletions codex-rs/core/src/guardian/review_session.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}

Expand Down
Loading
Loading