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

fix(core/compact): harden auto-compact flow

Browse Source
This commit is contained in:
James Peter
2026-01-29 23:30:16 +10:00
parent f04f91aee9
commit 654d0bba89
3 changed files with 142 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
code-rs/code-auto-drive-core/src/auto_coordinator.rs
+22 -1
View File
@@ -3336,7 +3336,28 @@ fn maybe_compact(
});
}
if apply_compaction(conversation, bounds, prev_summary, checkpoint.message).is_none() {
let summary_message = checkpoint.message;
let mut applied = apply_compaction(
conversation,
bounds,
prev_summary,
summary_message.clone(),
)
.is_some();
if !applied {
if let Some(retry_bounds) = compute_slice_bounds(conversation)
.filter(|retry_bounds| *retry_bounds != bounds)
{
applied = apply_compaction(
conversation,
retry_bounds,
prev_summary,
summary_message,
)
.is_some();
}
}
if !applied {
warn!("[Auto coordinator] apply_compaction returned None; bounds={bounds:?}");
event_tx.send(AutoCoordinatorEvent::Thinking {
delta: "Failed to compact history because the conversation changed while applying the summary. Continuing without compaction.".to_string(),
code-rs/core/src/codex/compact.rs
+31 -18
View File
@@ -42,6 +42,7 @@ const COMPACT_TOOL_ARGS_MAX_BYTES: usize = 4 * 1024;
const COMPACT_TOOL_OUTPUT_MAX_BYTES: usize = 4 * 1024;
const COMPACT_IMAGE_URL_MAX_BYTES: usize = 512;
const MAX_COMPACTION_SNIPPETS: usize = 12;
const COMPACT_STREAM_TIMEOUT: Duration = Duration::from_secs(120);
/// Determine whether to use remote compaction (ChatGPT-based) or local compaction.
///
@@ -799,26 +800,38 @@ async fn drain_to_completed(
prompt: &Prompt,
) -> CodexResult<()> {
let mut stream = turn_context.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,
None,
));
};
match event {
Ok(ResponseEvent::OutputItemDone { item, .. }) => {
let mut state = sess.state.lock().unwrap();
state.history.record_items(std::slice::from_ref(&item));
let result = tokio::time::timeout(COMPACT_STREAM_TIMEOUT, async {
loop {
let maybe_event = stream.next().await;
let Some(event) = maybe_event else {
return Err(CodexErr::Stream(
"stream closed before response.completed".into(),
None,
None,
));
};
match event {
Ok(ResponseEvent::OutputItemDone { item, .. }) => {
let mut state = sess.state.lock().unwrap();
state.history.record_items(std::slice::from_ref(&item));
}
Ok(ResponseEvent::Completed { .. }) => {
return Ok(());
}
Ok(_) => continue,
Err(e) => return Err(e),
}
Ok(ResponseEvent::Completed { .. }) => {
return Ok(());
}
Ok(_) => continue,
Err(e) => return Err(e),
}
})
.await;
match result {
Ok(res) => res,
Err(_) => Err(CodexErr::Stream(
"compaction stream timed out".into(),
None,
None,
)),
}
}
code-rs/core/src/codex/streaming.rs
+89 -37
View File
@@ -1532,6 +1532,7 @@ async fn run_agent(sess: Arc<Session>, turn_context: Arc<TurnContext>, sub_id: S
// infinite loops. As long as compaction works well in getting us way below
// the token limit, we shouldn't need more than one compaction per iteration.
let mut did_proactive_compact_this_iteration = false;
let mut auto_compact_pending = false;
loop {
// Note that pending_input would be something like a message the user
@@ -1570,6 +1571,12 @@ async fn run_agent(sess: Arc<Session>, turn_context: Arc<TurnContext>, sub_id: S
}
}
let compact_snapshot = if auto_compact_pending && !is_review_mode {
Some(sess.turn_input_with_history(pending_input_tail.clone()))
} else {
None
};
// Do not duplicate the initial input in `pending_input`.
// It is already recorded to history above; ephemeral items are appended separately.
if first_iteration {
@@ -1579,6 +1586,45 @@ async fn run_agent(sess: Arc<Session>, turn_context: Arc<TurnContext>, sub_id: S
sess.record_conversation_items(&pending_input).await;
}
if auto_compact_pending && !is_review_mode {
let compacted_history = if compact::should_use_remote_compact_task(&sess).await {
run_inline_remote_auto_compact_task(
Arc::clone(&sess),
Arc::clone(&turn_context),
Vec::new(),
)
.await
} else {
compact::run_inline_auto_compact_task(
Arc::clone(&sess),
Arc::clone(&turn_context),
)
.await
};
if !compacted_history.is_empty() {
let mut rebuilt = compacted_history;
if !pending_input_tail.is_empty() {
let previous_input_snapshot = compact_snapshot.unwrap_or_default();
let (missing_calls, filtered_outputs) = reconcile_pending_tool_outputs(
&pending_input_tail,
&rebuilt,
&previous_input_snapshot,
);
if !missing_calls.is_empty() {
rebuilt.extend(missing_calls);
}
if !filtered_outputs.is_empty() {
rebuilt.extend(filtered_outputs);
}
}
sess.replace_history(rebuilt);
pending_input_tail.clear();
did_proactive_compact_this_iteration = true;
}
auto_compact_pending = false;
}
// Construct the input that we will send to the model. When using the
// Chat completions API (or ZDR clients), the model needs the full
// conversation history on each turn. The rollout file, however, should
@@ -1748,43 +1794,6 @@ async fn run_agent(sess: Arc<Session>, turn_context: Arc<TurnContext>, sub_id: S
let token_limit_reached = most_recent_usage_tokens
.map_or(false, |tokens| tokens >= limit);
// As long as compaction works well in getting us way below the token limit,
// we shouldn't worry about being in an infinite loop. However, guard against
// repeated compaction attempts within a single iteration.
if token_limit_reached && !did_proactive_compact_this_iteration {
did_proactive_compact_this_iteration = true;
let attempt_req = sess.current_request_ordinal();
let order = sess.next_background_order(&sub_id, attempt_req, None);
sess
.notify_background_event_with_order(
&sub_id,
order,
"Token limit reached; running /compact and continuing…".to_string(),
)
.await;
// Choose between local and remote compact based on auth mode,
// matching upstream codex-rs behavior
if compact::should_use_remote_compact_task(&sess).await {
let _ = run_inline_remote_auto_compact_task(
Arc::clone(&sess),
Arc::clone(&turn_context),
Vec::new(),
)
.await;
} else {
let _ = compact::run_inline_auto_compact_task(
Arc::clone(&sess),
Arc::clone(&turn_context),
)
.await;
}
// Restart this loop with the newly compacted history so the
// next turn can see the trimmed conversation state.
continue;
}
// If there are responses, add them to pending input for the next iteration
if !responses.is_empty() {
if !is_review_mode {
@@ -1797,6 +1806,49 @@ async fn run_agent(sess: Arc<Session>, turn_context: Arc<TurnContext>, sub_id: S
did_proactive_compact_this_iteration = false;
}
// As long as compaction works well in getting us way below the token limit,
// we shouldn't worry about being in an infinite loop. However, guard against
// repeated compaction attempts within a single iteration.
if token_limit_reached && !did_proactive_compact_this_iteration && !is_review_mode {
let attempt_req = sess.current_request_ordinal();
let order = sess.next_background_order(&sub_id, attempt_req, None);
sess
.notify_background_event_with_order(
&sub_id,
order,
"Token limit reached; running /compact and continuing…".to_string(),
)
.await;
if responses.is_empty() {
did_proactive_compact_this_iteration = true;
// Choose between local and remote compact based on auth mode,
// matching upstream codex-rs behavior
if compact::should_use_remote_compact_task(&sess).await {
let _ = run_inline_remote_auto_compact_task(
Arc::clone(&sess),
Arc::clone(&turn_context),
Vec::new(),
)
.await;
} else {
let _ = compact::run_inline_auto_compact_task(
Arc::clone(&sess),
Arc::clone(&turn_context),
)
.await;
}
// Restart this loop with the newly compacted history so the
// next turn can see the trimmed conversation state.
continue;
}
if !auto_compact_pending {
auto_compact_pending = true;
}
}
if responses.is_empty() {
debug!("Turn completed");
last_task_message = get_last_assistant_message_from_turn(