mirror of
https://github.com/openai/codex.git
synced 2026-10-19 17:28:46 +00:00
Persist resume metadata in compaction checkpoints (#47323)
## Why Compaction companion records do not represent all resume state, including the last started turn and previous turn settings. The selected multi-agent runtime also needs a persisted value when no companion turn context is present. ## What changed - Add optional `resume_metadata` to `CompactedItem`, recording the multi-agent version, last started turn ID, and previous turn settings. Preserve the distinction between absent metadata and explicitly empty values while retaining compatibility with older records. - Capture this metadata under the settings persistence lock alongside the compaction checkpoint. - Rewrite inherited metadata to use the child agent's runtime, and clear previous turn settings when the child must rebuild context. - Clear pending continuation state before delivering acceptance of standalone settings updates, while holding the persistence lock. ## Testing Add coverage for metadata round trips with empty values, legacy reader compatibility, checkpoint companion records, inherited fork metadata, and continuation invalidation while settings event delivery is blocked. GitOrigin-RevId: f30b7feaa7e2fdb5b65fc3c53b37fd7a7fe27903
This commit is contained in:
BIN
Binary file not shown.
@@ -4510,6 +4510,7 @@ mod tests {
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
|
||||
turn_id: "turn-compact".into(),
|
||||
|
||||
@@ -217,6 +217,7 @@ fn ignores_legacy_abort_without_turn_id_and_context_only_records() {
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}));
|
||||
let security_risk = project(RolloutItem::SecurityRiskScore(SecurityRiskScore {
|
||||
scores: BTreeMap::from([("action_risk".to_string(), 0.92)]),
|
||||
|
||||
@@ -206,6 +206,7 @@ async fn thread_resume_paginated_model_context_preserves_original_metadata() ->
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
)
|
||||
.await?;
|
||||
|
||||
@@ -70,7 +70,7 @@ pub(super) fn agent_nickname_candidates(config: &Config, role_name: Option<&str>
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn keep_forked_rollout_item(item: &RolloutItem, preserve_reference_context_item: bool) -> bool {
|
||||
fn keep_forked_rollout_item(item: &RolloutItem, preserve_context_baselines: bool) -> bool {
|
||||
match item {
|
||||
RolloutItem::ResponseItem(envelope) => match &envelope.item {
|
||||
ResponseItem::Message { role, phase, .. } => match role.as_str() {
|
||||
@@ -107,7 +107,7 @@ fn keep_forked_rollout_item(item: &RolloutItem, preserve_reference_context_item:
|
||||
// Full-history forks preserve the cached prompt prefix and can keep diffing
|
||||
// from the parent's durable baseline. Truncated forks drop part of that prompt,
|
||||
// so they must rebuild context on their first child turn.
|
||||
RolloutItem::TurnContext(_) | RolloutItem::WorldState(_) => preserve_reference_context_item,
|
||||
RolloutItem::TurnContext(_) | RolloutItem::WorldState(_) => preserve_context_baselines,
|
||||
// Child threads inherit model context, not the parent's cumulative usage state.
|
||||
RolloutItem::TokenUsageRecord(_) => false,
|
||||
RolloutItem::Compacted(_) | RolloutItem::EventMsg(_) | RolloutItem::SessionMeta(_) => true,
|
||||
@@ -936,9 +936,8 @@ impl LocalAgentControl {
|
||||
} else {
|
||||
Vec::new()
|
||||
};
|
||||
let mut preserve_reference_context_item =
|
||||
matches!(fork_mode, SpawnAgentForkMode::FullHistory);
|
||||
if preserve_reference_context_item {
|
||||
let mut preserve_context_baselines = matches!(fork_mode, SpawnAgentForkMode::FullHistory);
|
||||
if preserve_context_baselines {
|
||||
for item in forked_rollout_items.iter().rev() {
|
||||
let RolloutItem::Compacted(compacted) = item else {
|
||||
continue;
|
||||
@@ -946,7 +945,7 @@ impl LocalAgentControl {
|
||||
// Legacy checkpoints force the child to rebuild context regardless of the
|
||||
// live parent's reference baseline; an older superseded checkpoint does not.
|
||||
if compacted.replacement_history.is_none() {
|
||||
preserve_reference_context_item = false;
|
||||
preserve_context_baselines = false;
|
||||
}
|
||||
break;
|
||||
}
|
||||
@@ -1001,7 +1000,7 @@ impl LocalAgentControl {
|
||||
// If the child will rebuild its initial context, drop the inherited
|
||||
// instructions; startup will add the current requirements and effort
|
||||
// instructions once.
|
||||
return preserve_reference_context_item;
|
||||
return preserve_context_baselines;
|
||||
}
|
||||
let (
|
||||
Some(parent_developer_instructions),
|
||||
@@ -1019,7 +1018,7 @@ impl LocalAgentControl {
|
||||
}
|
||||
|
||||
*replaced = true;
|
||||
let replacement = if preserve_reference_context_item {
|
||||
let replacement = if preserve_context_baselines {
|
||||
subagent_developer_instructions.as_str()
|
||||
} else {
|
||||
""
|
||||
@@ -1034,7 +1033,7 @@ impl LocalAgentControl {
|
||||
true
|
||||
};
|
||||
forked_rollout_items.retain_mut(|item| {
|
||||
if !keep_forked_rollout_item(item, preserve_reference_context_item)
|
||||
if !keep_forked_rollout_item(item, preserve_context_baselines)
|
||||
|| destination_history_mode == Some(ThreadHistoryMode::Paginated)
|
||||
&& matches!(
|
||||
&*item,
|
||||
@@ -1054,8 +1053,15 @@ impl LocalAgentControl {
|
||||
retain_forked_item(response_item, &mut replaced_parent_developer_instructions)
|
||||
}
|
||||
RolloutItem::Compacted(compacted) => {
|
||||
// This checkpoint belongs to the inherited parent prefix.
|
||||
// This compaction becomes part of the subagent's initial history. Rewrite its
|
||||
// metadata to describe the child rather than the parent.
|
||||
compacted.latest_token_usage_record = None;
|
||||
if let Some(resume_metadata) = &mut compacted.resume_metadata {
|
||||
resume_metadata.multi_agent_version = Some(multi_agent_version);
|
||||
if !preserve_context_baselines {
|
||||
resume_metadata.previous_turn_settings = None;
|
||||
}
|
||||
}
|
||||
// Parent-local review evidence must not become the child's authorization.
|
||||
// Root user authorization is collected separately by the host.
|
||||
compacted.guardian_history = None;
|
||||
@@ -1097,7 +1103,7 @@ impl LocalAgentControl {
|
||||
// context omitted the parent's developer fragment, append the child's override so its
|
||||
// instructions still reach the model exactly once.
|
||||
if let Some(subagent_developer_instructions) = subagent_developer_instructions.as_ref()
|
||||
&& preserve_reference_context_item
|
||||
&& preserve_context_baselines
|
||||
&& !replaced_parent_developer_instructions
|
||||
&& !subagent_developer_instructions.is_empty()
|
||||
&& parent_thread
|
||||
@@ -1111,7 +1117,7 @@ impl LocalAgentControl {
|
||||
));
|
||||
forked_rollout_items.push(RolloutItem::ResponseItem(developer_message.into()));
|
||||
}
|
||||
if preserve_reference_context_item
|
||||
if preserve_context_baselines
|
||||
&& multi_agent_version == MultiAgentVersion::V2
|
||||
&& let Some(subagent_usage_hint) = options
|
||||
.multi_agent_v2_usage_hints
|
||||
|
||||
@@ -1903,7 +1903,7 @@ async fn spawn_agent_without_fork_from_paginated_parent_stays_fresh_and_paginate
|
||||
#[test_case::test_case(true; "thread context enabled")]
|
||||
#[test_case::test_case(false; "thread context disabled")]
|
||||
#[tokio::test]
|
||||
async fn spawn_agent_fork_drops_inherited_token_usage_state(thread_context_enabled: bool) {
|
||||
async fn spawn_agent_fork_sanitizes_inherited_compaction_metadata(thread_context_enabled: bool) {
|
||||
let mut harness = AgentControlHarness::new().await;
|
||||
let _ = harness.config.features.disable(Feature::MultiAgentV2);
|
||||
harness
|
||||
@@ -1912,6 +1912,15 @@ async fn spawn_agent_fork_drops_inherited_token_usage_state(thread_context_enabl
|
||||
.set_enabled(Feature::GuardianThreadContext, thread_context_enabled)
|
||||
.expect("test context mode");
|
||||
let (parent_thread_id, parent_thread) = harness.start_paginated_thread().await;
|
||||
let parent_resume_metadata = codex_history::CompactionResumeMetadata {
|
||||
multi_agent_version: Some(MultiAgentVersion::V2),
|
||||
last_started_turn_id: Some("parent-turn".into()),
|
||||
previous_turn_settings: Some(codex_history::PreviousTurnSettings {
|
||||
model: "parent-model".into(),
|
||||
comp_hash: None,
|
||||
realtime_active: None,
|
||||
}),
|
||||
};
|
||||
let parent_usage = TokenUsage {
|
||||
total_tokens: 120,
|
||||
..TokenUsage::default()
|
||||
@@ -1936,12 +1945,13 @@ async fn spawn_agent_fork_drops_inherited_token_usage_state(thread_context_enabl
|
||||
retained_context: None,
|
||||
guardian_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
window_number: Some(1),
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: Some(parent_record.clone()),
|
||||
resume_metadata: Some(parent_resume_metadata.clone()),
|
||||
}),
|
||||
RolloutItem::TokenUsageRecord(parent_record),
|
||||
rollout_response_item(spawn_agent_call(&parent_spawn_call_id)),
|
||||
@@ -2018,6 +2028,20 @@ async fn spawn_agent_fork_drops_inherited_token_usage_state(thread_context_enabl
|
||||
}),
|
||||
"child rollout should not inherit parent token usage checkpoints"
|
||||
);
|
||||
let inherited_resume_metadata = lines
|
||||
.iter()
|
||||
.find_map(|line| match &line.item {
|
||||
RolloutItem::Compacted(item) => item.resume_metadata.as_ref(),
|
||||
_ => None,
|
||||
})
|
||||
.expect("inherited checkpoint");
|
||||
assert_eq!(
|
||||
inherited_resume_metadata,
|
||||
&codex_history::CompactionResumeMetadata {
|
||||
multi_agent_version: Some(MultiAgentVersion::V1),
|
||||
..parent_resume_metadata
|
||||
}
|
||||
);
|
||||
let child_record = lines.iter().rev().find_map(|line| match &line.item {
|
||||
RolloutItem::TokenUsageRecord(record) => Some(record),
|
||||
_ => None,
|
||||
@@ -2068,6 +2092,7 @@ async fn spawn_agent_numeric_fork_from_compacted_paginated_parent_clamps_to_prov
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
rollout_response_item(ResponseItem::Message {
|
||||
id: None,
|
||||
@@ -2627,6 +2652,7 @@ async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history(
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
RolloutItem::RetainedContext(answer_event),
|
||||
RolloutItem::ResponseItem(delivery),
|
||||
@@ -2832,6 +2858,7 @@ async fn spawn_agent_full_fork_restores_instructions_after_compaction_discards_p
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
RolloutItem::TurnContext(turn_context.to_turn_context_item()),
|
||||
rollout_response_item(spawn_agent_call(&parent_spawn_call_id)),
|
||||
@@ -2989,6 +3016,15 @@ async fn spawn_agent_full_fork_legacy_compaction_rebuilds_child_instructions_onc
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: Some(codex_history::CompactionResumeMetadata {
|
||||
multi_agent_version: Some(MultiAgentVersion::V2),
|
||||
last_started_turn_id: None,
|
||||
previous_turn_settings: Some(codex_history::PreviousTurnSettings {
|
||||
model: "parent-model".into(),
|
||||
comp_hash: None,
|
||||
realtime_active: None,
|
||||
}),
|
||||
}),
|
||||
}),
|
||||
];
|
||||
if let Some(instructions) = parent_developer_instructions {
|
||||
@@ -3057,6 +3093,27 @@ async fn spawn_agent_full_fork_legacy_compaction_rebuilds_child_instructions_onc
|
||||
.get_thread(child_thread_id)
|
||||
.await
|
||||
.expect("child thread should be registered");
|
||||
child_thread
|
||||
.flush_rollout()
|
||||
.await
|
||||
.expect("flush child checkpoint");
|
||||
let (items, _, _) = codex_rollout::RolloutRecorder::load_rollout_items(
|
||||
&child_thread.rollout_path().expect("child rollout"),
|
||||
)
|
||||
.await
|
||||
.expect("read child checkpoint");
|
||||
let inherited_state = items.into_iter().find_map(|item| match item {
|
||||
RolloutItem::Compacted(item) => item.resume_metadata,
|
||||
_ => None,
|
||||
});
|
||||
assert_eq!(
|
||||
inherited_state,
|
||||
Some(codex_history::CompactionResumeMetadata {
|
||||
multi_agent_version: Some(MultiAgentVersion::V2),
|
||||
last_started_turn_id: None,
|
||||
previous_turn_settings: None,
|
||||
})
|
||||
);
|
||||
while child_thread
|
||||
.session
|
||||
.reference_context_item()
|
||||
|
||||
@@ -24,6 +24,7 @@ impl Session {
|
||||
window_id: Some(window_ids.window_id.to_string()),
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: state.latest_token_usage_record.clone(),
|
||||
resume_metadata: None,
|
||||
})];
|
||||
if let Some(world_state) = history.world_state_checkpoint() {
|
||||
items.push(RolloutItem::WorldState(world_state));
|
||||
|
||||
@@ -286,20 +286,6 @@ use self::turn_context::TurnContext;
|
||||
#[cfg(test)]
|
||||
mod rollout_reconstruction_tests;
|
||||
|
||||
/// Notes from the previous real user turn.
|
||||
///
|
||||
/// Conceptually this is the same role that `previous_model` used to fill, but
|
||||
/// it can carry other prior-turn settings that matter when constructing
|
||||
/// sensible state-change diffs or full-context reinjection, such as model
|
||||
/// switches, compaction compatibility, or detecting a prior
|
||||
/// `realtime_active -> false` transition.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub(crate) struct PreviousTurnSettings {
|
||||
pub(crate) model: String,
|
||||
pub(crate) comp_hash: Option<String>,
|
||||
pub(crate) realtime_active: Option<bool>,
|
||||
}
|
||||
|
||||
use crate::exec_policy::ExecPolicyUpdateError;
|
||||
use crate::guardian::GuardianReviewSessionManager;
|
||||
use crate::mcp::McpEnvironmentScope;
|
||||
@@ -341,7 +327,9 @@ use codex_core_plugins::RecommendedPluginCandidatesInput;
|
||||
use codex_git_utils::get_git_repo_root;
|
||||
use codex_history::CodexHarnessMetadata;
|
||||
use codex_history::CompactedItem;
|
||||
use codex_history::CompactionResumeMetadata;
|
||||
use codex_history::InitialHistory;
|
||||
pub(crate) use codex_history::PreviousTurnSettings;
|
||||
use codex_history::ResponseItemEnvelope;
|
||||
use codex_mcp::McpConfig;
|
||||
use codex_mcp::effective_mcp_servers;
|
||||
@@ -4021,28 +4009,13 @@ impl Session {
|
||||
.get_or_insert_default()
|
||||
.compaction_model_hash = metadata.compaction_model_hash;
|
||||
}
|
||||
let mut compacted_item = CompactedItem {
|
||||
message: metadata.message,
|
||||
replacement_history: Some(items.clone()),
|
||||
retained_context: None,
|
||||
guardian_history: None,
|
||||
mcp_resource_origins: self.services.mcp_runtime.resource_origin_checkpoint(),
|
||||
window_number: Some(metadata.window_number),
|
||||
first_window_id: Some(metadata.window_ids.first_window_id.to_string()),
|
||||
previous_window_id: metadata
|
||||
.window_ids
|
||||
.previous_window_id
|
||||
.map(|id| id.to_string()),
|
||||
window_id: Some(metadata.window_ids.window_id.to_string()),
|
||||
compaction_response_id: metadata.compaction_response_id,
|
||||
latest_token_usage_record: self.state.lock().await.latest_token_usage_record.clone(),
|
||||
};
|
||||
let replacement_history = items.clone();
|
||||
// Wait for accepted updates to finish persisting, then keep later updates from
|
||||
// overtaking the current settings snapshot while its checkpoint is written.
|
||||
let _settings_guard = thread_settings::acquire_persistence_lock(self).await;
|
||||
// Compaction starts a new history window, so its WorldState baseline must be full.
|
||||
let mut world_state_item = None;
|
||||
{
|
||||
let compacted_item = {
|
||||
let mut state = self.state.lock().await;
|
||||
state.replace_annotated_history(
|
||||
items,
|
||||
@@ -4051,15 +4024,34 @@ impl Session {
|
||||
reviewer_compaction_hash: metadata.reviewer_compaction_hash,
|
||||
},
|
||||
);
|
||||
compacted_item.guardian_history = state.history.guardian_history_checkpoint();
|
||||
compacted_item.retained_context = Some(state.history.retained_context().clone());
|
||||
state.reasoning_effort_pin = ReasoningEffortPin::Compacted;
|
||||
if let Some(world_state) = world_state_baseline {
|
||||
let snapshot = world_state.snapshot();
|
||||
world_state_item = Some(WorldStateItem::full(snapshot.clone().into_object()));
|
||||
state.history.set_world_state_baseline(snapshot);
|
||||
}
|
||||
}
|
||||
CompactedItem {
|
||||
message: metadata.message,
|
||||
replacement_history: Some(replacement_history),
|
||||
guardian_history: state.history.guardian_history_checkpoint(),
|
||||
retained_context: Some(state.history.retained_context().clone()),
|
||||
mcp_resource_origins: self.services.mcp_runtime.resource_origin_checkpoint(),
|
||||
window_number: Some(metadata.window_number),
|
||||
first_window_id: Some(metadata.window_ids.first_window_id.to_string()),
|
||||
previous_window_id: metadata
|
||||
.window_ids
|
||||
.previous_window_id
|
||||
.map(|id| id.to_string()),
|
||||
window_id: Some(metadata.window_ids.window_id.to_string()),
|
||||
compaction_response_id: metadata.compaction_response_id,
|
||||
latest_token_usage_record: state.latest_token_usage_record.clone(),
|
||||
resume_metadata: Some(CompactionResumeMetadata {
|
||||
multi_agent_version: self.multi_agent_version(),
|
||||
last_started_turn_id: state.last_started_turn_id.clone(),
|
||||
previous_turn_settings: state.previous_turn_settings(),
|
||||
}),
|
||||
}
|
||||
};
|
||||
|
||||
let mut rollout_items = vec![RolloutItem::Compacted(compacted_item)];
|
||||
// Persist the baseline after the replacement history that established it.
|
||||
|
||||
@@ -1243,6 +1243,7 @@ async fn record_initial_history_resumed_rollback_drops_incomplete_user_turn_comp
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(
|
||||
codex_protocol::protocol::ThreadRolledBackEvent { num_turns: 1 },
|
||||
@@ -1307,6 +1308,7 @@ async fn record_initial_history_requires_surviving_full_snapshot_without_user_tu
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
],
|
||||
};
|
||||
@@ -1343,6 +1345,7 @@ async fn record_initial_history_resumed_does_not_seed_reference_context_item_aft
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
];
|
||||
|
||||
@@ -1417,6 +1420,7 @@ async fn reconstruct_history_prefers_compacted_window_over_session_meta() {
|
||||
window_id: Some(compacted_window_id.to_string()),
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
];
|
||||
|
||||
@@ -1457,6 +1461,7 @@ async fn reconstruct_history_replays_world_state_from_latest_compaction_window()
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
RolloutItem::WorldState(WorldStateItem::full(object!({
|
||||
"environment": {"status": "starting", "cwd": "/workspace"}
|
||||
@@ -1527,6 +1532,7 @@ async fn bounded_replay_matches_full_replay_after_empty_turn_compactions() {
|
||||
window_id: Some(window_ids[window_number].to_string()),
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
RolloutItem::WorldState(WorldStateItem::full(object!({
|
||||
"environment": {"window": window_number, "status": "starting"}
|
||||
@@ -1636,6 +1642,7 @@ async fn reconstruct_history_preserves_legacy_compaction_count_with_session_meta
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
];
|
||||
|
||||
@@ -1691,6 +1698,7 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_does_
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
];
|
||||
|
||||
@@ -1732,6 +1740,7 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_clear
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
codex_protocol::protocol::TurnStartedEvent {
|
||||
@@ -1844,6 +1853,7 @@ async fn record_initial_history_resumed_turn_context_after_compaction_reestablis
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
RolloutItem::TurnContext(previous_context_item),
|
||||
RolloutItem::EventMsg(EventMsg::TurnComplete(
|
||||
@@ -2021,6 +2031,7 @@ async fn record_initial_history_resumed_aborted_turn_without_id_clears_active_tu
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
];
|
||||
|
||||
@@ -2286,6 +2297,7 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_compaction_clea
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
];
|
||||
|
||||
@@ -2469,6 +2481,7 @@ async fn record_initial_history_resumed_replaced_incomplete_compacted_turn_clear
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
// A newer TurnStarted replaces the incomplete compacted turn without a matching
|
||||
// completion/abort for the old one.
|
||||
|
||||
@@ -2300,6 +2300,7 @@ async fn reconstruct_history_uses_replacement_history_verbatim() {
|
||||
window_id: Some(window_id.to_string()),
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
})];
|
||||
|
||||
let reconstructed = session
|
||||
@@ -3098,6 +3099,7 @@ fn latest_token_usage_record_stops_at_compaction_checkpoint() {
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record,
|
||||
resume_metadata: None,
|
||||
})
|
||||
};
|
||||
|
||||
@@ -5718,6 +5720,117 @@ async fn mcp_attribution_checkpoints_cover_batch_prefixes_compaction_and_restore
|
||||
assert_eq!(restored.mcp_attribution_snapshot(), expected);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn standalone_settings_invalidate_continuation_before_delivering_acceptance() {
|
||||
let (mut session, _) = make_session_and_context().await;
|
||||
let (tx, rx) = async_channel::bounded(1);
|
||||
session.tx_event = tx;
|
||||
session.state.lock().await.last_started_turn_id = Some("superseded-turn".into());
|
||||
session
|
||||
.tx_event
|
||||
.send(Event {
|
||||
id: "occupied".into(),
|
||||
msg: EventMsg::ThreadSettingsApplied(
|
||||
codex_protocol::protocol::ThreadSettingsAppliedEvent {
|
||||
thread_id: Some(session.thread_id()),
|
||||
thread_settings: session.thread_settings_snapshot().await,
|
||||
},
|
||||
),
|
||||
})
|
||||
.await
|
||||
.expect("fill event channel");
|
||||
let session = Arc::new(session);
|
||||
let mut update = Box::pin(tokio::task::unconstrained(thread_settings::update(
|
||||
&session,
|
||||
"settings".into(),
|
||||
codex_protocol::protocol::ThreadSettingsOverrides::default(),
|
||||
)));
|
||||
assert!(futures::poll!(update.as_mut()).is_pending());
|
||||
assert_eq!(session.state.lock().await.last_started_turn_id, None);
|
||||
let mut checkpoint = Box::pin(session.checkpoint_thread_settings());
|
||||
assert!(futures::poll!(checkpoint.as_mut()).is_pending());
|
||||
rx.recv().await.expect("release event delivery");
|
||||
update.await;
|
||||
checkpoint.await.expect("checkpoint after settings update");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn compaction_persists_resume_metadata_and_companion_records() {
|
||||
let (mut session, turn_context) = make_session_and_context().await;
|
||||
let rollout_path = attach_thread_persistence(&mut session).await;
|
||||
let turn_context = Arc::new(turn_context);
|
||||
let turn_context_baseline = turn_context.to_turn_context_item();
|
||||
let world_state = Arc::new(build_world_state_from_turn_context(&session, &turn_context).await);
|
||||
let previous_turn_settings = PreviousTurnSettings {
|
||||
model: "previous-model".to_string(),
|
||||
comp_hash: Some("comp-hash".to_string()),
|
||||
realtime_active: Some(true),
|
||||
};
|
||||
session
|
||||
.set_previous_turn_settings(Some(previous_turn_settings.clone()))
|
||||
.await;
|
||||
|
||||
session.state.lock().await.last_started_turn_id = Some("checkpoint-turn".into());
|
||||
session.multi_agent_version = std::sync::OnceLock::from(MultiAgentVersion::V2);
|
||||
let expected = CompactionResumeMetadata {
|
||||
multi_agent_version: Some(MultiAgentVersion::V2),
|
||||
last_started_turn_id: Some("checkpoint-turn".into()),
|
||||
previous_turn_settings: Some(previous_turn_settings),
|
||||
};
|
||||
let expected_settings = codex_protocol::protocol::ThreadSettingsAppliedEvent {
|
||||
thread_id: Some(session.thread_id()),
|
||||
thread_settings: session.thread_settings_snapshot().await,
|
||||
};
|
||||
|
||||
for with_baselines in [true, false] {
|
||||
let (window_number, window_ids) = session.advance_auto_compact_window().await;
|
||||
session
|
||||
.replace_compacted_history(
|
||||
vec![ResponseItemEnvelope::new(user_message("compacted context"))],
|
||||
with_baselines.then_some(turn_context_baseline.clone()),
|
||||
with_baselines.then_some(Arc::clone(&world_state)),
|
||||
CompactedHistoryMetadata {
|
||||
message: String::new(),
|
||||
window_number,
|
||||
window_ids,
|
||||
compaction_response_id: None,
|
||||
compaction_model_hash: None,
|
||||
reviewer_compaction_hash: None,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
session.flush_rollout().await.expect("flush checkpoints");
|
||||
let (items, _, _) = RolloutRecorder::load_rollout_items(&rollout_path)
|
||||
.await
|
||||
.expect("read checkpoints");
|
||||
let compaction_items = items
|
||||
.into_iter()
|
||||
.skip_while(|item| !matches!(item, RolloutItem::Compacted(_)))
|
||||
.collect::<Vec<_>>();
|
||||
let [
|
||||
RolloutItem::Compacted(first),
|
||||
RolloutItem::WorldState(first_world_state),
|
||||
RolloutItem::TurnContext(first_turn_context),
|
||||
RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied(first_settings)),
|
||||
RolloutItem::Compacted(second),
|
||||
RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied(second_settings)),
|
||||
] = compaction_items.as_slice()
|
||||
else {
|
||||
panic!("unexpected compaction records: {compaction_items:#?}");
|
||||
};
|
||||
assert_eq!(first.resume_metadata.as_ref(), Some(&expected));
|
||||
assert_eq!(second.resume_metadata.as_ref(), Some(&expected));
|
||||
assert_eq!(
|
||||
first_world_state,
|
||||
&WorldStateItem::full(world_state.snapshot().into_object())
|
||||
);
|
||||
assert_eq!(first_turn_context, &turn_context_baseline);
|
||||
assert_eq!(first_settings, &expected_settings);
|
||||
assert_eq!(second_settings, &expected_settings);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn session_settings_commit_keeps_snapshot_across_postcommit_wait() {
|
||||
let (session, _turn_context, _rx) = make_session_and_context_with_auth_and_config_and_rx(
|
||||
@@ -12424,6 +12537,7 @@ async fn sample_rollout(
|
||||
window_id: Some(window_ids.window_id.to_string()),
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}));
|
||||
|
||||
let user2 = user_message("second user");
|
||||
@@ -12458,6 +12572,7 @@ async fn sample_rollout(
|
||||
window_id: Some(window_ids.window_id.to_string()),
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}));
|
||||
|
||||
let user3 = user_message("third user");
|
||||
|
||||
@@ -39,20 +39,25 @@ pub(super) async fn update(
|
||||
overrides: ThreadSettingsOverrides,
|
||||
) {
|
||||
let updates = prepare_update(overrides);
|
||||
if let Err(error) = apply_update(session, submission_id.clone(), updates).await {
|
||||
session
|
||||
.send_event_raw(Event {
|
||||
id: submission_id,
|
||||
msg: EventMsg::Error(ErrorEvent {
|
||||
misalignment: None,
|
||||
message: format!("invalid thread settings override: {error}"),
|
||||
codex_error_info: Some(CodexErrorInfo::BadRequest),
|
||||
}),
|
||||
})
|
||||
.await;
|
||||
} else {
|
||||
// Standalone settings changes supersede a pending automatic continuation.
|
||||
session.state.lock().await.last_started_turn_id = None;
|
||||
let _settings_guard = acquire_persistence_lock(session).await;
|
||||
match session.update_settings(updates).await {
|
||||
Ok(commit) => {
|
||||
// Standalone settings changes supersede a pending automatic continuation.
|
||||
session.state.lock().await.last_started_turn_id = None;
|
||||
emit_applied(session, submission_id, commit.snapshot).await;
|
||||
}
|
||||
Err(error) => {
|
||||
session
|
||||
.send_event_raw(Event {
|
||||
id: submission_id,
|
||||
msg: EventMsg::Error(ErrorEvent {
|
||||
misalignment: None,
|
||||
message: format!("invalid thread settings override: {error}"),
|
||||
codex_error_info: Some(CodexErrorInfo::BadRequest),
|
||||
}),
|
||||
})
|
||||
.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -73,6 +73,7 @@ fn restores_cumulative_item_and_compaction_checkpoints() {
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
]);
|
||||
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
//! Defines resume metadata stored directly on a compaction.
|
||||
|
||||
use codex_protocol::protocol::MultiAgentVersion;
|
||||
use schemars::JsonSchema;
|
||||
use serde::Deserialize;
|
||||
use serde::Serialize;
|
||||
|
||||
/// Resume metadata that is not represented by the companion records after a compaction.
|
||||
///
|
||||
/// Presence distinguishes compactions that explicitly persisted these values from older
|
||||
/// compactions that did not. Each field is authoritative, including an intentionally absent value.
|
||||
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
|
||||
pub struct CompactionResumeMetadata {
|
||||
/// Selected runtime, including when the companion turn context is absent.
|
||||
pub multi_agent_version: Option<MultiAgentVersion>,
|
||||
/// Turn identity used to admit continuations after cold resume.
|
||||
pub last_started_turn_id: Option<String>,
|
||||
pub previous_turn_settings: Option<PreviousTurnSettings>,
|
||||
}
|
||||
|
||||
/// Previous user-turn settings used to reconstruct context changes after resume.
|
||||
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
|
||||
pub struct PreviousTurnSettings {
|
||||
pub model: String,
|
||||
pub comp_hash: Option<String>,
|
||||
pub realtime_active: Option<bool>,
|
||||
}
|
||||
@@ -1,5 +1,9 @@
|
||||
//! Model-history and persisted-rollout domain types.
|
||||
|
||||
mod compaction_resume_metadata;
|
||||
pub use compaction_resume_metadata::CompactionResumeMetadata;
|
||||
pub use compaction_resume_metadata::PreviousTurnSettings;
|
||||
|
||||
mod compaction_checkpoint;
|
||||
pub use compaction_checkpoint::CompactionCheckpoint;
|
||||
|
||||
@@ -244,6 +248,9 @@ pub struct CompactedItem {
|
||||
/// `thread/resume` can restore token usage totals from this field without scanning arbitrarily
|
||||
/// far past the compaction.
|
||||
pub latest_token_usage_record: Option<TokenUsageRecord>,
|
||||
/// Resume metadata for values not represented by the companion rollout records.
|
||||
/// Presence distinguishes explicitly persisted values from legacy fallback reconstruction.
|
||||
pub resume_metadata: Option<CompactionResumeMetadata>,
|
||||
}
|
||||
|
||||
impl Serialize for CompactedItem {
|
||||
|
||||
@@ -177,6 +177,8 @@ pub(super) struct CompactedItemWire<'a> {
|
||||
compaction_response_id: Option<Cow<'a, str>>,
|
||||
#[serde(default)]
|
||||
latest_token_usage_record: Option<Cow<'a, TokenUsageRecord>>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
resume_metadata: Option<Cow<'a, crate::CompactionResumeMetadata>>,
|
||||
}
|
||||
|
||||
impl<'a> From<&'a CompactedItem> for CompactedItemWire<'a> {
|
||||
@@ -216,6 +218,7 @@ impl<'a> From<&'a CompactedItem> for CompactedItemWire<'a> {
|
||||
.map(|window_id| WindowIdWire::Id(Cow::Borrowed(window_id))),
|
||||
compaction_response_id: item.compaction_response_id.as_deref().map(Cow::Borrowed),
|
||||
latest_token_usage_record: item.latest_token_usage_record.as_ref().map(Cow::Borrowed),
|
||||
resume_metadata: item.resume_metadata.as_ref().map(Cow::Borrowed),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -281,6 +284,7 @@ impl TryFrom<CompactedItemWire<'_>> for CompactedItem {
|
||||
window_id,
|
||||
compaction_response_id: item.compaction_response_id.map(Cow::into_owned),
|
||||
latest_token_usage_record: item.latest_token_usage_record.map(Cow::into_owned),
|
||||
resume_metadata: item.resume_metadata.map(Cow::into_owned),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -414,6 +414,7 @@ fn compacted_replacement_history_stores_metadata_in_an_aligned_sidecar() -> Resu
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
};
|
||||
|
||||
let serialized = serde_json::to_value(item)?;
|
||||
@@ -451,6 +452,34 @@ fn compacted_replacement_history_stores_metadata_in_an_aligned_sidecar() -> Resu
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn compacted_resume_metadata_presence_round_trips_empty_values() -> Result<()> {
|
||||
let resume_metadata = CompactionResumeMetadata {
|
||||
multi_agent_version: None,
|
||||
last_started_turn_id: None,
|
||||
previous_turn_settings: None,
|
||||
};
|
||||
let item = CompactedItem {
|
||||
message: "summary".to_string(),
|
||||
replacement_history: None,
|
||||
retained_context: None,
|
||||
guardian_history: None,
|
||||
mcp_resource_origins: None,
|
||||
window_number: None,
|
||||
first_window_id: None,
|
||||
previous_window_id: None,
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: Some(resume_metadata.clone()),
|
||||
};
|
||||
|
||||
let serialized = serde_json::to_value(&item)?;
|
||||
assert_eq!(serialized["resume_metadata"], json!(resume_metadata));
|
||||
assert_eq!(serde_json::from_value::<CompactedItem>(serialized)?, item);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
/// Rejects checkpoint sidecars that cannot be paired unambiguously with history.
|
||||
fn compacted_replacement_history_rejects_misaligned_metadata() {
|
||||
@@ -525,6 +554,11 @@ fn compacted_metadata_remains_compatible_with_legacy_response_item_readers() ->
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: Some(CompactionResumeMetadata {
|
||||
multi_agent_version: Some(MultiAgentVersion::V2),
|
||||
last_started_turn_id: Some("turn-1".to_string()),
|
||||
previous_turn_settings: None,
|
||||
}),
|
||||
}))?;
|
||||
|
||||
let restored: RolloutItem = serde_json::from_value(compacted_line.clone())?;
|
||||
@@ -733,6 +767,7 @@ fn compacted_item_serializes_window_number_and_id() -> Result<()> {
|
||||
window_id: Some("019b3f6e-7a10-7cc3-8b6e-1d09e2f7a001".to_string()),
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
};
|
||||
|
||||
assert_eq!(
|
||||
@@ -772,6 +807,7 @@ fn compacted_item_migrates_legacy_numeric_window_id() -> Result<()> {
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}
|
||||
);
|
||||
Ok(())
|
||||
|
||||
@@ -2173,7 +2173,7 @@ pub struct TurnStartedEvent {
|
||||
pub collaboration_mode_kind: ModeKind,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
|
||||
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, JsonSchema, TS)]
|
||||
pub struct ThreadSettingsAppliedEvent {
|
||||
/// Logical task that owns this snapshot, independent of the physical rollout file.
|
||||
/// Absent in older histories; copied snapshots retain their original owner's ID.
|
||||
|
||||
@@ -27,6 +27,7 @@ pub mod state_db;
|
||||
mod writer_lock;
|
||||
|
||||
pub use codex_history::CompactedItem;
|
||||
pub use codex_history::CompactionResumeMetadata;
|
||||
pub use codex_history::InitialHistory;
|
||||
pub use codex_history::ResponseItemEnvelope;
|
||||
pub use codex_history::ResumedHistory;
|
||||
|
||||
@@ -328,6 +328,7 @@ fn builder_from_items_falls_back_to_filename() {
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
})];
|
||||
|
||||
let builder = builder_from_items(items.as_slice(), path.as_path()).expect("builder");
|
||||
|
||||
@@ -891,5 +891,6 @@ fn compacted(message: &str, replacement_history: Option<Vec<ResponseItem>>) -> R
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -248,6 +248,7 @@ fn compacted(replacement_history: Vec<ResponseItem>) -> RolloutItem {
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1783,6 +1784,7 @@ async fn migration_compacts_subagent_prefix_and_does_not_project_it() {
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
RolloutItem::Compacted(CompactedItem {
|
||||
message: "latest checkpoint".to_string(),
|
||||
@@ -1807,6 +1809,7 @@ async fn migration_compacts_subagent_prefix_and_does_not_project_it() {
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
started("child-turn"),
|
||||
RolloutItem::TurnContext(TurnContextItem {
|
||||
|
||||
@@ -674,6 +674,7 @@ mod tests {
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
});
|
||||
|
||||
let first = sync
|
||||
|
||||
@@ -85,6 +85,7 @@ fn legacy_transcript_preview_scans_tail_across_compaction() {
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
rollout_user_message("recent user"),
|
||||
rollout_agent_message("commentary one\ncommentary two"),
|
||||
@@ -220,6 +221,7 @@ fn legacy_transcript_preview_falls_back_for_oversized_hidden_record() {
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
}),
|
||||
rollout_agent_message("recent assistant"),
|
||||
],
|
||||
@@ -251,6 +253,7 @@ fn legacy_transcript_preview_falls_back_when_scan_budget_is_exhausted() {
|
||||
window_id: None,
|
||||
compaction_response_id: None,
|
||||
latest_token_usage_record: None,
|
||||
resume_metadata: None,
|
||||
});
|
||||
let mut items = vec![rollout_user_message("older user")];
|
||||
items.extend(std::iter::repeat_n(compacted, 9));
|
||||
|
||||
Reference in New Issue
Block a user