From 8017c55d82b990321f2b103da54bdc18bcd41fa4 Mon Sep 17 00:00:00 2001 From: Hare Date: Tue, 20 Oct 2026 05:34:22 +0900 Subject: [PATCH] fix: retain annotated compaction history and honor non-System input policy --- crates/session-store/src/input_metadata.rs | 17 +- crates/session-store/src/public_material.rs | 3 +- crates/session-store/src/public_snapshot.rs | 42 ++- crates/worker/src/session_capture.rs | 319 +++++++++++--------- crates/worker/src/worker.rs | 195 ++++++------ crates/worker/tests/compact_events_test.rs | 129 ++++++++ 6 files changed, 446 insertions(+), 259 deletions(-) diff --git a/crates/session-store/src/input_metadata.rs b/crates/session-store/src/input_metadata.rs index 2d29d07da..eeed42b9c 100644 --- a/crates/session-store/src/input_metadata.rs +++ b/crates/session-store/src/input_metadata.rs @@ -238,11 +238,20 @@ impl HistoryInputMetadata { } pub fn public_use_policy(&self) -> protocol::SessionEntryUsePolicy { - let has_public = self.public.is_some(); protocol::SessionEntryUsePolicy { - observation_readable: has_public - && self.usage.observation == InputObservationUse::Readable, - memory_evidence: has_public && self.usage.memory_evidence == InputEvidenceUse::Include, + observation_readable: self.usage.observation == InputObservationUse::Readable, + memory_evidence: self.usage.memory_evidence == InputEvidenceUse::Include, + } + } + + pub fn public_system_use_policy(&self) -> protocol::SessionEntryUsePolicy { + if self.public.is_some() { + self.public_use_policy() + } else { + protocol::SessionEntryUsePolicy { + observation_readable: false, + memory_evidence: false, + } } } diff --git a/crates/session-store/src/public_material.rs b/crates/session-store/src/public_material.rs index a7cdccaf2..150852174 100644 --- a/crates/session-store/src/public_material.rs +++ b/crates/session-store/src/public_material.rs @@ -903,12 +903,12 @@ fn canonical_body( if allowed == crate::InputEvidenceUse::Exclude { return None; } - let public = input.public_item()?; if let LoggedItem::Message { role: LoggedRole::System, .. } = &entry.item { + let public = input.public_item()?; return Some(CanonicalBody::Message { origin: &entry.metadata.origin, role: LoggedRole::System, @@ -917,7 +917,6 @@ fn canonical_body( }]), }); } - return None; } match &entry.item { LoggedItem::Message { diff --git a/crates/session-store/src/public_snapshot.rs b/crates/session-store/src/public_snapshot.rs index ce23ee221..cf03cf752 100644 --- a/crates/session-store/src/public_snapshot.rs +++ b/crates/session-store/src/public_snapshot.rs @@ -1627,7 +1627,7 @@ pub(crate) fn project_session_snapshot_for_segment( .metadata .input_metadata .as_ref() - .map(crate::HistoryInputMetadata::public_use_policy), + .map(crate::HistoryInputMetadata::public_system_use_policy), )), LogEntry::RunYielded { ts, @@ -1795,11 +1795,19 @@ fn history_entry( data: SessionSnapshotEntryData, ) -> SessionSnapshotEntry { SessionSnapshotEntry { - use_policy: entry - .metadata - .input_metadata - .as_ref() - .map(crate::HistoryInputMetadata::public_use_policy), + use_policy: entry.metadata.input_metadata.as_ref().map(|input| { + if matches!( + entry.item, + LoggedItem::Message { + role: LoggedRole::System, + .. + } + ) { + input.public_system_use_policy() + } else { + input.public_use_policy() + } + }), entry_id: entry.metadata.entry_id.0.clone(), timestamp, provenance: provenance(&entry.metadata.origin), @@ -1927,21 +1935,29 @@ pub fn session_tool_attachment_id( fn project_history_item(entry: &LoggedHistoryEntry) -> Option { if let Some(input) = &entry.metadata.input_metadata { - if !matches!( + if matches!( entry.item, LoggedItem::Message { role: LoggedRole::System, .. } ) { + let public = input.public_projection(); + return Some(SessionSnapshotEntryData::SystemItem { + item_kind: public.kind.clone(), + content: public.body.clone(), + data: Some( + serde_json::to_value(public).expect("public system input is serializable"), + ), + }); + } + let usage = input.usage(); + if usage.recording_material == crate::InputEvidenceUse::Exclude + && usage.memory_evidence == crate::InputEvidenceUse::Exclude + && usage.observation == crate::InputObservationUse::Hidden + { return None; } - let public = input.public_projection(); - return Some(SessionSnapshotEntryData::SystemItem { - item_kind: public.kind.clone(), - content: public.body.clone(), - data: Some(serde_json::to_value(public).expect("public system input is serializable")), - }); } project_item(&entry.metadata.entry_id.0, &entry.item) } diff --git a/crates/worker/src/session_capture.rs b/crates/worker/src/session_capture.rs index e53da9105..c255d7dad 100644 --- a/crates/worker/src/session_capture.rs +++ b/crates/worker/src/session_capture.rs @@ -353,18 +353,21 @@ impl SessionCapture { .into_iter() .map(|entry| { let (item, use_policy) = if let Some(input) = &entry.annotation.input_metadata { - let item = if matches!( + let is_system = matches!( entry.item, Item::Message { role: Role::System, .. } - ) { - Item::system_message(input.public_projection().body) + ); + if is_system { + ( + Item::system_message(input.public_projection().body), + Some(input.public_system_use_policy()), + ) } else { - entry.item - }; - (item, Some(input.public_use_policy())) + (entry.item, Some(input.public_use_policy())) + } } else { (entry.item, None) }; @@ -1132,147 +1135,181 @@ mod tests { // Real FsStore seed/fork/reopen -> ordinary history replay, public snapshot, // capture and both evidence consumers; no Feature or Tool execution occurs. #[test] - fn derived_tool_context_keeps_roles_identity_and_policy_through_fork_without_becoming_evidence() - { + fn tool_policy_axes_survive_restore_and_fork_without_exposing_derived_context() { use session_store::public_material::{ PublicMaterialLimits, PublicMaterialUse, SessionPublicMaterialReader, }; use session_store::*; - let root = tempfile::tempdir().unwrap(); - let store = FsStore::new(root.path()).unwrap(); - let session = new_session_id(); - let first = new_segment_id(); - let fork = new_segment_id(); - let items = vec![ - Item::tool_call("original", "ArbitraryTool", "{}"), - Item::tool_result("original", "original result"), - Item::tool_call("derived", "ArbitraryTool", "PRIVATE_SYNTHETIC_CONTEXT"), - Item::tool_result("derived", "PRIVATE_SYNTHETIC_RESULT"), - ]; - let mut history: Vec<_> = items - .iter() - .cloned() - .map(|item| LoggedHistoryEntry { - item: LoggedItem::from(item), - metadata: LoggedSessionHistoryMetadata::legacy_unknown(), - }) - .collect(); - let sources = history[..2] - .iter() - .map(|entry| entry.metadata.entry_id.clone()) - .collect::>(); - for entry in &mut history[2..] { - entry.metadata.derivation = Some(LoggedHistoryDerivation { - sources: sources.clone(), - }); - entry.metadata.input_metadata = Some( - HistoryInputMetadata::new( - HistoryInputUsePolicy { - recording_material: InputEvidenceUse::Exclude, - memory_evidence: InputEvidenceUse::Exclude, - observation: InputObservationUse::Hidden, - compaction: InputCompactionUse::Retain, - }, - None, - None, - ) - .unwrap(), - ); - } - for segment in [first, fork] { - store - .create_segment( - session, - segment, - &[LogEntry::AnnotatedSegmentStart { - ts: 1, - session_id: session, - system_prompt: None, - config: Default::default(), - history: history.clone(), - forked_from: (segment == fork).then_some(SegmentOrigin { - segment_id: first, - at_turn_index: 0, - }), - compacted_from: None, - }], - ) - .unwrap(); - } - let reopened = FsStore::new(root.path()).unwrap(); - let saved = reopened.read_all(session, fork).unwrap(); - let restored = - crate::session_history::restore_history_entries(session, fork, &saved).unwrap(); - for (index, entry) in restored.iter().enumerate() { - assert_eq!(entry.item, items[index]); - assert_eq!(entry.annotation, history[index].metadata); - } - let capture = SessionCapture::from_history_entries(fork.to_string(), restored); - assert!( - matches!(&capture.entries[2].item, Item::ToolCall { call_id, .. } if call_id == "derived") - ); - assert!( - matches!(&capture.entries[3].item, Item::ToolResult { call_id, .. } if call_id == "derived") - ); - let snapshot = public_snapshot::project_session_snapshot(session, &saved); - assert!( - !serde_json::to_string(&snapshot) - .unwrap() - .contains("PRIVATE_SYNTHETIC") - ); - for capture in [ - capture, - SessionCapture::from_session_snapshot(fork.to_string(), snapshot), - ] { - assert_eq!( - capture - .search(&SearchOptions { - kind: Some(ReferenceKind::Tool), - ..Default::default() - }) - .len(), - 2 - ); - for (index, entry) in history.iter().enumerate() { - let reference = SessionEntryRef::from_history_entry_id(&entry.metadata.entry_id); - assert_eq!( - !capture - .read(ReadSelector::Id(reference.as_str()), ReadOptions::default()) - .entries - .is_empty(), - index < 2 - ); - assert_eq!( - capture.evidence_for(reference.as_str()).is_some(), - index < 2 - ); + for recording in [InputEvidenceUse::Include, InputEvidenceUse::Exclude] { + for memory in [InputEvidenceUse::Include, InputEvidenceUse::Exclude] { + for observation in [InputObservationUse::Readable, InputObservationUse::Hidden] { + let root = tempfile::tempdir().unwrap(); + let store = FsStore::new(root.path()).unwrap(); + let session = new_session_id(); + let first = new_segment_id(); + let fork = new_segment_id(); + let items = vec![ + Item::tool_call("original", "ArbitraryTool", "{}"), + Item::tool_result("original", "original result"), + Item::tool_call("derived", "ArbitraryTool", "PRIVATE_SYNTHETIC_CONTEXT"), + Item::tool_result("derived", "PRIVATE_SYNTHETIC_RESULT"), + ]; + let mut history: Vec<_> = items + .iter() + .cloned() + .map(|item| LoggedHistoryEntry { + item: LoggedItem::from(item), + metadata: LoggedSessionHistoryMetadata::legacy_unknown(), + }) + .collect(); + let sources = history[..2] + .iter() + .map(|entry| entry.metadata.entry_id.clone()) + .collect::>(); + for entry in &mut history[..2] { + entry.metadata.input_metadata = Some( + HistoryInputMetadata::new( + HistoryInputUsePolicy { + recording_material: recording, + memory_evidence: memory, + observation, + compaction: InputCompactionUse::Retain, + }, + None, + None, + ) + .unwrap(), + ); + } + for entry in &mut history[2..] { + entry.metadata.derivation = Some(LoggedHistoryDerivation { + sources: sources.clone(), + }); + entry.metadata.input_metadata = Some( + HistoryInputMetadata::new( + HistoryInputUsePolicy { + recording_material: InputEvidenceUse::Exclude, + memory_evidence: InputEvidenceUse::Exclude, + observation: InputObservationUse::Hidden, + compaction: InputCompactionUse::Retain, + }, + None, + None, + ) + .unwrap(), + ); + } + for segment in [first, fork] { + store + .create_segment( + session, + segment, + &[LogEntry::AnnotatedSegmentStart { + ts: 1, + session_id: session, + system_prompt: None, + config: Default::default(), + history: history.clone(), + forked_from: (segment == fork).then_some(SegmentOrigin { + segment_id: first, + at_turn_index: 0, + }), + compacted_from: None, + }], + ) + .unwrap(); + } + let reopened = FsStore::new(root.path()).unwrap(); + let saved = reopened.read_all(session, fork).unwrap(); + let restored = + crate::session_history::restore_history_entries(session, fork, &saved) + .unwrap(); + for (index, entry) in restored.iter().enumerate() { + assert_eq!(entry.item, items[index]); + assert_eq!(entry.annotation, history[index].metadata); + } + let capture = SessionCapture::from_history_entries(fork.to_string(), restored); + assert!( + matches!(&capture.entries[2].item, Item::ToolCall { call_id, .. } if call_id == "derived") + ); + assert!( + matches!(&capture.entries[3].item, Item::ToolResult { call_id, .. } if call_id == "derived") + ); + let snapshot = public_snapshot::project_session_snapshot(session, &saved); + assert!( + !serde_json::to_string(&snapshot) + .unwrap() + .contains("PRIVATE_SYNTHETIC") + ); + for capture in [ + capture, + SessionCapture::from_session_snapshot(fork.to_string(), snapshot), + ] { + assert_eq!( + capture + .search(&SearchOptions { + kind: Some(ReferenceKind::Tool), + ..Default::default() + }) + .len(), + if observation == InputObservationUse::Readable { + 2 + } else { + 0 + } + ); + for (index, entry) in history.iter().enumerate() { + let reference = + SessionEntryRef::from_history_entry_id(&entry.metadata.entry_id); + assert_eq!( + !capture + .read( + ReadSelector::Id(reference.as_str()), + ReadOptions::default() + ) + .entries + .is_empty(), + index < 2 && observation == InputObservationUse::Readable + ); + assert_eq!( + capture.evidence_for(reference.as_str()).is_some(), + index < 2 && memory == InputEvidenceUse::Include + ); + } + } + for purpose in [ + PublicMaterialUse::Recording, + PublicMaterialUse::MemoryEvidence, + ] { + let reader = SessionPublicMaterialReader::open_fs_for_use( + root.path(), + session, + fork, + PublicMaterialLimits::default(), + reopened.retention(), + purpose, + ) + .unwrap(); + let ids = reader + .descriptors() + .map(|entry| entry.identity.entry_ref.clone()) + .collect::>(); + let included = match purpose { + PublicMaterialUse::Recording => recording, + PublicMaterialUse::MemoryEvidence => memory, + } == InputEvidenceUse::Include; + assert_eq!( + ids, + sources + .iter() + .filter(|_| included) + .map(|id| format!("E{}", id.0)) + .collect::>() + ); + } + } } } - for purpose in [ - PublicMaterialUse::Recording, - PublicMaterialUse::MemoryEvidence, - ] { - let reader = SessionPublicMaterialReader::open_fs_for_use( - root.path(), - session, - fork, - PublicMaterialLimits::default(), - reopened.retention(), - purpose, - ) - .unwrap(); - let ids = reader - .descriptors() - .map(|entry| entry.identity.entry_ref.clone()) - .collect::>(); - assert_eq!( - ids, - sources - .iter() - .map(|id| format!("E{}", id.0)) - .collect::>() - ); - } } #[test] diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 09d2159ff..955835c1b 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -6959,7 +6959,7 @@ impl Worker { // The old summary is consumed in full by the compaction service even when // the cut falls before it. The replacement summary supersedes it: retaining // it as well would duplicate inherited context on every compact. - let retained_history_entries = history_entries[retain_from..] + let mut retained_history_entries = history_entries[retain_from..] .iter() .filter(|entry| { crate::session_history::compaction_use(&entry.annotation) @@ -6967,16 +6967,16 @@ impl Worker { }) .cloned() .collect::>(); - let mut retained_items = retained_history_entries - .iter() - .map(|entry| entry.item.clone()) - .collect::>(); let default_workdir_alias = (self.workdir_sessions.len() == 1) .then(|| self.workdir_sessions.aliases().into_iter().next()) .flatten() .map(|alias| alias.as_str().to_string()); - let retained_image_references = - project_compaction_attachments(&mut retained_items, default_workdir_alias.as_deref()); + let retained_image_references = project_compaction_attachments( + retained_history_entries + .iter_mut() + .map(|entry| &mut entry.item), + default_workdir_alias.as_deref(), + ); // The compactor explores summaries, text details, and ViewImage call arguments. It // never needs durable image bodies; relevant images are nominated with add_reference. let mut entries_to_summarise = history_entries[..retain_from] @@ -7326,8 +7326,14 @@ impl Worker { Vec::new() }; let auto_read_tokens = agen::token_counter::total_tokens(&auto_read_messages, &[]).tokens; - let retained_estimate = agen::token_counter::total_tokens(&retained_items, &[]); - let retained_item_count = u64::try_from(retained_items.len()).unwrap_or(u64::MAX); + let retained_estimate = agen::token_counter::total_tokens( + &retained_history_entries + .iter() + .map(|entry| entry.item.clone()) + .collect::>(), + &[], + ); + let retained_item_count = u64::try_from(retained_history_entries.len()).unwrap_or(u64::MAX); let summarized_item_count = u64::try_from(items_to_summarise.len()).unwrap_or(u64::MAX); let compactor_entries = handle.entries(); let compactor_turns = compactor_entries @@ -7382,25 +7388,19 @@ impl Worker { let reference_message = build_related_file_message(&final_ctx.references, &retained_image_references); - // Count surviving user_messages before consuming `retained_items` + // Count surviving user_messages before consuming the annotated retained tail // — needed to align `self.user_segments` after the swap below. - let retained_user_msgs = retained_items + let retained_user_msgs = retained_history_entries .iter() - .filter(|i| i.is_user_message()) + .filter(|entry| entry.item.is_user_message()) .count(); // Build new history: [summary, ...auto-read, references, ...retained, task snapshot, TaskList synthetic call/result]. // The TaskStore snapshot trails the retained items so that, on resume, // `replay_history` walks any pre-compact Task* calls preserved verbatim - // in retained_items first and the trailing snapshot's `replace_with` + // in the retained tail first and the trailing snapshot's `replace_with` // is the final word — pre-compact `TaskCreate` calls cannot leak as // duplicate entries. - let mut new_history = Vec::with_capacity( - 1 + auto_read_messages.len() - + 3 - + reference_message.is_some() as usize - + retained_items.len(), - ); let mut compact_introduced_system_messages = Vec::with_capacity(2 + auto_read_messages.len() + reference_message.is_some() as usize); let summary_message = Item::system_message(format!( @@ -7424,49 +7424,13 @@ impl Worker { )); compact_introduced_system_messages.push(task_snapshot_message.clone()); - new_history.push(summary_message); - new_history.extend(context_messages); - new_history.extend(auto_read_messages); - if let Some(msg) = reference_message { - new_history.push(msg); - } - new_history.extend(retained_items); - new_history.push(task_snapshot_message); - new_history.push(Item::tool_call("compact-tasklist", "TaskList", "{}")); - new_history.push(Item::tool_result_with_content( - "compact-tasklist", - self.task_feature.snapshot_overview(), - task_snapshot_text.clone(), - )); - let result_estimate = agen::token_counter::total_tokens(&new_history, &[]); - if result_context_max_tokens > 0 && result_estimate.tokens > result_context_max_tokens { - // This history estimate includes retained items and auto-read files, not - // just the summary. Treat the configured result target as advisory; - // request-time context safety checks still apply after activation. - warn!( - result_tokens = result_estimate.tokens, - warning_threshold = result_context_max_tokens, - retained_tokens = retained_estimate.tokens, - summary_tokens, - auto_read_tokens, - "compacted result context exceeds warning threshold; continuing" - ); - self.alert( - AlertLevel::Warn, - AlertSource::Compactor, - format!( - "compacted result context is larger than expected (≈{} tokens; warning threshold {}); continuing compaction", - result_estimate.tokens, result_context_max_tokens - ), - ); - } let original_entries = self.session.history().entries(); let derived_sources = original_entries .iter() .map(|entry| entry.annotation.entry_id.clone()) .collect::>(); let generic_summary_input = session_store::SystemInputEnvelope::new( - new_history[0] + summary_message .as_text() .expect("compaction summary is text"), session_store::HistoryInputMetadata::new( @@ -7491,47 +7455,79 @@ impl Worker { }), generic_summary_input, ); - let mut original_cursor = 0usize; - let compacted_history_entries = new_history + let derived_entry = |item| { + let mut metadata = new_history_metadata( + WorkerHistoryProvenance::DerivedSummary, + Some(SessionHistoryDerivation { + sources: derived_sources.clone(), + }), + ); + metadata.input_metadata = Some( + session_store::HistoryInputMetadata::new( + session_store::HistoryInputUsePolicy { + recording_material: session_store::InputEvidenceUse::Exclude, + memory_evidence: session_store::InputEvidenceUse::Exclude, + observation: session_store::InputObservationUse::Hidden, + compaction: session_store::InputCompactionUse::Retain, + }, + None, + None, + ) + .expect("bounded private derived context policy"), + ); + HistoryEntry::new(item, metadata) + }; + // Retained items and annotations travel together, including when attachment + // projection changes an item. Equal model text never selects an identity. + let mut compacted_history_entries = + vec![HistoryEntry::new(summary_message, summary_entry.metadata)]; + compacted_history_entries.extend( + frozen_context.iter().map(|entry| { + HistoryEntry::new(entry.item.to_history_item(), entry.metadata.clone()) + }), + ); + compacted_history_entries.extend(auto_read_messages.into_iter().map(&derived_entry)); + if let Some(item) = reference_message { + compacted_history_entries.push(derived_entry(item)); + } + compacted_history_entries.extend(retained_history_entries); + compacted_history_entries.push(derived_entry(task_snapshot_message)); + compacted_history_entries.push(derived_entry(Item::tool_call( + "compact-tasklist", + "TaskList", + "{}", + ))); + compacted_history_entries.push(derived_entry(Item::tool_result_with_content( + "compact-tasklist", + self.task_feature.snapshot_overview(), + task_snapshot_text.clone(), + ))); + let new_history = compacted_history_entries .iter() - .cloned() - .enumerate() - .map(|(index, item)| { - if index == 0 { - HistoryEntry::new(item, summary_entry.metadata.clone()) - } else if let Some(context) = frozen_context.get(index - 1) { - HistoryEntry::new(item, context.metadata.clone()) - } else if let Some((offset, original)) = original_entries[original_cursor..] - .iter() - .enumerate() - .find(|(_, original)| original.item == item) - { - original_cursor += offset + 1; - HistoryEntry::new(item, original.annotation.clone()) - } else { - let mut metadata = new_history_metadata( - WorkerHistoryProvenance::DerivedSummary, - Some(SessionHistoryDerivation { - sources: derived_sources.clone(), - }), - ); - metadata.input_metadata = Some( - session_store::HistoryInputMetadata::new( - session_store::HistoryInputUsePolicy { - recording_material: session_store::InputEvidenceUse::Exclude, - memory_evidence: session_store::InputEvidenceUse::Exclude, - observation: session_store::InputObservationUse::Hidden, - compaction: session_store::InputCompactionUse::Retain, - }, - None, - None, - ) - .expect("bounded private derived context policy"), - ); - HistoryEntry::new(item, metadata) - } - }) + .map(|entry| entry.item.clone()) .collect::>(); + let result_estimate = agen::token_counter::total_tokens(&new_history, &[]); + if result_context_max_tokens > 0 && result_estimate.tokens > result_context_max_tokens { + // This history estimate includes retained items and auto-read files, not + // just the summary. Treat the configured result target as advisory; + // request-time context safety checks still apply after activation. + warn!( + result_tokens = result_estimate.tokens, + warning_threshold = result_context_max_tokens, + retained_tokens = retained_estimate.tokens, + summary_tokens, + auto_read_tokens, + "compacted result context exceeds warning threshold; continuing" + ); + self.alert( + AlertLevel::Warn, + AlertSource::Compactor, + format!( + "compacted result context is larger than expected (≈{} tokens; warning threshold {}); continuing compaction", + result_estimate.tokens, result_context_max_tokens + ), + ); + } let retained_user_segments = self .user_segments @@ -9238,13 +9234,14 @@ fn project_compactor_capture_attachments(entries: &mut [HistoryEntry]) -> /// Remove every durable binary attachment from the compacted tail and preserve /// re-readable ViewImage origins as related-file references. -fn project_compaction_attachments( - items: &mut [Item], +fn project_compaction_attachments<'a>( + items: impl IntoIterator, default_workdir_alias: Option<&str>, ) -> Vec { + let items = items.into_iter().collect::>(); let view_image_calls = items .iter() - .filter_map(|item| match item { + .filter_map(|item| match &**item { Item::ToolCall { call_id, name, @@ -10870,7 +10867,7 @@ mod build_summary_prompt_tests { Item::ToolResult { attachments, .. } if attachments.is_empty() )); - let references = project_compaction_attachments(&mut items, Some("main")); + let references = project_compaction_attachments(items.iter_mut(), Some("main")); assert_eq!( references, diff --git a/crates/worker/tests/compact_events_test.rs b/crates/worker/tests/compact_events_test.rs index c294f54aa..0869c0b5c 100644 --- a/crates/worker/tests/compact_events_test.rs +++ b/crates/worker/tests/compact_events_test.rs @@ -3655,3 +3655,132 @@ async fn journal_snapshot_counts_against_compactor_hard_input_cap_without_mutati ); } } +#[tokio::test] +async fn identical_reference_and_retained_inputs_preserve_distinct_metadata_after_compaction() { + use session_store::*; + let client = MockClient::new(vec![ + single_text_events("fixture"), + single_text_events("fork checkpoint"), + write_summary_tool_use_events("review-summary", "replacement summary"), + single_text_events("done"), + single_text_events("restore checkpoint"), + ]); + let mut fixture = SummaryInputFixture::new(MANUAL_ONLY_MANIFEST_TOML, client).await; + let make = |producer: &str, compaction| { + LoggedSystemHistoryEntry::from_input( + LoggedSessionHistoryOrigin::Provided { + producer: producer.into(), + }, + None, + SystemInputEnvelope::new( + "identical model context", + HistoryInputMetadata::new( + HistoryInputUsePolicy { + recording_material: InputEvidenceUse::Exclude, + memory_evidence: InputEvidenceUse::Exclude, + observation: InputObservationUse::Hidden, + compaction, + }, + Some( + FeatureInputPayload::encode( + "review:owner", + 1, + &serde_json::json!({"producer": producer}), + ) + .unwrap(), + ), + None, + ) + .unwrap(), + ) + .unwrap(), + ) + .to_history_entry() + }; + let historical = make("old-reference", InputCompactionUse::ReferenceInput); + let retained = make("retained-context", InputCompactionUse::Retain); + let repeated = make("second-retained-context", InputCompactionUse::Retain); + assert_ne!(historical.metadata.entry_id, retained.metadata.entry_id); + fixture + .restore_seed( + vec![historical.clone(), retained.clone(), repeated.clone()], + false, + ) + .await; + let session = fixture.worker.session_id(); + let original = fixture.worker.segment_id(); + fixture + .worker + .store() + .append( + session, + original, + &LogEntry::Extension { + ts: 2, + domain: "test.fork_drift".into(), + payload: serde_json::json!({}), + }, + ) + .unwrap(); + fixture.worker.run_text("fork checkpoint").await.unwrap(); + let fork = fixture.worker.segment_id(); + assert_ne!(fork, original); + let forked = replacement_seed(&fixture.worker); + for expected in [&historical, &retained, &repeated] { + assert!(forked.contains(expected)); + } + fixture.worker.compact(1_000_000).await.unwrap(); + let seed = replacement_seed(&fixture.worker); + let matching: Vec<_> = seed.iter().filter(|e| e.item == retained.item).collect(); + assert_eq!(matching.len(), 2); + assert_eq!( + matching[0].metadata, retained.metadata, + "a retained logical entry must not inherit identity/policy/payload from an equal-body reference that was removed" + ); + + assert_eq!(matching[1].metadata, repeated.metadata); + let compacted = fixture.worker.segment_id(); + let reopened = FsStore::new(fixture._store_dir.path()).unwrap(); + let records = reopened.read_all(session, compacted).unwrap(); + assert!( + matches!(&records[0], LogEntry::AnnotatedSegmentStart { compacted_from: Some(source), .. } if source.segment_id == fork) + ); + fixture.worker.run_text("restore checkpoint").await.unwrap(); + let records = fixture.worker.store().read_all(session, compacted).unwrap(); + let snapshot = public_snapshot::project_current_session_snapshot(&records); + let checkpoint = snapshot + .entries + .iter() + .rev() + .find(|entry| { + matches!( + entry.data, + protocol::SessionSnapshotEntryData::UserInput { .. } + ) + }) + .unwrap(); + fixture + .worker + .rewind_to(checkpoint.entry_id.clone()) + .await + .unwrap(); + for expected in [&retained, &repeated] { + assert!( + fixture + .worker + .session_history() + .entries() + .iter() + .any(|entry| entry.annotation == expected.metadata + && entry.item == Item::from(expected.item.clone())) + ); + } + assert!( + !fixture + .worker + .session_history() + .entries() + .iter() + .any(|entry| entry.annotation.entry_id == historical.metadata.entry_id) + ); +}