feat: fork pattern v2 (#15771)

Adds this:
```
properties.insert(
            "fork_turns".to_string(),
            JsonSchema::String {
                description: Some(
                    "Optional MultiAgentV2 fork mode. Use `none`, `all`, or a positive integer string such as `3` to fork only the most recent turns."
                        .to_string(),
                ),
            },
        );
        ```

---------

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
jif-oaiandCodex authored and GitHub committed 2026-03-31 13:06:08 +02:00
1 parent 2b8d29ac0d
commit 1fc8aa0e16
10 files changed
+702 -82

No files matched your search

@@ -7,6 +7,7 @@ use crate::event_mapping;
use codex_protocol::items::TurnItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::RolloutItem;
/// Return the indices of user message boundaries in a rollout.
@@ -40,6 +41,51 @@ pub(crate) fn user_message_positions_in_rollout(items: &[RolloutItem]) -> Vec<us
user_positions
}
/// Return the indices of fork-turn boundaries in a rollout.
///
/// A fork-turn boundary is either:
/// - a real user message boundary, or
/// - an assistant inter-agent envelope whose parsed `trigger_turn` is `true`.
///
/// Like `user_message_positions_in_rollout`, this applies `ThreadRolledBack` markers so indexing
/// reflects the effective post-rollback history. Rollback counts real user turns only, so a
/// rollback removes the stale suffix starting at the earliest rolled-back user boundary instead of
/// simply truncating the mixed fork-boundary list.
pub(crate) fn fork_turn_positions_in_rollout(items: &[RolloutItem]) -> Vec<usize> {
let mut user_positions = Vec::new();
let mut fork_turn_positions = Vec::new();
for (idx, item) in items.iter().enumerate() {
match item {
RolloutItem::ResponseItem(item) if is_real_user_message_boundary(item) => {
user_positions.push(idx);
fork_turn_positions.push(idx);
}
RolloutItem::ResponseItem(item) if is_trigger_turn_boundary(item) => {
fork_turn_positions.push(idx);
}
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(rollback)) => {
let num_turns = usize::try_from(rollback.num_turns).unwrap_or(usize::MAX);
if num_turns == 0 {
continue;
}
let Some(rollback_start_idx) = user_positions
.len()
.checked_sub(num_turns)
.map(|rollback_start| user_positions[rollback_start])
.or_else(|| user_positions.first().copied())
else {
continue;
};
let new_user_len = user_positions.len().saturating_sub(num_turns);
user_positions.truncate(new_user_len);
fork_turn_positions.retain(|position| *position < rollback_start_idx);
}
_ => {}
}
}
fork_turn_positions
}
/// Return a prefix of `items` obtained by cutting strictly before the nth user message.
///
/// The boundary index is 0-based from the start of `items` (so `n_from_start = 0` returns
@@ -68,6 +114,43 @@ pub(crate) fn truncate_rollout_before_nth_user_message_from_start(
items[..cut_idx].to_vec()
}
/// Return a suffix of `items` that keeps the last `n_from_end` fork turns.
///
/// If fewer than or equal to `n_from_end` fork turns exist, this returns the full rollout.
pub(crate) fn truncate_rollout_to_last_n_fork_turns(
items: &[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);
if fork_turn_positions.len() <= n_from_end {
return items.to_vec();
}
let keep_idx = fork_turn_positions[fork_turn_positions.len() - n_from_end];
items[keep_idx..].to_vec()
}
fn is_real_user_message_boundary(item: &ResponseItem) -> bool {
matches!(
event_mapping::parse_turn_item(item),
Some(TurnItem::UserMessage(_))
)
}
fn is_trigger_turn_boundary(item: &ResponseItem) -> bool {
let ResponseItem::Message { role, content, .. } = item else {
return false;
};
role == "assistant"
&& InterAgentCommunication::from_message_content(content)
.is_some_and(|communication| communication.trigger_turn)
}
#[cfg(test)]
#[path = "thread_rollout_truncation_tests.rs"]
mod tests;