mirror of
https://github.com/openai/codex.git
synced 2026-10-18 17:18:52 +00:00
Avoid cloning rollout history when truncating forks (#35982)
## What changed - Consume owned rollout vectors when truncating fork and subagent history, reusing their allocations instead of copying slices. - Project ephemeral fork responses and token-usage ownership before moving the source history into the new thread. - Preserve canonical `beforeTurnId` validation for rolled-back and synthetic legacy turns. ## Testing - Add truncation tests covering rolled-back and synthetic legacy `beforeTurnId` values. GitOrigin-RevId: 5785d004b74d7e057b82379a172676b023bcfb4f
This commit is contained in:
@@ -583,7 +583,7 @@ pub(super) async fn handle_pending_thread_resume_request(
|
||||
);
|
||||
let token_usage_turn_id = pending
|
||||
.include_turns
|
||||
.then(|| restored_token_usage_turn_id(&pending.history_items, &thread));
|
||||
.then(|| restored_token_usage_turn_id(&pending.history_items, thread.turns.as_slice()));
|
||||
let mut initial_turns_page = if let Some(mut page) = pending.paginated_initial_turns_page.take()
|
||||
{
|
||||
if let (Some(active_turn), Some(params)) =
|
||||
|
||||
@@ -3329,7 +3329,10 @@ impl ThreadRequestProcessor {
|
||||
config_snapshot.active_permission_profile,
|
||||
);
|
||||
let token_usage_turn_id = include_turns.then(|| {
|
||||
restored_token_usage_turn_id(response_history.get_rollout_items(), &thread)
|
||||
restored_token_usage_turn_id(
|
||||
response_history.get_rollout_items(),
|
||||
thread.turns.as_slice(),
|
||||
)
|
||||
});
|
||||
let mut initial_turns_page = if let Some(params) = initial_turns_page.as_ref() {
|
||||
let initial_turns_page_result = if paginated_resume {
|
||||
@@ -4177,10 +4180,45 @@ impl ThreadRequestProcessor {
|
||||
let parent_trace = self.request_trace_context(&request_id).await;
|
||||
let thread_source = thread_source.map(Into::into);
|
||||
|
||||
let (history_items, new_thread) = if let Some(prepared_fork) = prepared_fork {
|
||||
let history_items = Arc::clone(&prepared_fork.model_context);
|
||||
let new_thread = self
|
||||
.thread_manager
|
||||
let history_items = if prepared_fork.is_some() {
|
||||
source_history_items
|
||||
} else {
|
||||
let source_history_items = Arc::unwrap_or_clone(source_history_items);
|
||||
let history_items = match (last_turn_id.as_deref(), before_turn_id.as_deref()) {
|
||||
(Some(last_turn_id), None) => {
|
||||
truncate_rollout_after_turn_id(source_history_items, last_turn_id)
|
||||
.map_err(|err| core_thread_write_error("truncate thread for fork", err))?
|
||||
}
|
||||
(None, Some(before_turn_id)) => {
|
||||
truncate_rollout_before_turn_id(source_history_items, before_turn_id)
|
||||
.map_err(|err| core_thread_write_error("truncate thread for fork", err))?
|
||||
}
|
||||
(None, None) => source_history_items,
|
||||
(Some(_), Some(_)) => unreachable!("fork boundaries are mutually exclusive"),
|
||||
};
|
||||
Arc::new(history_items)
|
||||
};
|
||||
|
||||
let ephemeral_preview = if ephemeral {
|
||||
if paginated_source && last_turn_id.is_none() && before_turn_id.is_none() {
|
||||
source_thread.preview.clone()
|
||||
} else {
|
||||
preview_from_rollout_items(&history_items)
|
||||
}
|
||||
} else {
|
||||
String::new()
|
||||
};
|
||||
let ephemeral_turns = if ephemeral && include_turns {
|
||||
build_legacy_api_turns_from_rollout_items(&history_items)
|
||||
} else {
|
||||
Vec::new()
|
||||
};
|
||||
let ephemeral_token_usage_turn_id = (ephemeral && include_turns)
|
||||
.then(|| restored_token_usage_turn_id(&history_items, ephemeral_turns.as_slice()));
|
||||
let paginated_history_items = paginated_source.then(|| Arc::clone(&history_items));
|
||||
|
||||
let new_thread = if let Some(prepared_fork) = prepared_fork {
|
||||
self.thread_manager
|
||||
.fork_prepared_thread(
|
||||
config,
|
||||
prepared_fork,
|
||||
@@ -4188,37 +4226,22 @@ impl ThreadRequestProcessor {
|
||||
parent_trace,
|
||||
supports_openai_form_elicitation,
|
||||
)
|
||||
.await;
|
||||
(history_items, new_thread)
|
||||
.await
|
||||
} else {
|
||||
let history_items = match (last_turn_id.as_deref(), before_turn_id.as_deref()) {
|
||||
(Some(last_turn_id), None) => Arc::new(
|
||||
truncate_rollout_after_turn_id(&source_history_items, last_turn_id)
|
||||
.map_err(|err| core_thread_write_error("truncate thread for fork", err))?,
|
||||
),
|
||||
(None, Some(before_turn_id)) => Arc::new(
|
||||
truncate_rollout_before_turn_id(&source_history_items, before_turn_id)
|
||||
.map_err(|err| core_thread_write_error("truncate thread for fork", err))?,
|
||||
),
|
||||
(None, None) => Arc::clone(&source_history_items),
|
||||
(Some(_), Some(_)) => unreachable!("fork boundaries are mutually exclusive"),
|
||||
};
|
||||
let new_thread = self
|
||||
.thread_manager
|
||||
self.thread_manager
|
||||
.fork_thread_from_history(
|
||||
ForkSnapshot::Interrupted,
|
||||
config,
|
||||
InitialHistory::Resumed(ResumedHistory {
|
||||
conversation_id: source_thread_id,
|
||||
history: Arc::clone(&history_items),
|
||||
history: history_items,
|
||||
rollout_path: source_thread.rollout_path.clone(),
|
||||
}),
|
||||
thread_source,
|
||||
parent_trace,
|
||||
supports_openai_form_elicitation,
|
||||
)
|
||||
.await;
|
||||
(history_items, new_thread)
|
||||
.await
|
||||
};
|
||||
let NewThread {
|
||||
thread_id,
|
||||
@@ -4300,16 +4323,32 @@ impl ThreadRequestProcessor {
|
||||
let config_snapshot = forked_thread.config_snapshot().await;
|
||||
|
||||
// Persistent forks materialize their own rollout immediately. Ephemeral forks stay
|
||||
// pathless, so they rebuild their visible history from the copied source history instead.
|
||||
let mut thread = if session_configured.rollout_path.is_some() {
|
||||
// pathless, so their visible history is projected before the source history is consumed.
|
||||
let (mut thread, mut token_usage_turn_id) = if session_configured.rollout_path.is_some() {
|
||||
let stored_thread = self
|
||||
.read_stored_thread_for_new_fork(thread_id, include_turns && !paginated_source)
|
||||
.await?;
|
||||
self.stored_thread_to_api_thread(
|
||||
let (mut thread, history) = thread_from_stored_thread(
|
||||
stored_thread,
|
||||
fallback_model_provider.as_str(),
|
||||
include_turns && !paginated_source,
|
||||
)
|
||||
&self.config.cwd,
|
||||
);
|
||||
if include_turns && let Some(history) = history.as_ref() {
|
||||
populate_thread_turns_from_history(
|
||||
&mut thread,
|
||||
&history.items,
|
||||
/*active_turn*/ None,
|
||||
);
|
||||
}
|
||||
let token_usage_turn_id = include_turns.then(|| {
|
||||
restored_token_usage_turn_id(
|
||||
history
|
||||
.as_ref()
|
||||
.map_or(&[], |history| history.items.as_slice()),
|
||||
thread.turns.as_slice(),
|
||||
)
|
||||
});
|
||||
(thread, token_usage_turn_id)
|
||||
} else {
|
||||
let mut thread = build_thread_from_snapshot(
|
||||
thread_id,
|
||||
@@ -4318,24 +4357,19 @@ impl ThreadRequestProcessor {
|
||||
&config_snapshot,
|
||||
/*path*/ None,
|
||||
);
|
||||
thread.preview =
|
||||
if paginated_source && last_turn_id.is_none() && before_turn_id.is_none() {
|
||||
source_thread.preview.clone()
|
||||
} else {
|
||||
preview_from_rollout_items(&history_items)
|
||||
};
|
||||
thread.preview = ephemeral_preview;
|
||||
thread.forked_from_id = Some(source_thread_id.to_string());
|
||||
if include_turns {
|
||||
populate_thread_turns_from_history(
|
||||
&mut thread,
|
||||
&history_items,
|
||||
/*active_turn*/ None,
|
||||
);
|
||||
}
|
||||
thread
|
||||
thread.turns = ephemeral_turns;
|
||||
(thread, ephemeral_token_usage_turn_id)
|
||||
};
|
||||
if paginated_source && include_turns {
|
||||
thread.turns = self.paginated_thread_full_turns(thread_id).await?;
|
||||
token_usage_turn_id = Some(restored_token_usage_turn_id(
|
||||
paginated_history_items
|
||||
.as_deref()
|
||||
.map_or(&[], Vec::as_slice),
|
||||
thread.turns.as_slice(),
|
||||
));
|
||||
}
|
||||
if let Some(name) = source_thread_name {
|
||||
set_thread_name_from_title(&mut thread, name);
|
||||
@@ -4380,8 +4414,6 @@ impl ThreadRequestProcessor {
|
||||
|
||||
let notif = thread_started_notification(thread);
|
||||
let connection_id = request_id.connection_id;
|
||||
let token_usage_turn_id =
|
||||
include_turns.then(|| restored_token_usage_turn_id(&history_items, &response.thread));
|
||||
self.outgoing
|
||||
.send_response_with_thread_originator(request_id, response, thread_originator)
|
||||
.await;
|
||||
|
||||
@@ -12,7 +12,6 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use codex_app_server_protocol::ServerNotification;
|
||||
use codex_app_server_protocol::Thread;
|
||||
use codex_app_server_protocol::ThreadHistoryBuilder;
|
||||
use codex_app_server_protocol::ThreadTokenUsage;
|
||||
use codex_app_server_protocol::ThreadTokenUsageUpdatedNotification;
|
||||
@@ -58,10 +57,10 @@ pub(super) async fn send_thread_token_usage_update_to_connection(
|
||||
|
||||
pub(super) fn restored_token_usage_turn_id(
|
||||
rollout_items: &[RolloutItem],
|
||||
thread: &Thread,
|
||||
turns: &[Turn],
|
||||
) -> String {
|
||||
latest_token_usage_turn_id_from_rollout_items(rollout_items, thread.turns.as_slice())
|
||||
.unwrap_or_else(|| latest_token_usage_turn_id(thread))
|
||||
latest_token_usage_turn_id_from_rollout_items(rollout_items, turns)
|
||||
.unwrap_or_else(|| latest_token_usage_turn_id(turns))
|
||||
}
|
||||
|
||||
/// Identifies the turn that was active when the latest `TokenCount` record appeared.
|
||||
@@ -96,13 +95,12 @@ fn latest_token_usage_turn_id_from_rollout_items(
|
||||
/// Normal replay derives the owner from the rollout position of the latest
|
||||
/// `TokenCount` event. This fallback only preserves a stable wire shape for
|
||||
/// unusual histories where that rollout information cannot be read.
|
||||
fn latest_token_usage_turn_id(thread: &Thread) -> String {
|
||||
thread
|
||||
.turns
|
||||
fn latest_token_usage_turn_id(turns: &[Turn]) -> String {
|
||||
turns
|
||||
.iter()
|
||||
.rev()
|
||||
.find(|turn| matches!(turn.status, TurnStatus::Completed | TurnStatus::Failed))
|
||||
.or_else(|| thread.turns.last())
|
||||
.or_else(|| turns.last())
|
||||
.map(|turn| turn.id.clone())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
@@ -655,7 +655,7 @@ impl AgentControl {
|
||||
.unwrap_or_default();
|
||||
if let SpawnAgentForkMode::LastNTurns(last_n_turns) = fork_mode {
|
||||
forked_rollout_items =
|
||||
truncate_rollout_to_last_n_fork_turns(&forked_rollout_items, *last_n_turns);
|
||||
truncate_rollout_to_last_n_fork_turns(forked_rollout_items, *last_n_turns);
|
||||
}
|
||||
let multi_agent_v2_usage_hint_texts_to_filter: Vec<String> =
|
||||
if multi_agent_version == MultiAgentVersion::V2 {
|
||||
|
||||
@@ -1987,19 +1987,24 @@ fn truncate_before_nth_user_message(
|
||||
n: usize,
|
||||
snapshot_state: &SnapshotTurnState,
|
||||
) -> InitialHistory {
|
||||
let items = history.get_rollout_items().to_vec();
|
||||
let mut items = match history {
|
||||
InitialHistory::New | InitialHistory::Cleared => Vec::new(),
|
||||
InitialHistory::Resumed(resumed) => Arc::unwrap_or_clone(resumed.history),
|
||||
InitialHistory::Forked(items) => items,
|
||||
};
|
||||
let user_positions = truncation::user_message_positions_in_rollout(&items);
|
||||
let rolled = if snapshot_state.ends_mid_turn && n >= user_positions.len() {
|
||||
if let Some(cut_idx) = snapshot_state
|
||||
.active_turn_start_index
|
||||
.or_else(|| user_positions.last().copied())
|
||||
{
|
||||
items[..cut_idx].to_vec()
|
||||
items.truncate(cut_idx);
|
||||
items
|
||||
} else {
|
||||
items
|
||||
}
|
||||
} else {
|
||||
truncation::truncate_rollout_before_nth_user_message_from_start(&items, n)
|
||||
truncation::truncate_rollout_before_nth_user_message_from_start(items, n)
|
||||
};
|
||||
|
||||
if rolled.is_empty() {
|
||||
|
||||
@@ -134,23 +134,24 @@ pub(crate) fn fork_turn_positions_in_rollout(items: &[RolloutItem]) -> Vec<usize
|
||||
/// If fewer than or equal to `n_from_start` user messages exist, this returns the full
|
||||
/// rollout unchanged.
|
||||
pub(crate) fn truncate_rollout_before_nth_user_message_from_start(
|
||||
items: &[RolloutItem],
|
||||
mut items: Vec<RolloutItem>,
|
||||
n_from_start: usize,
|
||||
) -> Vec<RolloutItem> {
|
||||
if n_from_start == usize::MAX {
|
||||
return items.to_vec();
|
||||
return items;
|
||||
}
|
||||
|
||||
let user_positions = user_message_positions_in_rollout(items);
|
||||
let user_positions = user_message_positions_in_rollout(&items);
|
||||
|
||||
// If fewer than or equal to n user messages exist, keep the full rollout.
|
||||
if user_positions.len() <= n_from_start {
|
||||
return items.to_vec();
|
||||
return items;
|
||||
}
|
||||
|
||||
// Cut strictly before the nth user message (do not keep the nth itself).
|
||||
let cut_idx = user_positions[n_from_start];
|
||||
items[..cut_idx].to_vec()
|
||||
items.truncate(cut_idx);
|
||||
items
|
||||
}
|
||||
|
||||
/// Return a rollout prefix ending after the requested persisted terminal turn.
|
||||
@@ -160,10 +161,10 @@ pub(crate) fn truncate_rollout_before_nth_user_message_from_start(
|
||||
/// generated while projecting legacy rollouts are intentionally unsupported
|
||||
/// because they do not provide a stable raw rollout boundary for a fork.
|
||||
pub fn truncate_rollout_after_turn_id(
|
||||
items: &[RolloutItem],
|
||||
mut items: Vec<RolloutItem>,
|
||||
last_turn_id: &str,
|
||||
) -> CodexResult<Vec<RolloutItem>> {
|
||||
let turns = build_turns_from_rollout_items(items);
|
||||
let turns = build_turns_from_rollout_items(&items);
|
||||
let turn = turns
|
||||
.iter()
|
||||
.find(|turn| turn.id == last_turn_id)
|
||||
@@ -202,36 +203,54 @@ pub fn truncate_rollout_after_turn_id(
|
||||
matches!(item, RolloutItem::EventMsg(EventMsg::TurnStarted(_))).then_some(index)
|
||||
})
|
||||
.unwrap_or(items.len());
|
||||
Ok(items[..cut_index].to_vec())
|
||||
items.truncate(cut_index);
|
||||
Ok(items)
|
||||
}
|
||||
|
||||
/// Return a rollout prefix ending immediately before the requested persisted turn.
|
||||
pub fn truncate_rollout_before_turn_id(
|
||||
items: &[RolloutItem],
|
||||
mut items: Vec<RolloutItem>,
|
||||
before_turn_id: &str,
|
||||
) -> CodexResult<Vec<RolloutItem>> {
|
||||
let turns = build_turns_from_rollout_items(items);
|
||||
if !turns.iter().any(|turn| turn.id == before_turn_id) {
|
||||
let cut_index = items.iter().position(|item| {
|
||||
matches!(
|
||||
item,
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(event))
|
||||
if event.turn_id == before_turn_id
|
||||
)
|
||||
});
|
||||
|
||||
let Some(cut_index) = cut_index else {
|
||||
// Older rollouts can expose generated turn IDs without a TurnStarted item to fork at.
|
||||
if build_turns_from_rollout_items(&items)
|
||||
.iter()
|
||||
.any(|turn| turn.id == before_turn_id)
|
||||
{
|
||||
return Err(CodexErr::InvalidRequest(format!(
|
||||
"beforeTurnId '{before_turn_id}' is not a persisted canonical turn in the source thread"
|
||||
)));
|
||||
}
|
||||
|
||||
return Err(CodexErr::InvalidRequest(format!(
|
||||
"beforeTurnId '{before_turn_id}' was not found in the source thread"
|
||||
)));
|
||||
};
|
||||
|
||||
// A persisted turn boundary proves the turn exists unless a later rollback removes it.
|
||||
if items[cut_index + 1..]
|
||||
.iter()
|
||||
.any(|item| matches!(item, RolloutItem::EventMsg(EventMsg::ThreadRolledBack(_))))
|
||||
&& !build_turns_from_rollout_items(&items)
|
||||
.iter()
|
||||
.any(|turn| turn.id == before_turn_id)
|
||||
{
|
||||
return Err(CodexErr::InvalidRequest(format!(
|
||||
"beforeTurnId '{before_turn_id}' was not found in the source thread"
|
||||
)));
|
||||
}
|
||||
|
||||
let cut_index = items
|
||||
.iter()
|
||||
.position(|item| {
|
||||
matches!(
|
||||
item,
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(event))
|
||||
if event.turn_id == before_turn_id
|
||||
)
|
||||
})
|
||||
.ok_or_else(|| {
|
||||
CodexErr::InvalidRequest(format!(
|
||||
"beforeTurnId '{before_turn_id}' is not a persisted canonical turn in the source thread"
|
||||
))
|
||||
})?;
|
||||
Ok(items[..cut_index].to_vec())
|
||||
items.truncate(cut_index);
|
||||
Ok(items)
|
||||
}
|
||||
|
||||
/// Return a suffix of `items` that keeps the last `n_from_end` fork turns.
|
||||
@@ -239,14 +258,14 @@ pub fn truncate_rollout_before_turn_id(
|
||||
/// If fewer than or equal to `n_from_end` fork turns exist, this keeps from the first fork-turn
|
||||
/// boundary and still drops pre-turn startup context.
|
||||
pub(crate) fn truncate_rollout_to_last_n_fork_turns(
|
||||
items: &[RolloutItem],
|
||||
mut items: Vec<RolloutItem>,
|
||||
n_from_end: usize,
|
||||
) -> Vec<RolloutItem> {
|
||||
if n_from_end == 0 {
|
||||
return Vec::new();
|
||||
}
|
||||
|
||||
let fork_turn_positions = fork_turn_positions_in_rollout(items);
|
||||
let fork_turn_positions = fork_turn_positions_in_rollout(&items);
|
||||
let Some(keep_idx) = fork_turn_positions
|
||||
.len()
|
||||
.checked_sub(n_from_end)
|
||||
@@ -255,7 +274,7 @@ pub(crate) fn truncate_rollout_to_last_n_fork_turns(
|
||||
else {
|
||||
return Vec::new();
|
||||
};
|
||||
items[keep_idx..].to_vec()
|
||||
items.split_off(keep_idx)
|
||||
}
|
||||
|
||||
fn is_real_user_message_boundary(item: &ResponseItem) -> bool {
|
||||
|
||||
@@ -105,7 +105,7 @@ fn truncates_rollout_after_terminal_canonical_turn_id() {
|
||||
];
|
||||
|
||||
let truncated =
|
||||
truncate_rollout_after_turn_id(&rollout, "turn-2").expect("truncate through turn-2");
|
||||
truncate_rollout_after_turn_id(rollout.clone(), "turn-2").expect("truncate through turn-2");
|
||||
|
||||
assert_eq!(
|
||||
serde_json::to_value(&truncated).unwrap(),
|
||||
@@ -123,13 +123,13 @@ fn truncates_rollout_before_terminal_canonical_turn_id() {
|
||||
];
|
||||
|
||||
let truncated =
|
||||
truncate_rollout_before_turn_id(&rollout, "turn-2").expect("truncate before turn-2");
|
||||
truncate_rollout_before_turn_id(rollout.clone(), "turn-2").expect("truncate before turn-2");
|
||||
assert_eq!(
|
||||
serde_json::to_value(&truncated).unwrap(),
|
||||
serde_json::to_value(&rollout[..2]).unwrap()
|
||||
);
|
||||
assert!(
|
||||
truncate_rollout_before_turn_id(&rollout, "turn-1")
|
||||
truncate_rollout_before_turn_id(rollout, "turn-1")
|
||||
.expect("truncate before turn-1")
|
||||
.is_empty()
|
||||
);
|
||||
@@ -143,7 +143,7 @@ fn truncates_rollout_before_in_progress_canonical_turn_id() {
|
||||
turn_started("turn-2"),
|
||||
];
|
||||
|
||||
let truncated = truncate_rollout_before_turn_id(&rollout, "turn-2")
|
||||
let truncated = truncate_rollout_before_turn_id(rollout.clone(), "turn-2")
|
||||
.expect("truncate before in-progress turn-2");
|
||||
|
||||
assert_eq!(
|
||||
@@ -152,6 +152,50 @@ fn truncates_rollout_before_in_progress_canonical_turn_id() {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn truncate_rollout_before_turn_id_rejects_rolled_back_turn() {
|
||||
let rollout = vec![
|
||||
turn_started("turn-1"),
|
||||
turn_completed("turn-1"),
|
||||
turn_started("turn-2"),
|
||||
turn_completed("turn-2"),
|
||||
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
|
||||
num_turns: 1,
|
||||
})),
|
||||
turn_started("turn-3"),
|
||||
turn_completed("turn-3"),
|
||||
];
|
||||
|
||||
let err = truncate_rollout_before_turn_id(rollout, "turn-2")
|
||||
.expect_err("rolled-back turn should not be a fork anchor");
|
||||
|
||||
assert!(matches!(
|
||||
err.details(),
|
||||
CodexErrorDetails::InvalidRequest(message)
|
||||
if message == "beforeTurnId 'turn-2' was not found in the source thread"
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn truncate_rollout_before_turn_id_rejects_synthetic_legacy_turn_id() {
|
||||
let rollout = vec![RolloutItem::EventMsg(EventMsg::UserMessage(
|
||||
UserMessageEvent {
|
||||
message: "legacy".to_string(),
|
||||
..Default::default()
|
||||
},
|
||||
))];
|
||||
|
||||
let err = truncate_rollout_before_turn_id(rollout, "rollout-0")
|
||||
.expect_err("synthetic turn should not be a fork anchor");
|
||||
|
||||
assert!(matches!(
|
||||
err.details(),
|
||||
CodexErrorDetails::InvalidRequest(message)
|
||||
if message
|
||||
== "beforeTurnId 'rollout-0' is not a persisted canonical turn in the source thread"
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn truncate_rollout_after_turn_id_rejects_rolled_back_turn() {
|
||||
let rollout = vec![
|
||||
@@ -166,7 +210,7 @@ fn truncate_rollout_after_turn_id_rejects_rolled_back_turn() {
|
||||
turn_completed("turn-3"),
|
||||
];
|
||||
|
||||
let err = truncate_rollout_after_turn_id(&rollout, "turn-2")
|
||||
let err = truncate_rollout_after_turn_id(rollout, "turn-2")
|
||||
.expect_err("rolled-back turn should not be a fork anchor");
|
||||
|
||||
assert!(matches!(
|
||||
@@ -185,7 +229,7 @@ fn truncate_rollout_after_turn_id_rejects_synthetic_legacy_turn_id() {
|
||||
},
|
||||
))];
|
||||
|
||||
let err = truncate_rollout_after_turn_id(&rollout, "rollout-0")
|
||||
let err = truncate_rollout_after_turn_id(rollout, "rollout-0")
|
||||
.expect_err("synthetic turn should not be a fork anchor");
|
||||
|
||||
assert!(matches!(
|
||||
@@ -200,7 +244,7 @@ fn truncate_rollout_after_turn_id_rejects_synthetic_legacy_turn_id() {
|
||||
fn truncate_rollout_after_turn_id_rejects_in_progress_turn() {
|
||||
let rollout = vec![turn_started("turn-1")];
|
||||
|
||||
let err = truncate_rollout_after_turn_id(&rollout, "turn-1")
|
||||
let err = truncate_rollout_after_turn_id(rollout, "turn-1")
|
||||
.expect_err("in-progress turn should not be a fork anchor");
|
||||
|
||||
assert!(matches!(
|
||||
@@ -245,8 +289,10 @@ fn truncates_rollout_from_start_before_nth_user_only() {
|
||||
.map(RolloutItem::ResponseItem)
|
||||
.collect();
|
||||
|
||||
let truncated =
|
||||
truncate_rollout_before_nth_user_message_from_start(&rollout, /*n_from_start*/ 1);
|
||||
let truncated = truncate_rollout_before_nth_user_message_from_start(
|
||||
rollout.clone(),
|
||||
/*n_from_start*/ 1,
|
||||
);
|
||||
let expected = vec![
|
||||
RolloutItem::ResponseItem(items[0].clone()),
|
||||
RolloutItem::ResponseItem(items[1].clone()),
|
||||
@@ -257,8 +303,10 @@ fn truncates_rollout_from_start_before_nth_user_only() {
|
||||
serde_json::to_value(&expected).unwrap()
|
||||
);
|
||||
|
||||
let truncated2 =
|
||||
truncate_rollout_before_nth_user_message_from_start(&rollout, /*n_from_start*/ 2);
|
||||
let truncated2 = truncate_rollout_before_nth_user_message_from_start(
|
||||
rollout.clone(),
|
||||
/*n_from_start*/ 2,
|
||||
);
|
||||
assert_eq!(
|
||||
serde_json::to_value(&truncated2).unwrap(),
|
||||
serde_json::to_value(&rollout).unwrap()
|
||||
@@ -273,7 +321,8 @@ fn truncation_max_keeps_full_rollout() {
|
||||
RolloutItem::ResponseItem(user_msg("u2")),
|
||||
];
|
||||
|
||||
let truncated = truncate_rollout_before_nth_user_message_from_start(&rollout, usize::MAX);
|
||||
let truncated =
|
||||
truncate_rollout_before_nth_user_message_from_start(rollout.clone(), usize::MAX);
|
||||
|
||||
assert_eq!(
|
||||
serde_json::to_value(&truncated).unwrap(),
|
||||
@@ -300,7 +349,7 @@ fn truncates_rollout_from_start_applies_thread_rollback_markers() {
|
||||
// Effective user history after applying rollback(1) is: u1, u3, u4.
|
||||
// So n_from_start=2 should cut before u4 (not u3).
|
||||
let truncated = truncate_rollout_before_nth_user_message_from_start(
|
||||
&rollout_items,
|
||||
rollout_items.clone(),
|
||||
/*n_from_start*/ 2,
|
||||
);
|
||||
let expected = rollout_items[..7].to_vec();
|
||||
@@ -329,10 +378,8 @@ async fn ignores_session_prefix_messages_when_truncating_rollout_from_start() {
|
||||
.map(RolloutItem::ResponseItem)
|
||||
.collect();
|
||||
|
||||
let truncated = truncate_rollout_before_nth_user_message_from_start(
|
||||
&rollout_items,
|
||||
/*n_from_start*/ 1,
|
||||
);
|
||||
let truncated =
|
||||
truncate_rollout_before_nth_user_message_from_start(rollout_items, /*n_from_start*/ 1);
|
||||
let expected: Vec<RolloutItem> = vec![
|
||||
RolloutItem::ResponseItem(items[0].clone()),
|
||||
RolloutItem::ResponseItem(items[1].clone()),
|
||||
@@ -365,7 +412,7 @@ fn truncates_rollout_to_last_n_fork_turns_counts_trigger_turn_messages() {
|
||||
RolloutItem::ResponseItem(assistant_msg("a4")),
|
||||
];
|
||||
|
||||
let truncated = truncate_rollout_to_last_n_fork_turns(&rollout, /*n_from_end*/ 2);
|
||||
let truncated = truncate_rollout_to_last_n_fork_turns(rollout.clone(), /*n_from_end*/ 2);
|
||||
let expected = rollout[4..].to_vec();
|
||||
|
||||
assert_eq!(
|
||||
@@ -436,7 +483,7 @@ fn truncates_rollout_to_last_n_fork_turns_drops_startup_prefix_even_when_under_l
|
||||
RolloutItem::ResponseItem(assistant_msg("answer")),
|
||||
];
|
||||
|
||||
let truncated = truncate_rollout_to_last_n_fork_turns(&rollout, /*n_from_end*/ 2);
|
||||
let truncated = truncate_rollout_to_last_n_fork_turns(rollout.clone(), /*n_from_end*/ 2);
|
||||
let expected = rollout[1..].to_vec();
|
||||
|
||||
assert_eq!(
|
||||
@@ -462,7 +509,7 @@ fn truncates_rollout_to_last_n_fork_turns_applies_thread_rollback_markers() {
|
||||
RolloutItem::ResponseItem(assistant_msg("a3")),
|
||||
];
|
||||
|
||||
let truncated = truncate_rollout_to_last_n_fork_turns(&rollout, /*n_from_end*/ 2);
|
||||
let truncated = truncate_rollout_to_last_n_fork_turns(rollout.clone(), /*n_from_end*/ 2);
|
||||
|
||||
assert_eq!(
|
||||
serde_json::to_value(&truncated).unwrap(),
|
||||
@@ -504,7 +551,7 @@ fn truncates_rollout_to_last_n_fork_turns_discards_trigger_boundaries_in_rolled_
|
||||
RolloutItem::ResponseItem(assistant_msg("a2")),
|
||||
];
|
||||
|
||||
let truncated = truncate_rollout_to_last_n_fork_turns(&rollout, /*n_from_end*/ 2);
|
||||
let truncated = truncate_rollout_to_last_n_fork_turns(rollout.clone(), /*n_from_end*/ 2);
|
||||
|
||||
let expected = rollout[1..].to_vec();
|
||||
|
||||
@@ -534,7 +581,7 @@ fn truncates_rollout_to_last_n_fork_turns_discards_rolled_back_assistant_instruc
|
||||
RolloutItem::ResponseItem(assistant_msg("a3")),
|
||||
];
|
||||
|
||||
let truncated = truncate_rollout_to_last_n_fork_turns(&rollout, /*n_from_end*/ 1);
|
||||
let truncated = truncate_rollout_to_last_n_fork_turns(rollout.clone(), /*n_from_end*/ 1);
|
||||
let expected = rollout[5..].to_vec();
|
||||
|
||||
assert_eq!(
|
||||
@@ -555,7 +602,7 @@ fn truncates_rollout_to_last_n_fork_turns_keeps_full_rollout_when_n_is_large() {
|
||||
RolloutItem::ResponseItem(assistant_msg("a2")),
|
||||
];
|
||||
|
||||
let truncated = truncate_rollout_to_last_n_fork_turns(&rollout, /*n_from_end*/ 10);
|
||||
let truncated = truncate_rollout_to_last_n_fork_turns(rollout.clone(), /*n_from_end*/ 10);
|
||||
|
||||
assert_eq!(
|
||||
serde_json::to_value(&truncated).unwrap(),
|
||||
|
||||
Reference in New Issue
Block a user