github/codex
4
0 Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-13 14:46:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Preserve harness metadata across conversation history (#38058)

Browse Source 
## What changed

- Wrap response items with optional harness-owned metadata while keeping the persisted response payload shape backward compatible.
- Store compacted-history metadata in an aligned sidecar and reject malformed sidecars.
- Carry metadata through history normalization, compaction, resume, fork, truncation, migration, and rollout recording without sending it to model providers.

## Testing

- Cover legacy rollout compatibility, metadata serialization and validation, compaction retention, and resume/fork reconstruction.

GitOrigin-RevId: 17a6127c9048ff984cb4df4d97c0751a38c29f5f
This commit is contained in:
rka-oai authored and copyberry committed 2026-08-11 19:30:12 +00:00
1 parent 44d992c14e
commit 3a6f747d77
67 files changed
+2075 -655
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

codex-rs/core/src/thread_manager_tests.rs
+37 -37
View File
@@ -409,7 +409,7 @@ fn truncates_before_requested_user_message() {
let initial: Vec<RolloutItem> = items
.iter()
.cloned()
.map(RolloutItem::ResponseItem)
.map(|item| RolloutItem::ResponseItem(item.into()))
.collect();
let truncated = truncate_before_nth_user_message(
InitialHistory::Forked(initial),
@@ -423,9 +423,9 @@ fn truncates_before_requested_user_message() {
);
let got_items = truncated.get_rollout_items();
let expected_items = vec![
RolloutItem::ResponseItem(items[0].clone()),
RolloutItem::ResponseItem(items[1].clone()),
RolloutItem::ResponseItem(items[2].clone()),
RolloutItem::ResponseItem(items[0].clone().into()),
RolloutItem::ResponseItem(items[1].clone().into()),
RolloutItem::ResponseItem(items[2].clone().into()),
];
assert_eq!(
serde_json::to_value(got_items).unwrap(),
@@ -435,7 +435,7 @@ fn truncates_before_requested_user_message() {
let initial2: Vec<RolloutItem> = items
.iter()
.cloned()
.map(RolloutItem::ResponseItem)
.map(|item| RolloutItem::ResponseItem(item.into()))
.collect();
let truncated2 = truncate_before_nth_user_message(
InitialHistory::Forked(initial2.clone()),
@@ -456,10 +456,10 @@ fn truncates_before_requested_user_message() {
#[test]
fn out_of_range_truncation_drops_only_unfinished_suffix_mid_turn() {
let items = vec![
RolloutItem::ResponseItem(user_msg("u1")),
RolloutItem::ResponseItem(assistant_msg("a1")),
RolloutItem::ResponseItem(user_msg("u2")),
RolloutItem::ResponseItem(assistant_msg("partial")),
RolloutItem::ResponseItem(user_msg("u1").into()),
RolloutItem::ResponseItem(assistant_msg("a1").into()),
RolloutItem::ResponseItem(user_msg("u2").into()),
RolloutItem::ResponseItem(assistant_msg("partial").into()),
];
let truncated = truncate_before_nth_user_message(
@@ -501,8 +501,8 @@ fn fork_thread_accepts_legacy_usize_snapshot_argument() {
#[test]
fn out_of_range_truncation_drops_pre_user_active_turn_prefix() {
let items = vec![
RolloutItem::ResponseItem(user_msg("u1")),
RolloutItem::ResponseItem(assistant_msg("a1")),
RolloutItem::ResponseItem(user_msg("u1").into()),
RolloutItem::ResponseItem(assistant_msg("a1").into()),
RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
turn_id: "turn-2".to_string(),
trace_id: None,
@@ -510,8 +510,8 @@ fn out_of_range_truncation_drops_pre_user_active_turn_prefix() {
model_context_window: None,
collaboration_mode_kind: Default::default(),
})),
RolloutItem::ResponseItem(user_msg("u2")),
RolloutItem::ResponseItem(assistant_msg("partial")),
RolloutItem::ResponseItem(user_msg("u2").into()),
RolloutItem::ResponseItem(assistant_msg("partial").into()),
];
let snapshot_state = snapshot_turn_state(&InitialHistory::Forked(items.clone()));
@@ -553,7 +553,7 @@ async fn ignores_session_prefix_messages_when_truncating() {
let rollout_items: Vec<RolloutItem> = items
.iter()
.cloned()
.map(RolloutItem::ResponseItem)
.map(|item| RolloutItem::ResponseItem(item.into()))
.collect();
let truncated = truncate_before_nth_user_message(
@@ -569,10 +569,10 @@ async fn ignores_session_prefix_messages_when_truncating() {
let got_items = truncated.get_rollout_items();
let expected: Vec<RolloutItem> = vec![
RolloutItem::ResponseItem(items[0].clone()),
RolloutItem::ResponseItem(items[1].clone()),
RolloutItem::ResponseItem(items[2].clone()),
RolloutItem::ResponseItem(items[3].clone()),
RolloutItem::ResponseItem(items[0].clone().into()),
RolloutItem::ResponseItem(items[1].clone().into()),
RolloutItem::ResponseItem(items[2].clone().into()),
RolloutItem::ResponseItem(items[3].clone().into()),
];
assert_eq!(
@@ -1596,7 +1596,7 @@ async fn rollout_path_resume_and_fork_read_history_through_thread_store() {
config.clone(),
InitialHistory::Resumed(ResumedHistory {
conversation_id: source.thread_id,
history: Arc::new(vec![RolloutItem::ResponseItem(user_msg("hello"))]),
history: Arc::new(vec![RolloutItem::ResponseItem(user_msg("hello").into())]),
rollout_path: Some(rollout_path.clone()),
}),
auth_manager.clone(),
@@ -1751,7 +1751,7 @@ async fn injected_models_manager_controls_refresh_policy() {
#[test]
fn interrupted_fork_snapshot_appends_interrupt_boundary() {
let committed_history =
InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("hello"))]);
InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("hello").into())]);
assert_eq!(
serde_json::to_value(
@@ -1765,8 +1765,8 @@ fn interrupted_fork_snapshot_appends_interrupt_boundary() {
)
.expect("serialize interrupted fork history"),
serde_json::to_value(vec![
RolloutItem::ResponseItem(user_msg("hello")),
RolloutItem::ResponseItem(contextual_user_interrupted_marker()),
RolloutItem::ResponseItem(user_msg("hello").into()),
RolloutItem::ResponseItem(contextual_user_interrupted_marker().into()),
RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
turn_id: None,
started_at: None,
@@ -1789,7 +1789,7 @@ fn interrupted_fork_snapshot_appends_interrupt_boundary() {
)
.expect("serialize interrupted empty fork history"),
serde_json::to_value(vec![
RolloutItem::ResponseItem(contextual_user_interrupted_marker()),
RolloutItem::ResponseItem(contextual_user_interrupted_marker().into()),
RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
turn_id: None,
started_at: None,
@@ -1805,7 +1805,7 @@ fn interrupted_fork_snapshot_appends_interrupt_boundary() {
#[test]
fn disabled_interrupted_fork_snapshot_appends_only_interrupt_event() {
let committed_history =
InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("hello"))]);
InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("hello").into())]);
assert_eq!(
serde_json::to_value(
@@ -1819,7 +1819,7 @@ fn disabled_interrupted_fork_snapshot_appends_only_interrupt_event() {
)
.expect("serialize disabled interrupted fork history"),
serde_json::to_value(vec![
RolloutItem::ResponseItem(user_msg("hello")),
RolloutItem::ResponseItem(user_msg("hello").into()),
RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
turn_id: None,
started_at: None,
@@ -1857,9 +1857,9 @@ fn disabled_interrupted_fork_snapshot_appends_only_interrupt_event() {
#[test]
fn interrupted_snapshot_is_not_mid_turn() {
let interrupted_history = InitialHistory::Forked(vec![
RolloutItem::ResponseItem(user_msg("hello")),
RolloutItem::ResponseItem(assistant_msg("partial")),
RolloutItem::ResponseItem(contextual_user_interrupted_marker()),
RolloutItem::ResponseItem(user_msg("hello").into()),
RolloutItem::ResponseItem(assistant_msg("partial").into()),
RolloutItem::ResponseItem(contextual_user_interrupted_marker().into()),
RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent {
turn_id: Some("turn-1".to_string()),
started_at: None,
@@ -1930,7 +1930,7 @@ fn completed_legacy_event_history_is_not_mid_turn() {
#[test]
fn mixed_response_and_legacy_user_event_history_is_mid_turn() {
let mixed_history = InitialHistory::Forked(vec![
RolloutItem::ResponseItem(user_msg("hello")),
RolloutItem::ResponseItem(user_msg("hello").into()),
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
client_id: None,
message: "hello".to_string(),
@@ -1984,8 +1984,8 @@ async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_histor
.resume_thread_with_history(
config.clone(),
InitialHistory::Forked(vec![
RolloutItem::ResponseItem(user_msg("hello")),
RolloutItem::ResponseItem(assistant_msg("partial")),
RolloutItem::ResponseItem(user_msg("hello").into()),
RolloutItem::ResponseItem(assistant_msg("partial").into()),
]),
auth_manager,
/*parent_trace*/ None,
@@ -2029,7 +2029,7 @@ async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_histor
.filter(|item| !matches!(item, RolloutItem::SessionMeta(_)))
.collect();
let interrupted_marker_json = serde_json::to_value(RolloutItem::ResponseItem(
contextual_user_interrupted_marker(),
contextual_user_interrupted_marker().into(),
))
.expect("serialize interrupted marker");
let interrupted_abort_json = serde_json::to_value(RolloutItem::EventMsg(
@@ -2104,8 +2104,8 @@ async fn interrupted_fork_snapshot_preserves_explicit_turn_id() {
model_context_window: None,
collaboration_mode_kind: Default::default(),
})),
RolloutItem::ResponseItem(user_msg("hello")),
RolloutItem::ResponseItem(assistant_msg("partial")),
RolloutItem::ResponseItem(user_msg("hello").into()),
RolloutItem::ResponseItem(assistant_msg("partial").into()),
]),
auth_manager,
/*parent_trace*/ None,
@@ -2200,8 +2200,8 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_
.resume_thread_with_history(
config.clone(),
InitialHistory::Forked(vec![
RolloutItem::ResponseItem(user_msg("hello")),
RolloutItem::ResponseItem(assistant_msg("partial")),
RolloutItem::ResponseItem(user_msg("hello").into()),
RolloutItem::ResponseItem(assistant_msg("partial").into()),
]),
auth_manager,
/*parent_trace*/ None,
@@ -2244,7 +2244,7 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_
.filter(|item| !matches!(item, RolloutItem::SessionMeta(_)))
.collect();
let interrupted_marker_json = serde_json::to_value(RolloutItem::ResponseItem(
contextual_user_interrupted_marker(),
contextual_user_interrupted_marker().into(),
))
.expect("serialize interrupted marker");
assert_eq!(