hanzoai/dev
2
0
Fork 0
mirror of https://github.com/hanzoai/dev.git synced 2026-10-25 03:23:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Unify compaction flow and persist GPT-5 overrides

Browse Source
This commit is contained in:
James Peter
2025-09-18 12:06:33 +10:00
parent e1acde2829
commit 26e538a66d
10 changed files with 458 additions and 376 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/chat_completions.rs
+2 -1
View File
@@ -34,6 +34,7 @@ use codex_protocol::models::ResponseItem;
pub(crate) async fn stream_chat_completions(
prompt: &Prompt,
model_family: &ModelFamily,
model_slug: &str,
client: &reqwest::Client,
provider: &ModelProviderInfo,
debug_logger: &Arc<Mutex<DebugLogger>>,
@@ -284,7 +285,7 @@ pub(crate) async fn stream_chat_completions(
let tools_json = create_tools_json_for_chat_completions_api(&prompt.tools)?;
let mut payload = json!({
"model": model_family.slug,
"model": model_slug,
"messages": messages,
"stream": true,
"tools": tools_json,
codex-rs/core/src/client.rs
+29 -3
View File
@@ -114,6 +114,11 @@ impl ModelClient {
self.effort
}
/// Get the reasoning summary configuration
pub fn get_reasoning_summary(&self) -> ReasoningSummaryConfig {
self.summary
}
/// Get the text verbosity configuration
#[allow(dead_code)]
pub fn get_text_verbosity(&self) -> TextVerbosityConfig {
@@ -133,10 +138,19 @@ impl ModelClient {
match self.provider.wire_api {
WireApi::Responses => self.stream_responses(prompt).await,
WireApi::Chat => {
let effective_family = prompt
.model_family_override
.as_ref()
.unwrap_or(&self.config.model_family);
let model_slug = prompt
.model_override
.as_deref()
.unwrap_or(self.config.model.as_str());
// Create the raw streaming connection first.
let response_stream = stream_chat_completions(
prompt,
&self.config.model_family,
effective_family,
model_slug,
&self.client,
&self.provider,
&self.debug_logger,
@@ -231,6 +245,11 @@ impl ModelClient {
// For Azure, we send `store: true` and preserve reasoning item IDs.
let azure_workaround = self.provider.is_azure_responses_endpoint();
let model_slug = prompt
.model_override
.as_deref()
.unwrap_or(self.config.model.as_str());
let payload = ResponsesApiRequest {
model: &self.config.model,
instructions: &full_instructions,
@@ -248,6 +267,9 @@ impl ModelClient {
};
let mut payload_json = serde_json::to_value(&payload)?;
if let Some(model_value) = payload_json.get_mut("model") {
*model_value = serde_json::Value::String(model_slug.to_string());
}
if azure_workaround {
attach_item_ids(&mut payload_json, &input_with_instructions);
}
@@ -260,12 +282,16 @@ impl ModelClient {
let endpoint = self
.provider
.get_full_url(&auth_manager.as_ref().and_then(|m| m.auth()));
trace!("POST to {}: {}", endpoint, serde_json::to_string(&payload)?);
trace!(
"POST to {}: {}",
endpoint,
serde_json::to_string(&payload_json)?
);
// Start logging the request and get a request_id to track the response
let request_id = if let Ok(logger) = self.debug_logger.lock() {
logger
.start_request_log(&endpoint, &serde_json::to_value(&payload)?)
.start_request_log(&endpoint, &payload_json)
.unwrap_or_default()
} else {
String::new()
codex-rs/core/src/client_common.rs
+76 -30
View File
@@ -33,7 +33,7 @@ const USER_INSTRUCTIONS_END: &str = "\n\n</user_instructions>";
pub const REVIEW_PROMPT: &str = include_str!("../review_prompt.md");
/// API request payload for a single model turn
#[derive(Default, Debug, Clone)]
#[derive(Debug, Clone)]
pub struct Prompt {
/// Conversation context input items.
pub input: Vec<ResponseItem>,
@@ -59,16 +59,44 @@ pub struct Prompt {
/// Optional override for the built-in BASE_INSTRUCTIONS.
pub base_instructions_override: Option<String>,
/// Whether to prepend the default developer instructions block.
pub include_additional_instructions: bool,
/// Optional `text.format` for structured outputs (used by side-channel requests).
pub text_format: Option<TextFormat>,
/// Optional per-request model slug override.
pub model_override: Option<String>,
/// Optional per-request model family override matching `model_override`.
pub model_family_override: Option<ModelFamily>,
}
impl Default for Prompt {
fn default() -> Self {
Self {
input: Vec::new(),
store: false,
user_instructions: None,
environment_context: None,
tools: Vec::new(),
status_items: Vec::new(),
base_instructions_override: None,
include_additional_instructions: true,
text_format: None,
model_override: None,
model_family_override: None,
}
}
}
impl Prompt {
pub(crate) fn get_full_instructions(&self, model: &ModelFamily) -> Cow<'_, str> {
let effective_model = self.model_family_override.as_ref().unwrap_or(model);
let base = self
.base_instructions_override
.as_deref()
.unwrap_or(model.base_instructions.deref());
.unwrap_or(effective_model.base_instructions.deref());
let mut sections: Vec<&str> = vec![base];
// When there are no custom instructions, add apply_patch_tool_instructions if:
@@ -81,7 +109,7 @@ impl Prompt {
_ => false,
});
if self.base_instructions_override.is_none()
&& model.needs_special_apply_patch_instructions
&& effective_model.needs_special_apply_patch_instructions
&& !is_apply_patch_tool_present
{
sections.push(APPLY_PATCH_TOOL_INSTRUCTIONS);
@@ -96,37 +124,55 @@ impl Prompt {
}
fn get_formatted_environment_context(&self) -> Option<String> {
self.environment_context
.as_ref()
.map(|ec| {
let ec_str = serde_json::to_string_pretty(ec).unwrap_or_else(|_| format!("{:?}", ec));
format!("{ENVIRONMENT_CONTEXT_START}{ec_str}{ENVIRONMENT_CONTEXT_END}")
})
self.environment_context.as_ref().map(|ec| {
let ec_str = serde_json::to_string_pretty(ec).unwrap_or_else(|_| format!("{:?}", ec));
format!("{ENVIRONMENT_CONTEXT_START}{ec_str}{ENVIRONMENT_CONTEXT_END}")
})
}
pub(crate) fn get_formatted_input(&self) -> Vec<ResponseItem> {
let mut input_with_instructions =
Vec::with_capacity(self.input.len() + self.status_items.len() + 3);
input_with_instructions.push(ResponseItem::Message {
id: None,
role: "developer".to_string(),
content: vec![ContentItem::InputText {
text: ADDITIONAL_INSTRUCTIONS.to_string(),
}],
});
if let Some(ec) = self.get_formatted_environment_context() {
if self.include_additional_instructions {
input_with_instructions.push(ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText { text: ec }],
});
}
if let Some(ui) = self.get_formatted_user_instructions() {
input_with_instructions.push(ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText { text: ui }],
role: "developer".to_string(),
content: vec![ContentItem::InputText {
text: ADDITIONAL_INSTRUCTIONS.to_string(),
}],
});
if let Some(ec) = self.get_formatted_environment_context() {
let has_environment_context = self.input.iter().any(|item| {
matches!(item, ResponseItem::Message { role, content, .. }
if role == "user"
&& content.iter().any(|c| matches!(c,
ContentItem::InputText { text } if text.contains(ENVIRONMENT_CONTEXT_START.trim())
)))
});
if !has_environment_context {
input_with_instructions.push(ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText { text: ec }],
});
}
}
if let Some(ui) = self.get_formatted_user_instructions() {
let has_user_instructions = self.input.iter().any(|item| {
matches!(item, ResponseItem::Message { role, content, .. }
if role == "user"
&& content.iter().any(|c| matches!(c,
ContentItem::InputText { text } if text.contains(USER_INSTRUCTIONS_START)
)))
});
if !has_user_instructions {
input_with_instructions.push(ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText { text: ui }],
});
}
}
}
// Deduplicate function call outputs before adding to input
let mut seen_call_ids = std::collections::HashSet::new();
@@ -149,10 +195,10 @@ impl Prompt {
// Add status items at the end so they're fresh for each request
input_with_instructions.extend(self.status_items.clone());
// Limit screenshots to maximum 5 (keep first and last 4)
limit_screenshots_in_input(&mut input_with_instructions);
input_with_instructions
}
@@ -449,7 +495,7 @@ mod tests {
tool_choice: "auto",
parallel_tool_calls: false,
reasoning: None,
store: true,
store: false,
stream: true,
include: vec![],
prompt_cache_key: None,
@@ -477,7 +523,7 @@ mod tests {
tool_choice: "auto",
parallel_tool_calls: false,
reasoning: None,
store: true,
store: false,
stream: true,
include: vec![],
prompt_cache_key: None,
codex-rs/core/src/codex.rs
+259 -245
View File
@@ -20,7 +20,7 @@ use codex_apply_patch::maybe_parse_apply_patch_verified;
// unused: AuthManager
// unused: ConversationHistoryResponseEvent
use codex_protocol::protocol::TurnAbortReason;
// unused: TurnAbortedEvent
use codex_protocol::protocol::TurnAbortedEvent;
use futures::prelude::*;
use mcp_types::CallToolResult;
use serde::Serialize;
@@ -39,6 +39,11 @@ use crate::agent_tool::AgentStatusUpdatePayload;
use crate::protocol::WebSearchBeginEvent;
use crate::protocol::WebSearchCompleteEvent;
use codex_protocol::models::WebSearchAction;
use codex_protocol::protocol::RolloutItem;
mod compact;
use self::compact::build_compacted_history;
use self::compact::collect_user_messages;
/// Initial submission ID for session configuration
pub(crate) const INITIAL_SUBMIT_ID: &str = "";
@@ -97,6 +102,27 @@ impl ConfirmGuardPatternRuntime {
}
}
trait MutexExt<T> {
fn lock_unchecked(&self) -> std::sync::MutexGuard<'_, T>;
}
impl<T> MutexExt<T> for Mutex<T> {
fn lock_unchecked(&self) -> std::sync::MutexGuard<'_, T> {
#[expect(clippy::expect_used)]
self.lock().expect("poisoned lock")
}
}
#[derive(Clone)]
pub(crate) struct TurnContext {
pub(crate) client: ModelClient,
pub(crate) cwd: PathBuf,
pub(crate) base_instructions: Option<String>,
pub(crate) user_instructions: Option<String>,
pub(crate) approval_policy: AskForApproval,
pub(crate) sandbox_policy: SandboxPolicy,
}
/// Gather ephemeral, per-turn context that should not be persisted to history.
/// Combines environment info and (when enabled) a live browser snapshot and status.
struct EphemeralJar {
@@ -426,6 +452,7 @@ use crate::client::ModelClient;
use crate::client_common::Prompt;
use crate::client_common::ResponseEvent;
use crate::environment_context::EnvironmentContext;
use crate::user_instructions::UserInstructions;
use crate::config::{persist_model_selection, Config};
use crate::config_types::ShellEnvironmentPolicy;
use crate::conversation_history::ConversationHistory;
@@ -600,7 +627,7 @@ impl Codex {
#[derive(Default)]
struct State {
approved_commands: HashSet<Vec<String>>,
current_agent: Option<AgentAgent>,
current_task: Option<AgentTask>,
pending_approvals: HashMap<String, oneshot::Sender<ReviewDecision>>,
pending_input: Vec<ResponseInputItem>,
history: ConversationHistory,
@@ -615,6 +642,7 @@ struct State {
event_seq_by_sub_id: HashMap<String, u64>,
/// 1-based ordinal of the current HTTP request attempt in this session.
request_ordinal: u64,
next_internal_sub_id: u64,
}
/// Buffers partial turn progress produced during a single HTTP streaming attempt.
@@ -796,23 +824,30 @@ impl Session {
}
impl Session {
pub fn set_agent(&self, agent: AgentAgent) {
pub fn set_task(&self, agent: AgentTask) {
let mut state = self.state.lock().unwrap();
if let Some(current_agent) = state.current_agent.take() {
current_agent.abort(TurnAbortReason::Replaced);
if let Some(current_task) = state.current_task.take() {
current_task.abort(TurnAbortReason::Replaced);
}
state.current_agent = Some(agent);
state.current_task = Some(agent);
}
pub fn remove_agent(&self, sub_id: &str) {
pub fn remove_task(&self, sub_id: &str) {
let mut state = self.state.lock().unwrap();
if let Some(agent) = &state.current_agent {
if let Some(agent) = &state.current_task {
if agent.sub_id == sub_id {
state.current_agent.take();
state.current_task.take();
}
}
}
pub(crate) fn next_internal_sub_id(&self) -> String {
let mut state = self.state.lock().unwrap();
let id = state.next_internal_sub_id;
state.next_internal_sub_id = state.next_internal_sub_id.saturating_add(1);
format!("auto-compact-{id}")
}
/// Sends the given event to the client and swallows the send error, if
/// any, logging it as an error.
pub(crate) async fn send_event(&self, event: Event) {
@@ -885,6 +920,17 @@ impl Session {
state.request_ordinal
}
fn make_turn_context(&self) -> Arc<TurnContext> {
Arc::new(TurnContext {
client: self.client.clone(),
cwd: self.cwd.clone(),
base_instructions: self.base_instructions.clone(),
user_instructions: self.user_instructions.clone(),
approval_policy: self.approval_policy,
sandbox_policy: self.sandbox_policy.clone(),
})
}
pub async fn request_command_approval(
&self,
sub_id: String,
@@ -1102,6 +1148,25 @@ impl Session {
if let Err(e) = rec.record_state(snapshot).await {
error!("failed to record rollout state: {e:#}");
}
let rollout_items: Vec<RolloutItem> = items
.iter()
.cloned()
.map(RolloutItem::ResponseItem)
.collect();
if !rollout_items.is_empty() {
if let Err(e) = rec.record_items(&rollout_items).await {
error!("failed to record rollout items: {e:#}");
}
}
}
}
pub(crate) async fn persist_rollout_items(&self, items: &[RolloutItem]) {
let recorder = {
let guard = self.rollout.lock().unwrap();
guard.as_ref().cloned()
};
if let Some(rec) = recorder {
if let Err(e) = rec.record_items(items).await {
error!("failed to record rollout items: {e:#}");
}
@@ -1375,6 +1440,8 @@ impl Session {
// Concatenate filtered history with current turn's extras (which includes current ephemeral images)
let result = [filtered_history, extra].concat();
debug_history("turn_input_with_history", &result);
// Count total images in result for debugging
let total_images = result
.iter()
@@ -1396,10 +1463,49 @@ impl Session {
result
}
pub(crate) fn build_initial_context(&self, turn_context: &TurnContext) -> Vec<ResponseItem> {
let mut items = Vec::new();
if let Some(user_instructions) = turn_context.user_instructions.as_deref() {
items.push(UserInstructions::new(user_instructions.to_string()).into());
}
items.push(ResponseItem::from(EnvironmentContext::new(
Some(turn_context.cwd.clone()),
Some(turn_context.approval_policy),
Some(turn_context.sandbox_policy.clone()),
Some(self.user_shell.clone()),
)));
items
}
pub(crate) fn reconstruct_history_from_rollout(
&self,
turn_context: &TurnContext,
rollout_items: &[RolloutItem],
) -> Vec<ResponseItem> {
let mut history = self.build_initial_context(turn_context);
for item in rollout_items {
match item {
RolloutItem::ResponseItem(response_item) => {
history.push(response_item.clone());
}
RolloutItem::Compacted(compacted) => {
let user_messages = collect_user_messages(&history);
history = build_compacted_history(
self.build_initial_context(turn_context),
&user_messages,
&compacted.message,
);
}
_ => {}
}
}
history
}
/// Returns the input if there was no agent running to inject into
pub fn inject_input(&self, input: Vec<InputItem>) -> Result<(), Vec<InputItem>> {
let mut state = self.state.lock().unwrap();
if state.current_agent.is_some() {
if state.current_task.is_some() {
state
.pending_input
.push(response_input_from_core_items(input));
@@ -1449,8 +1555,8 @@ impl Session {
// `pending_input` by an earlier code path. Clearing it would drop the
// user's message and prevent the next turn from ever starting.
state.turn_scratchpad = None;
// Take current agent while holding the lock, then drop the lock BEFORE calling abort
let current = state.current_agent.take();
// Take current task while holding the lock, then drop the lock BEFORE calling abort
let current = state.current_task.take();
drop(state);
if let Some(agent) = current {
agent.abort(TurnAbortReason::Interrupted);
@@ -1512,6 +1618,7 @@ impl State {
// Preserve request_ordinal so reconfigurations (e.g., /reasoning)
// do not reset provider ordering mid-session.
request_ordinal: self.request_ordinal,
next_internal_sub_id: self.next_internal_sub_id,
..Default::default()
}
}
@@ -1533,15 +1640,28 @@ pub(crate) struct ApplyPatchCommandContext {
}
/// A series of Turns in response to user input.
pub(crate) struct AgentAgent {
pub(crate) struct AgentTask {
sess: Arc<Session>,
sub_id: String,
handle: AbortHandle,
}
impl AgentAgent {
fn spawn(sess: Arc<Session>, sub_id: String, input: Vec<InputItem>) -> Self {
let handle = tokio::spawn(run_agent(Arc::clone(&sess), sub_id.clone(), input)).abort_handle();
impl AgentTask {
fn spawn(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
sub_id: String,
input: Vec<InputItem>,
) -> Self {
let handle = {
let sess_clone = Arc::clone(&sess);
let tc_clone = Arc::clone(&turn_context);
let sub_clone = sub_id.clone();
tokio::spawn(async move {
run_agent(sess_clone, tc_clone, sub_clone, input).await;
})
.abort_handle()
};
Self {
sess,
sub_id,
@@ -1551,17 +1671,28 @@ impl AgentAgent {
fn compact(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
sub_id: String,
input: Vec<InputItem>,
compact_instructions: String,
) -> Self {
let handle = tokio::spawn(run_compact_agent(
Arc::clone(&sess),
sub_id.clone(),
input,
compact_instructions,
))
.abort_handle();
let handle = {
let sess_clone = Arc::clone(&sess);
let tc_clone = Arc::clone(&turn_context);
let sub_clone = sub_id.clone();
tokio::spawn(async move {
let _ = compact::perform_compaction(
sess_clone,
tc_clone,
sub_clone,
input,
compact_instructions,
true,
)
.await;
})
.abort_handle()
};
Self {
sess,
sub_id,
@@ -1569,16 +1700,15 @@ impl AgentAgent {
}
}
fn abort(self, _reason: TurnAbortReason) {
// TOCTOU?
fn abort(self, reason: TurnAbortReason) {
if !self.handle.is_finished() {
self.handle.abort();
let stamped = self
let event = self
.sess
.make_event(&self.sub_id, EventMsg::Error(ErrorEvent { message: "Turn interrupted".to_string() }));
let tx_event = self.sess.tx_event.clone();
.make_event(&self.sub_id, EventMsg::TurnAborted(TurnAbortedEvent { reason }));
let sess = self.sess.clone();
tokio::spawn(async move {
tx_event.send(stamped).await.ok();
sess.send_event(event).await;
});
}
}
@@ -1719,7 +1849,7 @@ async fn submission_loop(
config = Arc::clone(&new_config);
// Optionally resume an existing rollout.
let mut restored_items: Option<Vec<ResponseItem>> = None;
let mut restored_items: Option<Vec<RolloutItem>> = None;
let rollout_recorder: Option<RolloutRecorder> =
if let Some(path) = resume_path.as_ref() {
match RolloutRecorder::resume(&config, path).await {
@@ -1863,12 +1993,20 @@ async fn submission_loop(
last_screenshot_info: Mutex::new(None),
confirm_guard: ConfirmGuardRuntime::from_config(&config.confirm_guard),
}));
let mut replay_history_items: Option<Vec<ResponseItem>> = None;
// Patch restored state into the newly created session.
if let Some(sess_arc) = &sess {
if let Some(items) = &restored_items {
let mut st = sess_arc.state.lock().unwrap();
st.history.record_items(items.iter());
let turn_context = sess_arc.make_turn_context();
let reconstructed = sess_arc.reconstruct_history_from_rollout(&turn_context, items);
{
let mut st = sess_arc.state.lock().unwrap();
st.history = ConversationHistory::new();
st.history.record_items(reconstructed.iter());
}
replay_history_items = Some(reconstructed);
}
}
@@ -1896,7 +2034,7 @@ async fn submission_loop(
}
}
// If we resumed from a rollout, replay the prior transcript into the UI.
if let Some(items) = restored_items {
if let Some(items) = replay_history_items.clone() {
let event = sess_arc.make_event(
&sub.id,
EventMsg::ReplayHistory(crate::protocol::ReplayHistoryEvent { items }),
@@ -1905,7 +2043,7 @@ async fn submission_loop(
warn!("failed to send ReplayHistory event: {e}");
}
}
// Initialize agent manager after SessionConfigured is sent
if !agent_manager_initialized {
let mut manager = AGENT_MANAGER.write().await;
@@ -1947,12 +2085,13 @@ async fn submission_loop(
sess.cleanup_old_status_items().await;
// Abort synchronously here to avoid a race that can kill the
// newly spawned agent if the async abort runs after set_agent.
// newly spawned agent if the async abort runs after set_task.
sess.abort();
// Spawn a new agent for this user input.
let agent = AgentAgent::spawn(Arc::clone(sess), sub.id, items);
sess.set_agent(agent);
let turn_context = sess.make_turn_context();
let agent = AgentTask::spawn(Arc::clone(&sess), turn_context, sub.id.clone(), items);
sess.set_task(agent);
}
Op::ExecApproval { id, decision } => {
let sess = match sess.as_ref() {
@@ -2038,20 +2177,12 @@ async fn submission_loop(
}
};
// Create a summarization request as user input
const SUMMARIZATION_PROMPT: &str = include_str!("prompt_for_compact_command.md");
// Attempt to inject input into current agent
// Attempt to inject input into current task
if let Err(items) = sess.inject_input(vec![InputItem::Text {
text: "Start Summarization".to_string(),
}]) {
let agent = AgentAgent::compact(
sess.clone(),
sub.id,
items,
SUMMARIZATION_PROMPT.to_string(),
);
sess.set_agent(agent);
let turn_context = sess.make_turn_context();
compact::spawn_compact_task(sess.clone(), turn_context, sub.id.clone(), items);
}
}
Op::Shutdown => {
@@ -2114,7 +2245,7 @@ async fn submission_loop(
/// back to the model in the next turn.
/// - If the model sends only an assistant message, we record it in the
/// conversation history and consider the agent complete.
async fn run_agent(sess: Arc<Session>, sub_id: String, input: Vec<InputItem>) {
async fn run_agent(sess: Arc<Session>, turn_context: Arc<TurnContext>, sub_id: String, input: Vec<InputItem>) {
if input.is_empty() {
return;
}
@@ -2161,6 +2292,7 @@ async fn run_agent(sess: Arc<Session>, sub_id: String, input: Vec<InputItem>) {
.into_iter()
.map(ResponseItem::from)
.collect::<Vec<ResponseItem>>();
let pending_input_tail = pending_input.clone();
// Do not duplicate the initial input in `pending_input`.
// It is already recorded to history above; ephemeral items are appended separately.
@@ -2176,7 +2308,7 @@ async fn run_agent(sess: Arc<Session>, sub_id: String, input: Vec<InputItem>) {
// conversation history on each turn. The rollout file, however, should
// only record the new items that originated in this turn so that it
// represents an append-only log without duplicates.
let turn_input: Vec<ResponseItem> = sess.turn_input_with_history(pending_input);
let turn_input: Vec<ResponseItem> = sess.turn_input_with_history(pending_input_tail.clone());
let turn_input_messages: Vec<String> = turn_input
.iter()
@@ -2191,7 +2323,17 @@ async fn run_agent(sess: Arc<Session>, sub_id: String, input: Vec<InputItem>) {
})
})
.collect();
match run_turn(&sess, &mut turn_diff_tracker, sub_id.clone(), turn_input).await {
match run_turn(
&sess,
&turn_context,
&mut turn_diff_tracker,
sub_id.clone(),
initial_response_item.clone(),
pending_input_tail,
turn_input,
)
.await
{
Ok(turn_output) => {
let mut items_to_record_in_conversation_history = Vec::<ResponseItem>::new();
let mut responses = Vec::<ResponseInputItem>::new();
@@ -2325,7 +2467,7 @@ async fn run_agent(sess: Arc<Session>, sub_id: String, input: Vec<InputItem>) {
}
}
}
sess.remove_agent(&sub_id);
sess.remove_task(&sub_id);
let event = sess.make_event(
&sub_id,
EventMsg::TaskComplete(TaskCompleteEvent {
@@ -2342,14 +2484,18 @@ async fn run_agent(sess: Arc<Session>, sub_id: String, input: Vec<InputItem>) {
}
async fn run_turn(
sess: &Session,
sess: &Arc<Session>,
turn_context: &Arc<TurnContext>,
turn_diff_tracker: &mut TurnDiffTracker,
sub_id: String,
input: Vec<ResponseItem>,
initial_user_item: ResponseItem,
pending_input_tail: Vec<ResponseItem>,
mut input: Vec<ResponseItem>,
) -> CodexResult<Vec<ProcessedResponseItem>> {
// Check if browser is enabled
let browser_enabled = codex_browser::global::get_browser_manager().await.is_some();
let tc = &**turn_context;
let tools = get_openai_tools(
&sess.tools_config,
Some(sess.mcp_connection_manager.list_all_tools()),
@@ -2374,18 +2520,21 @@ async fn run_turn(
let prompt = Prompt {
input: attempt_input.clone(),
user_instructions: sess.user_instructions.clone(),
store: !sess.disable_response_storage,
tools: tools.clone(),
base_instructions_override: sess.base_instructions.clone(),
user_instructions: tc.user_instructions.clone(),
environment_context: Some(EnvironmentContext::new(
Some(sess.cwd.clone()),
Some(sess.approval_policy),
Some(sess.sandbox_policy.clone()),
Some(tc.cwd.clone()),
Some(tc.approval_policy),
Some(tc.sandbox_policy.clone()),
Some(sess.user_shell.clone()),
)),
tools: tools.clone(),
status_items, // Include status items with this request
base_instructions_override: tc.base_instructions.clone(),
include_additional_instructions: true,
text_format: None,
model_override: None,
model_family_override: None,
};
// Start a new scratchpad for this HTTP attempt
@@ -2425,8 +2574,6 @@ async fn run_turn(
if looks_like_context_overflow {
did_auto_compact = true;
// Inform UI and run a one-off compact turn inline, then retry
sess
.notify_stream_error(
&sub_id,
@@ -2435,62 +2582,34 @@ async fn run_turn(
)
.await;
const SUMMARIZATION_PROMPT: &str =
include_str!("prompt_for_compact_command.md");
let compacted_history = compact::run_inline_auto_compact_task(
Arc::clone(&sess),
Arc::clone(&turn_context),
)
.await;
let compact_input = response_input_from_core_items(vec![InputItem::Text {
text: "Start Summarization".to_string(),
}]);
let compact_turn_input: Vec<ResponseItem> =
sess.turn_input_with_history(vec![compact_input.clone().into()]);
// Reset any partial attempt state and rebuild the request payload using the
// newly compacted history plus the current user turn items.
sess.clear_scratchpad();
let compact_prompt = Prompt {
input: compact_turn_input,
user_instructions: None,
store: !sess.disable_response_storage,
tools: Vec::new(),
base_instructions_override: Some(SUMMARIZATION_PROMPT.to_string()),
environment_context: None,
status_items: Vec::new(),
text_format: None,
};
match drain_to_completed(sess, &sub_id, &compact_prompt).await {
Ok(()) => {
// Keep only the summary to shrink history
{
let mut state = sess.state.lock().unwrap();
state.history.keep_last_messages(1);
}
// Reset any partial attempt state and retry immediately
sess.clear_scratchpad();
sess
.notify_stream_error(
&sub_id,
"/compact completed; retrying with condensed history…"
.to_string(),
)
.await;
attempt_input = input.clone();
continue;
}
Err(err) => {
sess
.notify_stream_error(
&sub_id,
format!(
"/compact failed: {err}; falling back to normal retries…"
),
)
.await;
if compacted_history.is_empty() {
attempt_input = input.clone();
} else {
let mut rebuilt = compacted_history;
rebuilt.push(initial_user_item.clone());
if !pending_input_tail.is_empty() {
rebuilt.extend(pending_input_tail.iter().cloned());
}
input = rebuilt.clone();
attempt_input = rebuilt;
}
continue;
}
}
}
// Use the configured provider-specific stream retry budget.
let max_retries = sess.client.get_provider().stream_max_retries();
let max_retries = tc.client.get_provider().stream_max_retries();
if retries < max_retries {
retries += 1;
let delay = match e {
@@ -2809,102 +2928,6 @@ async fn try_run_turn(
}
}
async fn run_compact_agent(
sess: Arc<Session>,
sub_id: String,
input: Vec<InputItem>,
compact_instructions: String,
) {
let start_event = sess.make_event(&sub_id, EventMsg::TaskStarted);
if sess.tx_event.send(start_event).await.is_err() {
return;
}
let initial_input_for_turn: ResponseInputItem = response_input_from_core_items(input);
let turn_input: Vec<ResponseItem> =
sess.turn_input_with_history(vec![initial_input_for_turn.clone().into()]);
let max_retries = sess.client.get_provider().stream_max_retries();
let mut retries = 0;
loop {
// Bump request_ordinal for this provider request attempt so
// downstream OrderMeta carries the correct `req` index.
sess.begin_http_attempt();
// Build status items (screenshots, system status) fresh for each attempt
let status_items = build_turn_status_items(&sess).await;
let prompt = Prompt {
input: turn_input.clone(),
user_instructions: None,
store: !sess.disable_response_storage,
environment_context: None,
tools: Vec::new(),
base_instructions_override: Some(compact_instructions.clone()),
status_items, // Include status items with this request
text_format: None,
};
let attempt_result = drain_to_completed(&sess, &sub_id, &prompt).await;
match attempt_result {
Ok(()) => {
// Record status items to conversation history after successful turn
if !prompt.status_items.is_empty() {
sess.record_conversation_items(&prompt.status_items).await;
}
break;
}
Err(CodexErr::Interrupted) => return,
Err(e) => {
if retries < max_retries {
retries += 1;
let delay = backoff(retries);
sess.notify_stream_error(
&sub_id,
format!(
"stream error: {e}; retrying {retries}/{max_retries} in {delay:?}…"
),
)
.await;
tokio::time::sleep(delay).await;
continue;
} else {
let event = sess.make_event(
&sub_id,
EventMsg::Error(ErrorEvent { message: e.to_string() }),
);
sess.send_event(event).await;
// Ensure the UI is released from running state even on errors.
let done = sess.make_event(
&sub_id,
EventMsg::TaskComplete(TaskCompleteEvent { last_agent_message: None }),
);
sess.send_event(done).await;
return;
}
}
}
}
sess.remove_agent(&sub_id);
let event = sess.make_event(
&sub_id,
EventMsg::AgentMessage(AgentMessageEvent {
message: "Compact agent completed".to_string(),
}),
);
sess.send_event(event).await;
let event = sess.make_event(
&sub_id,
EventMsg::TaskComplete(TaskCompleteEvent { last_agent_message: None }),
);
sess.send_event(event).await;
let mut state = sess.state.lock().unwrap();
state.history.keep_last_messages(1);
}
async fn handle_response_item(
sess: &Session,
turn_diff_tracker: &mut TurnDiffTracker,
@@ -3866,7 +3889,7 @@ async fn handle_web_fetch(sess: &Session, ctx: &ToolCallCtx, arguments: String)
return ResponseInputItem::FunctionCallOutput { call_id: call_id_clone, output: FunctionCallOutputPayload { content: body.to_string(), success: Some(true) } };
}
}
// If JS extraction failed, try a CDP outerHTML of root
let root = browser_manager.execute_cdp("DOM.getDocument", json!({"depth": 1})).await.ok();
if let Some(root) = root {
@@ -5964,42 +5987,6 @@ fn get_last_assistant_message_from_turn(responses: &[ResponseItem]) -> Option<St
})
}
async fn drain_to_completed(sess: &Session, sub_id: &str, prompt: &Prompt) -> CodexResult<()> {
let mut stream = sess.client.clone().stream(prompt).await?;
loop {
let maybe_event = stream.next().await;
let Some(event) = maybe_event else {
return Err(CodexErr::Stream(
"stream closed before response.completed".into(),
None,
));
};
match event {
Ok(ResponseEvent::OutputItemDone { item, sequence_number: _, output_index: _ }) => {
// Record only to in-memory conversation history; avoid state snapshot.
let mut state = sess.state.lock().unwrap();
state.history.record_items(std::slice::from_ref(&item));
}
Ok(ResponseEvent::Completed {
response_id: _,
token_usage,
}) => {
// some providers don't return token usage, so we default
// TODO: consider approximate token usage
let token_usage = token_usage.unwrap_or_default();
sess.tx_event
.send(sess.make_event(&sub_id, EventMsg::TokenCount(token_usage)))
.await
.ok();
return Ok(());
}
Ok(_) => continue,
Err(e) => return Err(e),
}
}
}
/// Capture a screenshot from the browser and store it for the next model request
async fn capture_browser_screenshot(_sess: &Session) -> Result<(PathBuf, String), String> {
let browser_manager = codex_browser::global::get_browser_manager()
@@ -6958,7 +6945,7 @@ async fn handle_browser_console(sess: &Session, ctx: &ToolCallCtx, arguments: St
let message = log_obj.get("message")
.and_then(|v| v.as_str())
.unwrap_or("");
output.push_str(&format!("[{}] [{}] {}\n", timestamp, level.to_uppercase(), message));
}
}
@@ -7383,3 +7370,30 @@ async fn handle_browser_history(sess: &Session, ctx: &ToolCallCtx, arguments: St
)
.await
}
fn debug_history(label: &str, items: &[ResponseItem]) {
let preview: Vec<String> = items
.iter()
.enumerate()
.map(|(idx, item)| match item {
ResponseItem::Message { role, content, .. } => {
let text = content
.iter()
.filter_map(|c| match c {
ContentItem::InputText { text }
| ContentItem::OutputText { text } => Some(text.as_str()),
_ => None,
})
.collect::<Vec<_>>()
.join(" ");
let snippet: String = text.chars().take(80).collect();
format!("{idx}:{role}:{snippet}")
}
_ => format!("{idx}:{:?}", item),
})
.collect();
let rendered = preview.join(" | ");
if std::env::var_os("CODEX_COMPACT_TRACE").is_some() {
eprintln!("[compact_history] {} => [{}]", label, rendered);
}
info!(target = "codex_core::compact_history", "{} => [{}]", label, rendered);
}
codex-rs/core/src/codex/compact.rs
+80 -66
View File
@@ -4,30 +4,30 @@ use super::AgentTask;
use super::MutexExt;
use super::Session;
use super::TurnContext;
use super::debug_history;
use super::get_last_assistant_message_from_turn;
use super::response_input_from_core_items;
use crate::Prompt;
use crate::client_common::ResponseEvent;
use crate::error::CodexErr;
use crate::error::Result as CodexResult;
use crate::model_family::find_family_for_model;
use crate::protocol::AgentMessageEvent;
use crate::protocol::CompactedItem;
use crate::protocol::ErrorEvent;
use crate::protocol::Event;
use crate::protocol::EventMsg;
use crate::protocol::InputItem;
use crate::protocol::InputMessageKind;
use crate::protocol::TaskCompleteEvent;
use crate::protocol::TaskStartedEvent;
use crate::protocol::TurnContextItem;
use crate::util::backoff;
use askama::Template;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::CompactedItem;
use codex_protocol::protocol::InputMessageKind;
use codex_protocol::protocol::RolloutItem;
use futures::prelude::*;
pub(super) const COMPACT_TRIGGER_TEXT: &str = "Start Summarization";
pub(super) const COMPACT_TRIGGER_TEXT: &str =
"Start Summarization. Only include details found in the prior conversation.";
const SUMMARIZATION_PROMPT: &str = include_str!("../../templates/compact/prompt.md");
#[derive(Template)]
@@ -56,12 +56,12 @@ pub(super) fn spawn_compact_task(
pub(super) async fn run_inline_auto_compact_task(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
) {
) -> Vec<ResponseItem> {
let sub_id = sess.next_internal_sub_id();
let input = vec![InputItem::Text {
text: COMPACT_TRIGGER_TEXT.to_string(),
}];
run_compact_task_inner(
perform_compaction(
sess,
turn_context,
sub_id,
@@ -69,67 +69,50 @@ pub(super) async fn run_inline_auto_compact_task(
SUMMARIZATION_PROMPT.to_string(),
false,
)
.await;
.await
}
pub(super) async fn run_compact_task(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
sub_id: String,
input: Vec<InputItem>,
compact_instructions: String,
) {
run_compact_task_inner(
sess,
turn_context,
sub_id,
input,
compact_instructions,
true,
)
.await;
}
async fn run_compact_task_inner(
pub(super) async fn perform_compaction(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
sub_id: String,
input: Vec<InputItem>,
compact_instructions: String,
remove_task_on_completion: bool,
) {
let model_context_window = turn_context.client.get_model_context_window();
let start_event = Event {
id: sub_id.clone(),
msg: EventMsg::TaskStarted(TaskStartedEvent {
model_context_window,
}),
};
) -> Vec<ResponseItem> {
sess.notify_background_event(&sub_id, "Compacting conversation...")
.await;
let start_event = sess.make_event(&sub_id, EventMsg::TaskStarted);
sess.send_event(start_event).await;
let initial_input_for_turn: ResponseInputItem = ResponseInputItem::from(input);
let initial_input_for_turn = response_input_from_core_items(input);
let instructions_override = compact_instructions;
let turn_input = sess.turn_input_with_history(vec![initial_input_for_turn.clone().into()]);
let prompt = Prompt {
let mut prompt = Prompt {
input: turn_input,
store: !sess.disable_response_storage,
user_instructions: None,
environment_context: None,
tools: Vec::new(),
status_items: Vec::new(),
base_instructions_override: Some(instructions_override),
include_additional_instructions: false,
text_format: None,
model_override: None,
model_family_override: None,
};
if turn_context.client.get_model() == "gpt-5-codex" {
prompt.model_override = Some("gpt-5".to_string());
if let Some(family) = find_family_for_model("gpt-5") {
prompt.model_family_override = Some(family);
}
}
let max_retries = turn_context.client.get_provider().stream_max_retries();
let mut retries = 0;
let rollout_item = RolloutItem::TurnContext(TurnContextItem {
cwd: turn_context.cwd.clone(),
approval_policy: turn_context.approval_policy,
sandbox_policy: turn_context.sandbox_policy.clone(),
model: turn_context.client.get_model(),
effort: turn_context.client.get_reasoning_effort(),
summary: turn_context.client.get_reasoning_summary(),
});
sess.persist_rollout_items(&[rollout_item]).await;
loop {
let attempt_result = drain_to_completed(&sess, turn_context.as_ref(), &prompt).await;
@@ -138,7 +121,7 @@ async fn run_compact_task_inner(
break;
}
Err(CodexErr::Interrupted) => {
return;
return Vec::new();
}
Err(e) => {
if retries < max_retries {
@@ -154,14 +137,14 @@ async fn run_compact_task_inner(
tokio::time::sleep(delay).await;
continue;
} else {
let event = Event {
id: sub_id.clone(),
msg: EventMsg::Error(ErrorEvent {
let event = sess.make_event(
&sub_id,
EventMsg::Error(ErrorEvent {
message: e.to_string(),
}),
};
);
sess.send_event(event).await;
return;
return Vec::new();
}
}
}
@@ -180,7 +163,13 @@ async fn run_compact_task_inner(
let new_history = build_compacted_history(initial_context, &user_messages, &summary_text);
{
let mut state = sess.state.lock_unchecked();
state.history.replace(new_history);
state.history = super::ConversationHistory::new();
}
sess.record_conversation_items(&new_history).await;
{
let state = sess.state.lock_unchecked();
let snapshot = state.history.contents();
debug_history("after_compact_record", &snapshot);
}
let rollout_item = RolloutItem::Compacted(CompactedItem {
@@ -188,20 +177,45 @@ async fn run_compact_task_inner(
});
sess.persist_rollout_items(&[rollout_item]).await;
let event = Event {
id: sub_id.clone(),
msg: EventMsg::AgentMessage(AgentMessageEvent {
message: "Compact task completed".to_string(),
}),
let message = if summary_text.trim().is_empty() {
"Compact task completed.".to_string()
} else {
summary_text.clone()
};
let event = sess.make_event(
&sub_id,
EventMsg::AgentMessage(AgentMessageEvent {
message: message.clone(),
}),
);
sess.send_event(event).await;
let event = Event {
id: sub_id.clone(),
msg: EventMsg::TaskComplete(TaskCompleteEvent {
let assistant_summary = ResponseItem::Message {
id: None,
role: "assistant".to_string(),
content: vec![ContentItem::OutputText { text: message }],
};
{
let mut state = sess.state.lock_unchecked();
state
.history
.record_items(std::slice::from_ref(&assistant_summary));
let snapshot = state.history.contents();
debug_history("after_compact_summary", &snapshot);
}
sess.persist_rollout_items(&[RolloutItem::ResponseItem(assistant_summary.clone())])
.await;
let event = sess.make_event(
&sub_id,
EventMsg::TaskComplete(TaskCompleteEvent {
last_agent_message: None,
}),
};
);
sess.send_event(event).await;
{
let state = sess.state.lock_unchecked();
state.history.contents()
}
}
fn content_items_to_text(content: &[ContentItem]) -> Option<String> {
@@ -289,7 +303,7 @@ async fn drain_to_completed(
));
};
match event {
Ok(ResponseEvent::OutputItemDone(item)) => {
Ok(ResponseEvent::OutputItemDone { item, .. }) => {
let mut state = sess.state.lock_unchecked();
state.history.record_items(std::slice::from_ref(&item));
}
codex-rs/core/src/conversation_manager.rs
+4 -11
View File
@@ -172,18 +172,11 @@ impl ConversationManager {
.await
.map_err(|e| CodexErr::Io(e))?;
// Extract only response items to seed the rollout file.
let response_items: Vec<ResponseItem> = history
.get_rollout_items()
.into_iter()
.filter_map(|ri| match ri {
codex_protocol::protocol::RolloutItem::ResponseItem(item) => Some(item),
_ => None,
})
.collect();
if !response_items.is_empty() {
// Persist rollout items to seed the resumed conversation.
let rollout_items = history.get_rollout_items();
if !rollout_items.is_empty() {
recorder
.record_items(&response_items)
.record_items(&rollout_items)
.await
.map_err(|e| CodexErr::Io(e))?;
}
codex-rs/core/src/lib.rs
+1
View File
@@ -70,6 +70,7 @@ pub mod shell;
pub mod spawn;
pub mod terminal;
mod tool_apply_patch;
mod user_instructions;
pub mod turn_diff_tracker;
pub use rollout::list::ConversationsPage;
pub use rollout::list::find_conversation_path_by_id_str;
codex-rs/core/src/prompt_for_compact_command.md
-5
View File
@@ -1,5 +0,0 @@
You have exceeded the maximum number of tokens, please stop coding and instead write a short memento message for the next agent. Your note should:
- Summarize what you finished and what still needs work. If there was a recent update_plan call, repeat its steps verbatim.
- List outstanding TODOs with file paths / line numbers so they're easy to find.
- Flag code that needs more tests (edge cases, performance, integration, etc.).
- Record any open bugs, quirks, or setup steps that will make it easier for the next agent to pick up where you left off.
codex-rs/core/src/rollout/policy.rs
+2 -2
View File
@@ -13,8 +13,8 @@ pub(crate) fn is_persisted_response_item(item: &RolloutItem) -> bool {
}
// Always persist session meta
RolloutItem::SessionMeta(_) => true,
// Do not persist variants not used by this fork.
RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) => false,
// Persist compacted summaries and turn context for accurate history reconstruction.
RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) => true,
}
}
codex-rs/core/src/rollout/recorder.rs
+5 -13
View File
@@ -28,7 +28,6 @@ use super::policy::is_persisted_response_item;
use crate::config::Config;
use crate::default_client::DEFAULT_ORIGINATOR;
use crate::git_info::collect_git_info;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::ResumedHistory;
use codex_protocol::protocol::RolloutItem;
@@ -43,7 +42,7 @@ pub struct SessionStateSnapshot {}
pub struct SavedSession {
pub session: SessionMeta,
#[serde(default)]
pub items: Vec<ResponseItem>,
pub items: Vec<RolloutItem>,
#[serde(default)]
pub state: SessionStateSnapshot,
pub session_id: uuid::Uuid,
@@ -166,14 +165,11 @@ impl RolloutRecorder {
Ok(Self { tx, rollout_path })
}
pub(crate) async fn record_items(&self, items: &[ResponseItem]) -> std::io::Result<()> {
pub(crate) async fn record_items(&self, items: &[RolloutItem]) -> std::io::Result<()> {
let mut filtered: Vec<RolloutItem> = Vec::new();
for item in items {
// Note that function calls may look a bit strange if they are
// "fully qualified MCP tool calls," so we could consider
// reformatting them in that case.
if super::policy::should_persist_response_item(item) {
filtered.push(RolloutItem::ResponseItem(item.clone()));
if is_persisted_response_item(item) {
filtered.push(item.clone());
}
}
if filtered.is_empty() {
@@ -195,11 +191,7 @@ impl RolloutRecorder {
let recorder = Self::new(config, RolloutRecorderParams::Resume { path: path.to_path_buf() }).await?;
let history = Self::get_rollout_history(path).await?;
let (session_id, items) = match history {
InitialHistory::Resumed(resumed) => (resumed.conversation_id.0, resumed
.history
.into_iter()
.filter_map(|ri| match ri { RolloutItem::ResponseItem(it) => Some(it), _ => None })
.collect::<Vec<ResponseItem>>()),
InitialHistory::Resumed(resumed) => (resumed.conversation_id.0, resumed.history),
_ => (uuid::Uuid::new_v4(), Vec::new()),
};
let saved = SavedSession {