Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: retain annotated compaction history and honor non-System input policy

Browse Source
This commit is contained in:
Hare
2026-10-20 05:34:22 +09:00
parent 90e38551cf
commit 8017c55d82
6 changed files with 446 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/session-store/src/input_metadata.rs
+13 -4
View File
@@ -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,
}
}
}
crates/session-store/src/public_material.rs
+1 -2
View File
@@ -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 {
crates/session-store/src/public_snapshot.rs
+29 -13
View File
@@ -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<SessionSnapshotEntryData> {
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)
}
crates/worker/src/session_capture.rs
+178 -141
View File
@@ -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::<Vec<_>>();
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::<Vec<_>>();
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::<Vec<_>>();
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::<Vec<_>>()
);
}
}
}
}
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::<Vec<_>>();
assert_eq!(
ids,
sources
.iter()
.map(|id| format!("E{}", id.0))
.collect::<Vec<_>>()
);
}
}
#[test]
crates/worker/src/worker.rs
+96 -99
View File
@@ -6959,7 +6959,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
// 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<C: LlmClient + 'static, St: Store> Worker<C, St> {
})
.cloned()
.collect::<Vec<_>>();
let mut retained_items = retained_history_entries
.iter()
.map(|entry| entry.item.clone())
.collect::<Vec<_>>();
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<C: LlmClient + 'static, St: Store> Worker<C, St> {
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::<Vec<_>>(),
&[],
);
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<C: LlmClient + 'static, St: Store> Worker<C, St> {
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<C: LlmClient + 'static, St: Store> Worker<C, St> {
));
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::<Vec<_>>();
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<C: LlmClient + 'static, St: Store> Worker<C, St> {
}),
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::<Vec<_>>();
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<A>(entries: &mut [HistoryEntry<A>]) ->
/// 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<Item = &'a mut Item>,
default_workdir_alias: Option<&str>,
) -> Vec<CompactionImageReference> {
let items = items.into_iter().collect::<Vec<_>>();
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,
crates/worker/tests/compact_events_test.rs
+129
View File
@@ -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)
);
}