From 4976b6fd00eab90216197022c55e0d724cc5813e Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 30 Sep 2026 10:56:01 +0900 Subject: [PATCH] fix(console): bound retained history reconciliation --- crates/protocol/src/lib.rs | 6 + crates/session-store/src/public_snapshot.rs | 1095 ++++++++++++++--- .../session-store/src/worker_session_store.rs | 313 ++++- .../console_history_fixture_server.ts | 8 +- .../console_history_shell_test.ts | 16 +- web/workspace/src/lib/generated/protocol.ts | 9 +- .../src/lib/workspace/console/history.test.ts | 151 ++- .../src/lib/workspace/console/history.ts | 95 +- .../src/lib/workspace/console/model.test.ts | 46 + .../src/lib/workspace/console/model.ts | 10 +- .../workers/[workerRef]/console/+page.svelte | 66 +- .../console/console-display.browser.spec.ts | 145 ++- 12 files changed, 1725 insertions(+), 235 deletions(-) diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 11e1ca1ba..851bead6e 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -853,6 +853,12 @@ pub struct SessionHistoryPage { pub session_id: String, /// Stable identity of the adopted active lineage used to fence stale pages. pub lineage_id: String, + /// Older lineage identities that the active lineage fully adopted through + /// one or more consecutive Compact rotations. Clients may retain already + /// loaded turns when their prior lineage appears here, while replacing the + /// cursor with this page's boundary. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub compact_ancestor_lineage_ids: Vec, pub turns: Vec, #[serde(default, skip_serializing_if = "Option::is_none")] pub next_cursor: Option, diff --git a/crates/session-store/src/public_snapshot.rs b/crates/session-store/src/public_snapshot.rs index 1b68daebd..1b9af6333 100644 --- a/crates/session-store/src/public_snapshot.rs +++ b/crates/session-store/src/public_snapshot.rs @@ -10,7 +10,7 @@ use protocol::{ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; -use std::collections::{HashMap, HashSet}; +use std::collections::HashSet; use std::path::Path; use thiserror::Error; @@ -63,8 +63,9 @@ pub const DEFAULT_RETAINED_HISTORY_MAX_RESPONSE_BYTES: u64 = 8 * 1024 * 1024; /// Explicit resource bounds for one retained-history page read. /// -/// The operation never truncates a turn to fit these limits. A lineage or one -/// selected page that exceeds a bound fails with [`RetainedHistoryReadError::ResourceLimit`]. +/// The operation never truncates a turn to fit these limits. Bounded lineage +/// metadata or one selected page that exceeds a bound fails with +/// [`RetainedHistoryReadError::ResourceLimit`]. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct RetainedHistoryReadLimits { pub max_scan_bytes: u64, @@ -108,15 +109,37 @@ struct RetainedHistoryCursor { version: u8, worker_name: String, session_id: String, + active_segment_id: String, lineage_id: String, before_turn_id: String, + segment_id: String, + before_offset: u64, + record_end_offset: Option, + seed_entry_index: Option, } #[derive(Debug)] struct LoadedLineageSegment { segment_id: SegmentId, adopted_through_turn: Option, - entries: Vec, + file_len: u64, + seed_end_offset: u64, + origin: Option, + seed_entries: Vec, +} + +#[derive(Debug, Clone)] +struct HistoryTurnPosition { + segment_id: SegmentId, + before_offset: u64, + record_end_offset: Option, + seed_entry_index: Option, +} + +#[derive(Debug)] +struct PositionedHistoryTurn { + turn: SessionConversationTurn, + position: HistoryTurnPosition, } #[derive(Debug, Clone, Copy)] @@ -125,7 +148,7 @@ enum LineageOriginKind { Compact, } -#[derive(Debug)] +#[derive(Debug, Clone)] struct LineageOrigin { kind: LineageOriginKind, origin: SegmentOrigin, @@ -207,50 +230,82 @@ pub fn read_retained_session_history_page( .ok_or(RetainedHistoryReadError::ActivePointerMissing)?; let store = WorkerSessionStore::open_read_only(aggregate_root.join("session")) .map_err(map_history_store_error)?; - let (lineage, lineage_id) = + let (lineage, lineage_id, metadata_bytes, metadata_entries) = load_adopted_lineage(&store, active.session_id, active_segment_id, limits)?; - let entries = project_adopted_lineage(active.session_id, &lineage)?; - let turns = group_conversation_turns(entries); - - let page_end = match cursor { - Some(cursor) => { - let cursor = decode_history_cursor(cursor)?; - if cursor.version != 1 - || cursor.worker_name != worker_name - || cursor.session_id != active.session_id.to_string() - || cursor.lineage_id != lineage_id - { - return Err(RetainedHistoryReadError::InvalidCursor); - } - turns + let cursor = cursor.map(decode_history_cursor).transpose()?; + let start = if let Some(cursor) = cursor.as_ref() { + if cursor.version != 2 + || cursor.worker_name != worker_name + || cursor.session_id != active.session_id.to_string() + || cursor.active_segment_id != active_segment_id.to_string() + || cursor.lineage_id != lineage_id + || !lineage .iter() - .position(|turn| turn.turn_id == cursor.before_turn_id) - .ok_or(RetainedHistoryReadError::InvalidCursor)? + .any(|segment| segment.segment_id.to_string() == cursor.segment_id) + { + return Err(RetainedHistoryReadError::InvalidCursor); } - None => turns.len(), + let segment_id = cursor + .segment_id + .parse() + .map_err(|_| RetainedHistoryReadError::InvalidCursor)?; + Some(( + HistoryTurnPosition { + segment_id, + before_offset: cursor.before_offset, + record_end_offset: cursor.record_end_offset, + seed_entry_index: cursor.seed_entry_index, + }, + cursor.before_turn_id.as_str(), + )) + } else { + None }; - let page_start = page_end.saturating_sub(page_limit); - let page_turns = turns[page_start..page_end].to_vec(); - let has_more = page_start > 0; - let next_cursor = has_more - .then(|| { - encode_history_cursor(&RetainedHistoryCursor { - version: 1, - worker_name: worker_name.to_owned(), - session_id: active.session_id.to_string(), - lineage_id: lineage_id.clone(), - before_turn_id: page_turns - .first() - .expect("a page with earlier history cannot be empty") - .turn_id - .clone(), - }) - }) - .transpose()?; + let mut turns = read_history_turns_backward( + &store, + active.session_id, + &lineage, + start, + page_limit.saturating_add(1), + RetainedHistoryReadLimits { + max_scan_bytes: limits.max_scan_bytes.saturating_sub(metadata_bytes), + max_entries: limits.max_entries.saturating_sub(metadata_entries), + ..limits + }, + )?; + let has_more = turns.len() > page_limit; + if has_more { + turns.truncate(page_limit); + } + let next_cursor = if has_more { + let oldest = turns + .last() + .expect("a page with earlier history cannot be empty"); + Some(encode_history_cursor(&RetainedHistoryCursor { + version: 2, + worker_name: worker_name.to_owned(), + session_id: active.session_id.to_string(), + active_segment_id: active_segment_id.to_string(), + lineage_id: lineage_id.clone(), + before_turn_id: oldest.turn.turn_id.clone(), + segment_id: oldest.position.segment_id.to_string(), + before_offset: oldest.position.before_offset, + record_end_offset: oldest.position.record_end_offset, + seed_entry_index: oldest.position.seed_entry_index, + })?) + } else { + None + }; + turns.reverse(); + let compact_ancestor_lineage_ids = compact_ancestor_lineage_ids(active.session_id, &lineage); let page = SessionHistoryPage { session_id: active.session_id.to_string(), lineage_id, - turns: page_turns, + compact_ancestor_lineage_ids, + turns: turns + .into_iter() + .map(|positioned| positioned.turn) + .collect(), next_cursor, has_more, }; @@ -268,17 +323,13 @@ fn load_adopted_lineage( session_id: SessionId, active_segment_id: SegmentId, limits: RetainedHistoryReadLimits, -) -> Result<(Vec, String), RetainedHistoryReadError> { +) -> Result<(Vec, String, u64, usize), RetainedHistoryReadError> { let mut lineage = Vec::new(); let mut visited = HashSet::new(); let mut segment_id = active_segment_id; let mut adopted_through_turn = None; let mut scanned_bytes = 0_u64; let mut scanned_entries = 0_usize; - let mut lineage_hash = Sha256::new(); - lineage_hash.update(b"yoi-retained-history-lineage-v1\0"); - lineage_hash.update(session_id.as_bytes()); - lineage_hash.update(active_segment_id.as_bytes()); loop { if lineage.len() >= limits.max_segments { @@ -291,84 +342,149 @@ fn load_adopted_lineage( .max_scan_bytes .checked_sub(scanned_bytes) .ok_or(RetainedHistoryReadError::ResourceLimit)?; - let (entries, segment_bytes) = store - .read_all_read_only_bounded(session_id, segment_id, remaining_scan_bytes) + let (first, segment_bytes, file_len) = store + .read_first_log_record_read_only_bounded(session_id, segment_id, remaining_scan_bytes) .map_err(map_history_store_error)?; scanned_bytes = scanned_bytes .checked_add(segment_bytes) .ok_or(RetainedHistoryReadError::ResourceLimit)?; - scanned_entries = scanned_entries - .checked_add(persisted_entry_units(&entries)) - .ok_or(RetainedHistoryReadError::ResourceLimit)?; - if scanned_entries > limits.max_entries { - return Err(RetainedHistoryReadError::ResourceLimit); - } - let origin = validate_segment_start(session_id, &entries)?; - - lineage_hash.update(segment_id.as_bytes()); - match adopted_through_turn { - Some(turn) => { - lineage_hash.update([1]); - lineage_hash.update((turn as u64).to_be_bytes()); + if let Some(first) = first.as_ref() { + scanned_entries = scanned_entries + .checked_add(persisted_entry_units(std::slice::from_ref(&first.entry))) + .ok_or(RetainedHistoryReadError::ResourceLimit)?; + if scanned_entries > limits.max_entries { + return Err(RetainedHistoryReadError::ResourceLimit); } - None => lineage_hash.update([0]), - } - if let Some(origin) = &origin { - lineage_hash.update(match origin.kind { - LineageOriginKind::Fork => [1], - LineageOriginKind::Compact => [2], - }); - lineage_hash.update(origin.origin.segment_id.as_bytes()); - lineage_hash.update((origin.origin.at_turn_index as u64).to_be_bytes()); - } else { - lineage_hash.update([0]); } + let origin = + validate_segment_start_entry(session_id, first.as_ref().map(|entry| &entry.entry))?; + let seed_entries = if origin.is_none() { + let mut seed_records = first + .as_ref() + .map(|first| vec![first.entry.clone()]) + .unwrap_or_default(); + if let Some(first) = first.as_ref() { + let remaining_scan_bytes = limits.max_scan_bytes.saturating_sub(scanned_bytes); + let (next, next_bytes) = store + .read_next_log_record_read_only_bounded( + session_id, + segment_id, + first.end_offset, + remaining_scan_bytes, + ) + .map_err(map_history_store_error)?; + scanned_bytes = scanned_bytes + .checked_add(next_bytes) + .ok_or(RetainedHistoryReadError::ResourceLimit)?; + if let Some(next) = next { + scanned_entries = scanned_entries + .checked_add(persisted_entry_units(std::slice::from_ref(&next.entry))) + .ok_or(RetainedHistoryReadError::ResourceLimit)?; + if scanned_entries > limits.max_entries { + return Err(RetainedHistoryReadError::ResourceLimit); + } + if matches!(&next.entry, LogEntry::InputSegmentsCheckpoint { .. }) { + seed_records.push(next.entry); + } + } + } + project_session_snapshot_for_segment(session_id, Some(segment_id), &seed_records) + .entries + .into_iter() + .filter(|entry| entry.provenance != SessionEntryProvenance::DerivedSummary) + .collect() + } else { + Vec::new() + }; lineage.push(LoadedLineageSegment { segment_id, adopted_through_turn, - entries, + file_len, + seed_end_offset: first.as_ref().map(|record| record.end_offset).unwrap_or(0), + origin: origin.clone(), + seed_entries, }); let Some(origin) = origin else { break; }; segment_id = origin.origin.segment_id; - adopted_through_turn = Some( - adopted_through_turn - .map(|boundary| boundary.min(origin.origin.at_turn_index)) - .unwrap_or(origin.origin.at_turn_index), - ); + adopted_through_turn = Some(origin.origin.at_turn_index); } - lineage.reverse(); - Ok((lineage, URL_SAFE_NO_PAD.encode(lineage_hash.finalize()))) + let lineage_id = lineage_identity(session_id, &lineage); + Ok((lineage, lineage_id, scanned_bytes, scanned_entries)) } -fn validate_segment_start( - expected_session_id: SessionId, - entries: &[LogEntry], -) -> Result, RetainedHistoryReadError> { - if entries.is_empty() { - // A freshly reserved active Segment can be empty until its first turn - // materializes SegmentStart. The Session manifest still supplies the - // same-Session fence, and an empty Segment has no parent lineage. - return Ok(None); +fn lineage_identity(session_id: SessionId, lineage: &[LoadedLineageSegment]) -> String { + let active_segment_id = lineage + .first() + .expect("an adopted lineage always contains its active Segment") + .segment_id; + let mut hash = Sha256::new(); + hash.update(b"yoi-retained-history-lineage-v2\0"); + hash.update(session_id.as_bytes()); + hash.update(active_segment_id.as_bytes()); + for (index, segment) in lineage.iter().enumerate() { + hash.update(segment.segment_id.as_bytes()); + let boundary = (index != 0) + .then_some(segment.adopted_through_turn) + .flatten(); + match boundary { + Some(turn) => { + hash.update([1]); + hash.update((turn as u64).to_be_bytes()); + } + None => hash.update([0]), + } + if let Some(origin) = &segment.origin { + hash.update(match origin.kind { + LineageOriginKind::Fork => [1], + LineageOriginKind::Compact => [2], + }); + hash.update(origin.origin.segment_id.as_bytes()); + hash.update((origin.origin.at_turn_index as u64).to_be_bytes()); + } else { + hash.update([0]); + } } - let Some(LogEntry::AnnotatedSegmentStart { + URL_SAFE_NO_PAD.encode(hash.finalize()) +} + +fn compact_ancestor_lineage_ids( + session_id: SessionId, + lineage: &[LoadedLineageSegment], +) -> Vec { + let mut ancestors = Vec::new(); + for index in 0..lineage.len().saturating_sub(1) { + if !matches!( + lineage[index].origin.as_ref().map(|origin| origin.kind), + Some(LineageOriginKind::Compact) + ) { + break; + } + ancestors.push(lineage_identity(session_id, &lineage[index + 1..])); + } + ancestors +} + +fn validate_segment_start_entry( + expected_session_id: SessionId, + entry: Option<&LogEntry>, +) -> Result, RetainedHistoryReadError> { + let Some(entry) = entry else { + return Ok(None); + }; + let LogEntry::AnnotatedSegmentStart { session_id, forked_from, compacted_from, .. - }) = entries.first() + } = entry else { return Err(RetainedHistoryReadError::CorruptLog); }; - if *session_id != expected_session_id - || entries - .iter() - .skip(1) - .any(|entry| matches!(entry, LogEntry::AnnotatedSegmentStart { .. })) - { + if *session_id != expected_session_id { return Err(RetainedHistoryReadError::CorruptLog); } match (forked_from, compacted_from) { @@ -399,113 +515,333 @@ fn persisted_entry_units(entries: &[LogEntry]) -> usize { }) } -fn project_adopted_lineage( +fn locate_adopted_segment_end( + store: &WorkerSessionStore, session_id: SessionId, - lineage: &[LoadedLineageSegment], -) -> Result, RetainedHistoryReadError> { - let mut entries = Vec::::new(); - let mut positions = HashMap::::new(); - let mut excluded_by_boundary = HashSet::::new(); + segment: &LoadedLineageSegment, + limits: RetainedHistoryReadLimits, + scanned_bytes: &mut u64, + scanned_entries: &mut usize, +) -> Result { + const FORWARD_BOUNDARY_TURN_THRESHOLD: usize = 64; - for segment in lineage { - let prefix = adopted_segment_prefix(&segment.entries, segment.adopted_through_turn)?; - let snapshot = - project_session_snapshot_for_segment(session_id, Some(segment.segment_id), prefix); - if prefix.len() < segment.entries.len() { - let adopted_ids = snapshot - .entries - .iter() - .map(|entry| entry.entry_id.as_str()) - .collect::>(); - for entry in project_session_snapshot_for_segment( - session_id, - Some(segment.segment_id), - &segment.entries, - ) - .entries - { - if !adopted_ids.contains(entry.entry_id.as_str()) { - excluded_by_boundary.insert(entry.entry_id); - } - } - } - for mut entry in snapshot.entries { - if entry.provenance == SessionEntryProvenance::DerivedSummary - || excluded_by_boundary.contains(&entry.entry_id) - { - continue; - } - if let Some(position) = positions.get(&entry.entry_id).copied() { - // A newer seed may carry a richer canonical projection (notably - // typed input segments). Keep the original durable timestamp and - // ordering while replacing the duplicate representation. - entry.timestamp = entries[position].timestamp; - entries[position] = entry; - } else { - positions.insert(entry.entry_id.clone(), entries.len()); - entries.push(entry); - } - } - } - Ok(entries) -} - -fn adopted_segment_prefix( - entries: &[LogEntry], - adopted_through_turn: Option, -) -> Result<&[LogEntry], RetainedHistoryReadError> { - let Some(boundary) = adopted_through_turn else { - return Ok(entries); + let Some(boundary) = segment.adopted_through_turn else { + return Ok(segment.file_len); }; - let seed_end = entries - .iter() - .position(|entry| { - !matches!( - entry, - LogEntry::AnnotatedSegmentStart { .. } | LogEntry::InputSegmentsCheckpoint { .. } - ) - }) - .unwrap_or(entries.len()); - if boundary == 0 { - return Ok(&entries[..seed_end]); + let inherited_turns = segment + .origin + .as_ref() + .map(|origin| origin.origin.at_turn_index) + .unwrap_or(0); + if boundary <= inherited_turns { + // This edge adopts no locally completed turns. The Segment's own + // origin remains an independent edge with its original boundary. + return Ok(segment.seed_end_offset); } - let mut current_turn_start = seed_end; - for (index, entry) in entries.iter().enumerate().skip(seed_end) { - match entry { - LogEntry::Invoke { .. } => current_turn_start = index, - LogEntry::TurnEnd { turn_count, .. } if *turn_count == boundary => { - return Ok(&entries[..=index]); + // Small/early boundaries are cheapest to resolve from the Segment seed and + // must not pay for an arbitrarily large unadopted suffix. Later boundaries + // are resolved backward so a large adopted prefix does not have to be read + // merely to find the nearby split point. + if boundary <= inherited_turns.saturating_add(FORWARD_BOUNDARY_TURN_THRESHOLD) { + let mut offset = segment.seed_end_offset; + loop { + let remaining = limits.max_scan_bytes.saturating_sub(*scanned_bytes); + let (record, bytes) = store + .read_next_log_record_read_only_bounded( + session_id, + segment.segment_id, + offset, + remaining, + ) + .map_err(map_history_store_error)?; + *scanned_bytes = scanned_bytes + .checked_add(bytes) + .ok_or(RetainedHistoryReadError::ResourceLimit)?; + let record = record.ok_or(RetainedHistoryReadError::CorruptLog)?; + *scanned_entries = scanned_entries + .checked_add(persisted_entry_units(std::slice::from_ref(&record.entry))) + .ok_or(RetainedHistoryReadError::ResourceLimit)?; + if *scanned_entries > limits.max_entries { + return Err(RetainedHistoryReadError::ResourceLimit); } - LogEntry::TurnEnd { turn_count, .. } if *turn_count > boundary => { - return Ok(&entries[..current_turn_start]); + offset = record.end_offset; + match record.entry { + LogEntry::TurnEnd { turn_count, .. } if turn_count == boundary => { + return Ok(offset); + } + LogEntry::TurnEnd { turn_count, .. } if turn_count > boundary => { + return Err(RetainedHistoryReadError::CorruptLog); + } + _ => {} + } + } + } + + let mut reader = store + .open_retained_segment_reader(session_id, segment.segment_id, segment.file_len) + .map_err(map_history_store_error)?; + loop { + let remaining = limits.max_scan_bytes.saturating_sub(*scanned_bytes); + let (record, bytes) = reader + .previous_record(remaining) + .map_err(map_history_store_error)?; + *scanned_bytes = scanned_bytes + .checked_add(bytes) + .ok_or(RetainedHistoryReadError::ResourceLimit)?; + let record = record.ok_or(RetainedHistoryReadError::CorruptLog)?; + *scanned_entries = scanned_entries + .checked_add(persisted_entry_units(std::slice::from_ref(&record.entry))) + .ok_or(RetainedHistoryReadError::ResourceLimit)?; + if *scanned_entries > limits.max_entries { + return Err(RetainedHistoryReadError::ResourceLimit); + } + match record.entry { + LogEntry::TurnEnd { turn_count, .. } if turn_count == boundary => { + return Ok(record.end_offset); + } + LogEntry::TurnEnd { turn_count, .. } if turn_count < boundary => { + return Err(RetainedHistoryReadError::CorruptLog); + } + LogEntry::AnnotatedSegmentStart { .. } => { + return Err(RetainedHistoryReadError::CorruptLog); } _ => {} } } - Ok(entries) } -fn group_conversation_turns(entries: Vec) -> Vec { +fn read_history_turns_backward( + store: &WorkerSessionStore, + session_id: SessionId, + lineage: &[LoadedLineageSegment], + start: Option<(HistoryTurnPosition, &str)>, + turn_limit: usize, + limits: RetainedHistoryReadLimits, +) -> Result, RetainedHistoryReadError> { + let mut scanned_bytes = 0_u64; + let mut scanned_entries = 0_usize; let mut turns = Vec::new(); - let mut current: Option = None; - for entry in entries { - if is_user_entry(&entry) { - if let Some(turn) = current.take() { - turns.push(turn); + let mut pending_entries = Vec::new(); + let mut seen_entries = HashSet::new(); + + let (start_segment_index, start_position, cursor_turn_id) = match start { + Some((position, turn_id)) => { + let index = lineage + .iter() + .position(|segment| segment.segment_id == position.segment_id) + .ok_or(RetainedHistoryReadError::InvalidCursor)?; + (index, Some(position), Some(turn_id)) + } + None => (0, None, None), + }; + + if let (Some(position), Some(turn_id)) = (start_position.as_ref(), cursor_turn_id) { + let segment = &lineage[start_segment_index]; + if let Some(seed_index) = position.seed_entry_index { + if start_segment_index + 1 != lineage.len() + || position.before_offset != 0 + || position.record_end_offset.is_some() + || segment + .seed_entries + .get(seed_index) + .is_none_or(|entry| entry.entry_id != turn_id || !is_user_entry(entry)) + { + return Err(RetainedHistoryReadError::InvalidCursor); + } + } else { + let record_end = position + .record_end_offset + .ok_or(RetainedHistoryReadError::InvalidCursor)?; + let remaining = limits.max_scan_bytes.saturating_sub(scanned_bytes); + let record = store + .read_log_record_range_read_only_bounded( + session_id, + segment.segment_id, + position.before_offset, + record_end, + remaining, + ) + .map_err(map_history_cursor_store_error)?; + scanned_bytes = scanned_bytes + .checked_add(record.end_offset - record.start_offset) + .ok_or(RetainedHistoryReadError::ResourceLimit)?; + if record.start_offset != position.before_offset + || !project_history_record(session_id, segment.segment_id, &record.entry) + .iter() + .any(|entry| entry.entry_id == turn_id && is_user_entry(entry)) + { + return Err(RetainedHistoryReadError::InvalidCursor); + } + scanned_entries = scanned_entries + .checked_add(persisted_entry_units(std::slice::from_ref(&record.entry))) + .ok_or(RetainedHistoryReadError::ResourceLimit)?; + if scanned_entries > limits.max_entries { + return Err(RetainedHistoryReadError::ResourceLimit); } - current = Some(SessionConversationTurn { - turn_id: entry.entry_id.clone(), - entries: vec![entry], - }); - } else if let Some(turn) = current.as_mut() { - turn.entries.push(entry); } } - if let Some(turn) = current { - turns.push(turn); + + for (segment_index, segment) in lineage.iter().enumerate().skip(start_segment_index) { + let starting_here = segment_index == start_segment_index; + if starting_here + && start_position + .as_ref() + .and_then(|position| position.seed_entry_index) + .is_some() + { + let seed_index = start_position + .as_ref() + .and_then(|position| position.seed_entry_index) + .expect("checked seed cursor"); + if collect_history_entries_backward( + &segment.seed_entries[..seed_index], + segment.segment_id, + 0, + None, + true, + turn_limit, + &mut seen_entries, + &mut pending_entries, + &mut turns, + ) { + return Ok(turns); + } + continue; + } + + let adopted_end = locate_adopted_segment_end( + store, + session_id, + segment, + limits, + &mut scanned_bytes, + &mut scanned_entries, + )?; + let before_offset = if starting_here { + start_position + .as_ref() + .map(|position| position.before_offset) + .unwrap_or(adopted_end) + } else { + adopted_end + }; + if before_offset > adopted_end { + return Err(RetainedHistoryReadError::InvalidCursor); + } + let mut reader = store + .open_retained_segment_reader(session_id, segment.segment_id, before_offset) + .map_err(map_history_store_error)?; + loop { + let remaining = limits.max_scan_bytes.saturating_sub(scanned_bytes); + let (record, bytes) = reader + .previous_record(remaining) + .map_err(map_history_store_error)?; + scanned_bytes = scanned_bytes + .checked_add(bytes) + .ok_or(RetainedHistoryReadError::ResourceLimit)?; + let Some(record) = record else { + break; + }; + scanned_entries = scanned_entries + .checked_add(persisted_entry_units(std::slice::from_ref(&record.entry))) + .ok_or(RetainedHistoryReadError::ResourceLimit)?; + if scanned_entries > limits.max_entries { + return Err(RetainedHistoryReadError::ResourceLimit); + } + if matches!(record.entry, LogEntry::AnnotatedSegmentStart { .. }) { + if record.start_offset != 0 { + return Err(RetainedHistoryReadError::CorruptLog); + } + continue; + } + let entries = project_history_record(session_id, segment.segment_id, &record.entry); + if collect_history_entries_backward( + &entries, + segment.segment_id, + record.start_offset, + Some(record.end_offset), + false, + turn_limit, + &mut seen_entries, + &mut pending_entries, + &mut turns, + ) { + return Ok(turns); + } + } + + if segment_index + 1 == lineage.len() + && collect_history_entries_backward( + &segment.seed_entries, + segment.segment_id, + 0, + None, + true, + turn_limit, + &mut seen_entries, + &mut pending_entries, + &mut turns, + ) + { + return Ok(turns); + } } - turns + Ok(turns) +} + +#[allow(clippy::too_many_arguments)] +fn collect_history_entries_backward( + entries: &[SessionSnapshotEntry], + segment_id: SegmentId, + before_offset: u64, + record_end_offset: Option, + seed: bool, + turn_limit: usize, + seen_entries: &mut HashSet, + pending_entries: &mut Vec, + turns: &mut Vec, +) -> bool { + for (index, entry) in entries.iter().enumerate().rev() { + if entry.provenance == SessionEntryProvenance::DerivedSummary + || !seen_entries.insert(entry.entry_id.clone()) + { + continue; + } + pending_entries.push(entry.clone()); + if is_user_entry(entry) { + let mut turn_entries = std::mem::take(pending_entries); + turn_entries.reverse(); + turns.push(PositionedHistoryTurn { + turn: SessionConversationTurn { + turn_id: entry.entry_id.clone(), + entries: turn_entries, + }, + position: HistoryTurnPosition { + segment_id, + before_offset, + record_end_offset, + seed_entry_index: seed.then_some(index), + }, + }); + if turns.len() >= turn_limit { + return true; + } + } + } + false +} + +fn project_history_record( + session_id: SessionId, + segment_id: SegmentId, + record: &LogEntry, +) -> Vec { + project_session_snapshot_for_segment(session_id, Some(segment_id), std::slice::from_ref(record)) + .entries + .into_iter() + .filter(|entry| entry.provenance != SessionEntryProvenance::DerivedSummary) + .collect() } fn is_user_entry(entry: &SessionSnapshotEntry) -> bool { @@ -553,6 +889,13 @@ fn map_history_worker_store_error(error: WorkerStoreError) -> RetainedHistoryRea } } +fn map_history_cursor_store_error(error: StoreError) -> RetainedHistoryReadError { + match error { + StoreError::ReadLimitExceeded => RetainedHistoryReadError::ResourceLimit, + _ => RetainedHistoryReadError::InvalidCursor, + } +} + fn map_history_store_error(error: StoreError) -> RetainedHistoryReadError { match error { StoreError::Io(error) if error.kind() == std::io::ErrorKind::NotFound => { @@ -1120,6 +1463,7 @@ mod tests { ); assert!(newest.has_more); assert!(newest.next_cursor.is_some()); + assert_eq!(newest.compact_ancestor_lineage_ids.len(), 2); assert_eq!( newest.turns[2] .entries @@ -1465,11 +1809,43 @@ mod tests { .unwrap_err(), RetainedHistoryReadError::InvalidCursor ); + let mut tampered = decode_history_cursor(&cursor).unwrap(); + tampered.before_turn_id = "user-1".to_string(); + let tampered = encode_history_cursor(&tampered).unwrap(); + assert_eq!( + read_retained_session_history_page( + &aggregate_a, + "worker-a", + Some(&tampered), + None, + RetainedHistoryReadLimits::default(), + ) + .unwrap_err(), + RetainedHistoryReadError::InvalidCursor + ); let aggregate = WorkerAggregateStore::new(&aggregate_a, "worker-a").unwrap(); let metadata = aggregate.read_by_name("worker-a").unwrap().unwrap(); let active = metadata.active.unwrap(); let old_segment = active.segment_id.unwrap(); + let mut appended = Vec::new(); + append_turn(&mut appended, 7); + let session = WorkerSessionStore::new(aggregate_a.join("session")).unwrap(); + for entry in appended { + session + .append(active.session_id, old_segment, &entry) + .unwrap(); + } + let older = read_retained_session_history_page( + &aggregate_a, + "worker-a", + Some(&cursor), + None, + RetainedHistoryReadLimits::default(), + ) + .unwrap(); + assert_eq!(turn_ids(&older), vec!["user-1"]); + let new_segment = crate::new_segment_id(); WorkerSessionStore::new(aggregate_a.join("session")) .unwrap() @@ -1509,6 +1885,347 @@ mod tests { ); } + #[test] + fn retained_history_reads_only_a_bounded_suffix_for_each_page() { + let worker_name = "worker-bounded-backward-pages"; + let session_id = crate::new_session_id(); + let segment_id = crate::new_segment_id(); + let mut log = vec![segment_start(session_id, Vec::new(), None, None)]; + for turn in 1..=200 { + append_turn(&mut log, turn); + } + let (_root, aggregate_root) = + persist_history_fixture(worker_name, session_id, segment_id, vec![(segment_id, log)]); + let limits = RetainedHistoryReadLimits { + max_scan_bytes: 64 * 1024, + ..RetainedHistoryReadLimits::default() + }; + assert!( + std::fs::metadata( + aggregate_root + .join("session/segments") + .join(format!("{segment_id}.jsonl")), + ) + .unwrap() + .len() + > limits.max_scan_bytes, + "the fixture must be larger than one page's complete scan budget", + ); + + let newest = + read_retained_session_history_page(&aggregate_root, worker_name, None, None, limits) + .unwrap(); + assert_eq!( + turn_ids(&newest), + vec!["user-196", "user-197", "user-198", "user-199", "user-200"] + ); + + let earlier = read_retained_session_history_page( + &aggregate_root, + worker_name, + newest.next_cursor.as_deref(), + None, + limits, + ) + .unwrap(); + assert_eq!( + turn_ids(&earlier), + vec!["user-191", "user-192", "user-193", "user-194", "user-195"] + ); + } + + #[test] + fn retained_history_does_not_scan_a_fork_sources_unadopted_suffix() { + let worker_name = "worker-bounded-fork-prefix"; + let session_id = crate::new_session_id(); + let source_segment = crate::new_segment_id(); + let active_segment = crate::new_segment_id(); + let mut source_log = vec![segment_start(session_id, Vec::new(), None, None)]; + for turn in 1..=200 { + append_turn(&mut source_log, turn); + } + let mut active_log = vec![segment_start( + session_id, + Vec::new(), + Some(SegmentOrigin { + segment_id: source_segment, + at_turn_index: 5, + }), + None, + )]; + append_turn(&mut active_log, 201); + let (_root, aggregate_root) = persist_history_fixture( + worker_name, + session_id, + active_segment, + vec![(source_segment, source_log), (active_segment, active_log)], + ); + let limits = RetainedHistoryReadLimits { + max_scan_bytes: 64 * 1024, + ..RetainedHistoryReadLimits::default() + }; + assert!( + std::fs::metadata( + aggregate_root + .join("session/segments") + .join(format!("{source_segment}.jsonl")), + ) + .unwrap() + .len() + > limits.max_scan_bytes, + ); + + let page = + read_retained_session_history_page(&aggregate_root, worker_name, None, None, limits) + .unwrap(); + assert_eq!( + turn_ids(&page), + vec!["user-2", "user-3", "user-4", "user-5", "user-201"] + ); + assert!(page.has_more); + } + + #[test] + fn retained_history_pages_within_one_root_seed_record() { + let worker_name = "worker-root-seed-pages"; + let session_id = crate::new_session_id(); + let segment_id = crate::new_segment_id(); + let history = (1..=12) + .flat_map(|turn| [user_message(turn), assistant_message(turn, "final")]) + .collect(); + let log = vec![ + segment_start(session_id, history, None, None), + LogEntry::InputSegmentsCheckpoint { + ts: 2, + user_segments: (1..=12) + .map(|turn| { + vec![Segment::Text { + content: format!("typed request {turn}"), + }] + }) + .collect(), + }, + ]; + let (_root, aggregate_root) = + persist_history_fixture(worker_name, session_id, segment_id, vec![(segment_id, log)]); + + let newest = read_retained_session_history_page( + &aggregate_root, + worker_name, + None, + None, + RetainedHistoryReadLimits::default(), + ) + .unwrap(); + assert_eq!( + turn_ids(&newest), + vec!["user-8", "user-9", "user-10", "user-11", "user-12"] + ); + assert_eq!( + newest.turns[0].entries[0].data, + SessionSnapshotEntryData::UserInput { + segments: vec![Segment::Text { + content: "typed request 8".to_string(), + }], + } + ); + let middle = read_retained_session_history_page( + &aggregate_root, + worker_name, + newest.next_cursor.as_deref(), + None, + RetainedHistoryReadLimits::default(), + ) + .unwrap(); + assert_eq!( + turn_ids(&middle), + vec!["user-3", "user-4", "user-5", "user-6", "user-7"] + ); + let oldest = read_retained_session_history_page( + &aggregate_root, + worker_name, + middle.next_cursor.as_deref(), + None, + RetainedHistoryReadLimits::default(), + ) + .unwrap(); + assert_eq!(turn_ids(&oldest), vec!["user-1", "user-2"]); + assert!(!oldest.has_more); + } + + #[test] + fn retained_history_keeps_each_nested_lineage_edge_boundary_local() { + let worker_name = "worker-nested-boundaries"; + let session_id = crate::new_session_id(); + let root_segment = crate::new_segment_id(); + let middle_segment = crate::new_segment_id(); + let active_segment = crate::new_segment_id(); + let mut root_log = vec![segment_start(session_id, Vec::new(), None, None)]; + for turn in 1..=3 { + append_turn(&mut root_log, turn); + } + let middle_log = vec![segment_start( + session_id, + vec![user_message(3), assistant_message(3, "final")], + Some(SegmentOrigin { + segment_id: root_segment, + at_turn_index: 3, + }), + None, + )]; + let mut active_log = vec![segment_start( + session_id, + Vec::new(), + Some(SegmentOrigin { + segment_id: middle_segment, + at_turn_index: 0, + }), + None, + )]; + append_turn(&mut active_log, 4); + let (_root, aggregate_root) = persist_history_fixture( + worker_name, + session_id, + active_segment, + vec![ + (root_segment, root_log), + (middle_segment, middle_log), + (active_segment, active_log), + ], + ); + + let page = read_retained_session_history_page( + &aggregate_root, + worker_name, + None, + None, + RetainedHistoryReadLimits::default(), + ) + .unwrap(); + assert_eq!( + turn_ids(&page), + vec!["user-1", "user-2", "user-3", "user-4"] + ); + } + + #[test] + fn retained_history_resolves_an_immediate_consecutive_compact_at_the_seed() { + let worker_name = "worker-consecutive-compact"; + let session_id = crate::new_session_id(); + let root_segment = crate::new_segment_id(); + let middle_segment = crate::new_segment_id(); + let active_segment = crate::new_segment_id(); + let mut root_log = vec![segment_start(session_id, Vec::new(), None, None)]; + for turn in 1..=5 { + append_turn(&mut root_log, turn); + } + let middle_log = vec![segment_start( + session_id, + Vec::new(), + None, + Some(SegmentOrigin { + segment_id: root_segment, + at_turn_index: 5, + }), + )]; + let active_log = vec![segment_start( + session_id, + Vec::new(), + None, + Some(SegmentOrigin { + segment_id: middle_segment, + at_turn_index: 5, + }), + )]; + let (_root, aggregate_root) = persist_history_fixture( + worker_name, + session_id, + active_segment, + vec![ + (root_segment, root_log), + (middle_segment, middle_log), + (active_segment, active_log), + ], + ); + + let page = read_retained_session_history_page( + &aggregate_root, + worker_name, + None, + None, + RetainedHistoryReadLimits::default(), + ) + .unwrap(); + assert_eq!( + turn_ids(&page), + vec!["user-1", "user-2", "user-3", "user-4", "user-5"] + ); + assert_eq!(page.compact_ancestor_lineage_ids.len(), 2); + } + + #[test] + fn retained_history_finds_a_late_adopted_boundary_from_the_tail() { + let worker_name = "worker-late-adopted-boundary"; + let session_id = crate::new_session_id(); + let source_segment = crate::new_segment_id(); + let active_segment = crate::new_segment_id(); + let mut source_log = vec![segment_start(session_id, Vec::new(), None, None)]; + append_turn(&mut source_log, 1); + source_log.insert( + 3, + LogEntry::AnnotatedAssistantItem { + ts: 12, + entry: history_message( + "assistant-1-large", + LoggedRole::Assistant, + "x".repeat(300 * 1024), + LoggedSessionHistoryOrigin::LegacyUnknown, + ), + }, + ); + for turn in 2..=101 { + append_turn(&mut source_log, turn); + } + let active_log = vec![segment_start( + session_id, + Vec::new(), + Some(SegmentOrigin { + segment_id: source_segment, + at_turn_index: 100, + }), + None, + )]; + let (_root, aggregate_root) = persist_history_fixture( + worker_name, + session_id, + active_segment, + vec![(source_segment, source_log), (active_segment, active_log)], + ); + let limits = RetainedHistoryReadLimits { + max_scan_bytes: 256 * 1024, + ..RetainedHistoryReadLimits::default() + }; + assert!( + std::fs::metadata( + aggregate_root + .join("session/segments") + .join(format!("{source_segment}.jsonl")), + ) + .unwrap() + .len() + > limits.max_scan_bytes, + ); + + let page = + read_retained_session_history_page(&aggregate_root, worker_name, None, None, limits) + .unwrap(); + assert_eq!( + turn_ids(&page), + vec!["user-96", "user-97", "user-98", "user-99", "user-100"] + ); + assert!(page.has_more); + } + #[test] fn retained_history_enforces_page_scan_and_response_limits_without_splitting_turns() { let worker_name = "worker-history-limits"; @@ -1574,7 +2291,7 @@ mod tests { let worker_name = "worker-oversized-turn"; let session_id = crate::new_session_id(); let segment_id = crate::new_segment_id(); - let body = "x".repeat(4096); + let body = "x".repeat(128 * 1024); let log = vec![ segment_start(session_id, Vec::new(), None, None), LogEntry::Invoke { diff --git a/crates/session-store/src/worker_session_store.rs b/crates/session-store/src/worker_session_store.rs index 9f46a5b2a..1f9af232d 100644 --- a/crates/session-store/src/worker_session_store.rs +++ b/crates/session-store/src/worker_session_store.rs @@ -17,7 +17,7 @@ use crate::{PasteArtifactLimits, SegmentId, SessionId}; use protocol::PasteArtifactRef; use serde::{Deserialize, Serialize}; use std::fs::{self, File, OpenOptions}; -use std::io::{Read, Seek, SeekFrom, Write}; +use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write}; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use std::time::SystemTime; @@ -42,6 +42,106 @@ struct SessionManifest { session_id: SessionId, } +const RETAINED_BACKWARD_READ_CHUNK_BYTES: usize = 64 * 1024; + +#[derive(Debug)] +pub(crate) struct RetainedLogRecord { + pub start_offset: u64, + pub end_offset: u64, + pub entry: LogEntry, +} + +pub(crate) struct RetainedSegmentReader { + file: File, + read_end: u64, + buffer_start: u64, + buffer: Vec, +} + +impl RetainedSegmentReader { + pub fn previous_record( + &mut self, + max_additional_bytes: u64, + ) -> Result<(Option, u64), StoreError> { + let mut read_bytes = 0_u64; + loop { + if !self.buffer.is_empty() { + if self.buffer.last() != Some(&b'\n') { + return Err(StoreError::Corrupt { + line: 0, + message: "retained Segment cursor is not at a record boundary".to_string(), + }); + } + let prior_newline = self.buffer[..self.buffer.len() - 1] + .iter() + .rposition(|byte| *byte == b'\n'); + if let Some(newline) = + prior_newline.or((self.buffer_start == 0).then_some(usize::MAX)) + { + let record_start_in_buffer = if newline == usize::MAX { + 0 + } else { + newline + 1 + }; + let record_start = self.buffer_start + record_start_in_buffer as u64; + let record_end = self.buffer_start + self.buffer.len() as u64; + let record = + self.buffer[record_start_in_buffer..self.buffer.len() - 1].to_vec(); + self.buffer.truncate(record_start_in_buffer); + self.read_end = record_start; + if record.is_empty() { + return Err(StoreError::Corrupt { + line: 0, + message: "empty retained Segment log record".to_string(), + }); + } + let entry = serde_json::from_slice(&record)?; + return Ok(( + Some(RetainedLogRecord { + start_offset: record_start, + end_offset: record_end, + entry, + }), + read_bytes, + )); + } + } else if self.read_end == 0 { + return Ok((None, read_bytes)); + } + + if self.buffer_start == 0 && !self.buffer.is_empty() { + return Err(StoreError::Corrupt { + line: 0, + message: "retained Segment log does not end at a record boundary".to_string(), + }); + } + if read_bytes >= max_additional_bytes { + return Err(StoreError::ReadLimitExceeded); + } + let end = if self.buffer.is_empty() { + self.read_end + } else { + self.buffer_start + }; + let available = max_additional_bytes - read_bytes; + let chunk_len = end + .min(RETAINED_BACKWARD_READ_CHUNK_BYTES as u64) + .min(available); + if chunk_len == 0 { + return Err(StoreError::ReadLimitExceeded); + } + let start = end - chunk_len; + self.file.seek(SeekFrom::Start(start))?; + let mut chunk = vec![0_u8; chunk_len as usize]; + self.file.read_exact(&mut chunk)?; + read_bytes += chunk_len; + chunk.extend_from_slice(&self.buffer); + self.buffer = chunk; + self.buffer_start = start; + } + } +} + impl WorkerSessionStore { /// Open the Session store rooted at `/session`. pub fn new(root: impl Into) -> Result { @@ -150,6 +250,198 @@ impl WorkerSessionStore { Ok((parse_jsonl(&bytes)?, byte_len)) } + pub(crate) fn read_first_log_record_read_only_bounded( + &self, + session_id: SessionId, + segment_id: SegmentId, + max_bytes: u64, + ) -> Result<(Option, u64, u64), StoreError> { + self.validate_retained_session(session_id)?; + let file = open_retained_file(&self.log_path(segment_id)).map_err(|error| { + if error.kind() == std::io::ErrorKind::NotFound { + StoreError::NotFound(segment_id) + } else { + StoreError::Io(error) + } + })?; + let file_len = file.metadata()?.len(); + if file_len == 0 { + return Ok((None, 0, 0)); + } + let mut reader = BufReader::new(file.take(max_bytes.saturating_add(1))); + let mut record = Vec::new(); + let read_bytes = reader.read_until(b'\n', &mut record)? as u64; + if read_bytes > max_bytes || record.last() != Some(&b'\n') { + return Err(StoreError::ReadLimitExceeded); + } + record.pop(); + if record.is_empty() { + return Err(StoreError::Corrupt { + line: 0, + message: "empty retained Segment log record".to_string(), + }); + } + let entry = serde_json::from_slice(&record)?; + Ok(( + Some(RetainedLogRecord { + start_offset: 0, + end_offset: read_bytes, + entry, + }), + read_bytes, + file_len, + )) + } + + pub(crate) fn read_next_log_record_read_only_bounded( + &self, + session_id: SessionId, + segment_id: SegmentId, + start_offset: u64, + max_bytes: u64, + ) -> Result<(Option, u64), StoreError> { + self.validate_retained_session(session_id)?; + let mut file = open_retained_file(&self.log_path(segment_id)).map_err(|error| { + if error.kind() == std::io::ErrorKind::NotFound { + StoreError::NotFound(segment_id) + } else { + StoreError::Io(error) + } + })?; + let file_len = file.metadata()?.len(); + if start_offset >= file_len { + return Ok((None, 0)); + } + file.seek(SeekFrom::Start(start_offset))?; + let mut reader = BufReader::new(file.take(max_bytes.saturating_add(1))); + let mut record = Vec::new(); + let read_bytes = reader.read_until(b'\n', &mut record)? as u64; + if read_bytes > max_bytes || record.last() != Some(&b'\n') { + return Err(StoreError::ReadLimitExceeded); + } + record.pop(); + if record.is_empty() { + return Err(StoreError::Corrupt { + line: 0, + message: "empty retained Segment log record".to_string(), + }); + } + let entry = serde_json::from_slice(&record)?; + Ok(( + Some(RetainedLogRecord { + start_offset, + end_offset: start_offset + read_bytes, + entry, + }), + read_bytes, + )) + } + + pub(crate) fn read_log_record_range_read_only_bounded( + &self, + session_id: SessionId, + segment_id: SegmentId, + start_offset: u64, + end_offset: u64, + max_bytes: u64, + ) -> Result { + self.validate_retained_session(session_id)?; + let record_len = + end_offset + .checked_sub(start_offset) + .ok_or_else(|| StoreError::Corrupt { + line: 0, + message: "retained Segment record range is reversed".to_string(), + })?; + if record_len == 0 || record_len > max_bytes { + return Err(StoreError::ReadLimitExceeded); + } + let mut file = open_retained_file(&self.log_path(segment_id)).map_err(|error| { + if error.kind() == std::io::ErrorKind::NotFound { + StoreError::NotFound(segment_id) + } else { + StoreError::Io(error) + } + })?; + if end_offset > file.metadata()?.len() { + return Err(StoreError::Corrupt { + line: 0, + message: "retained Segment record range exceeds the committed log".to_string(), + }); + } + file.seek(SeekFrom::Start(start_offset))?; + let mut record = vec![0_u8; record_len as usize]; + file.read_exact(&mut record)?; + if record.last() != Some(&b'\n') || record[..record.len() - 1].contains(&b'\n') { + return Err(StoreError::Corrupt { + line: 0, + message: "retained Segment cursor does not identify one record".to_string(), + }); + } + record.pop(); + let entry = serde_json::from_slice(&record)?; + Ok(RetainedLogRecord { + start_offset, + end_offset, + entry, + }) + } + + pub(crate) fn open_retained_segment_reader( + &self, + session_id: SessionId, + segment_id: SegmentId, + before_offset: u64, + ) -> Result { + self.validate_retained_session(session_id)?; + let mut file = open_retained_file(&self.log_path(segment_id)).map_err(|error| { + if error.kind() == std::io::ErrorKind::NotFound { + StoreError::NotFound(segment_id) + } else { + StoreError::Io(error) + } + })?; + let file_len = file.metadata()?.len(); + if before_offset > file_len { + return Err(StoreError::Corrupt { + line: 0, + message: "retained Segment cursor is beyond the committed log".to_string(), + }); + } + if before_offset > 0 { + file.seek(SeekFrom::Start(before_offset - 1))?; + let mut boundary = [0_u8; 1]; + file.read_exact(&mut boundary)?; + if boundary[0] != b'\n' { + return Err(StoreError::Corrupt { + line: 0, + message: "retained Segment cursor is not at a record boundary".to_string(), + }); + } + } + Ok(RetainedSegmentReader { + file, + read_end: before_offset, + buffer_start: before_offset, + buffer: Vec::new(), + }) + } + + fn validate_retained_session(&self, requested: SessionId) -> Result<(), StoreError> { + let retained_session_id = self.session_id.lock().map_err(|_| StoreError::Corrupt { + line: 0, + message: "Worker Session identity lock poisoned".to_string(), + })?; + if *retained_session_id != Some(requested) { + return Err(StoreError::Corrupt { + line: 0, + message: "active Worker Session identity does not match retained manifest" + .to_string(), + }); + } + Ok(()) + } + pub fn segment_log_len(&self, segment_id: SegmentId) -> Result { Ok(fs::metadata(self.log_path(segment_id))?.len()) } @@ -430,6 +722,25 @@ impl Store for WorkerSessionStore { } } +fn open_retained_file(path: &Path) -> std::io::Result { + #[cfg(any(target_os = "linux", target_os = "android"))] + { + use std::os::unix::fs::OpenOptionsExt; + OpenOptions::new() + .read(true) + .custom_flags(libc::O_NOATIME) + .open(path) + } + #[cfg(not(any(target_os = "linux", target_os = "android")))] + { + let _ = path; + Err(std::io::Error::new( + std::io::ErrorKind::Unsupported, + "read-only retained observation requires no-atime file reads", + )) + } +} + fn segment_log_paths(root: &Path) -> Result, StoreError> { let segments = root.join(SEGMENTS_DIR); if !segments.exists() { diff --git a/tools/web-ux/browser-tests/console_history_fixture_server.ts b/tools/web-ux/browser-tests/console_history_fixture_server.ts index 16a78d810..53035f71b 100644 --- a/tools/web-ux/browser-tests/console_history_fixture_server.ts +++ b/tools/web-ux/browser-tests/console_history_fixture_server.ts @@ -64,6 +64,7 @@ function historyPage(indices: number[], cursor: string | null) { page: { session_id: "session-a", lineage_id: "lineage-a", + compact_ancestor_lineage_ids: [], turns: indices.map((index) => ({ turn_id: `user-${index}`, entries: [historyEntry(index, "user"), historyEntry(index, "assistant")], @@ -162,7 +163,7 @@ Deno.serve({ hostname: "127.0.0.1", port }, async (request) => { ) { return json({ availability: "retained_snapshot", - identity: { session_id: "session-a", segment_id: "segment-a", entry_count: 0 }, + identity: { session_id: "session-a", segment_id: "segment-a", entry_count: 12 }, snapshot: { pending_submissions: { revision: 0, @@ -170,7 +171,10 @@ Deno.serve({ hostname: "127.0.0.1", port }, async (request) => { head_id: null, submissions: [], }, - entries: [], + entries: [7, 8, 9, 10, 11, 12].flatMap((index) => [ + historyEntry(index, "user"), + historyEntry(index, "assistant"), + ]), }, }); } diff --git a/tools/web-ux/browser-tests/console_history_shell_test.ts b/tools/web-ux/browser-tests/console_history_shell_test.ts index 66bf15fee..d2bcb45a2 100644 --- a/tools/web-ux/browser-tests/console_history_shell_test.ts +++ b/tools/web-ux/browser-tests/console_history_shell_test.ts @@ -98,10 +98,20 @@ Deno.test("production Console pages backward by real turns while preserving both ); } const navigation = page.getByRole("complementary", { name: "Conversation turns" }); - await navigation.getByText("question 8", { exact: true }).waitFor(); + await navigation.getByText("question 7", { exact: true }).waitFor(); assertEquals(await historyRequestCount(baseUrl), 1); - assertEquals(await transcript.getByText(/^question (8|9|10|11|12)$/).count(), 5); - assertEquals(await navigation.getByText(/^question (8|9|10|11|12)$/).count(), 5); + assertEquals(await transcript.getByText(/^question (7|8|9|10|11|12)$/).count(), 6); + assertEquals(await navigation.getByText(/^question (7|8|9|10|11|12)$/).count(), 6); + const currentQuestionTop = await transcript + .getByText("question 7", { exact: true }) + .evaluate((element) => element.getBoundingClientRect().top); + const retainedQuestionTop = await transcript + .getByText("question 8", { exact: true }) + .evaluate((element) => element.getBoundingClientRect().top); + assert( + currentQuestionTop < retainedQuestionTop, + "the unmatched current-snapshot turn must precede overlapping retained history", + ); const question12 = navigation.getByRole("button", { name: "Jump to conversation: question 12", diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index c8180688d..2fc6f64a3 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -155,7 +155,14 @@ export type SessionHistoryPage = { session_id: string, /** * Stable identity of the adopted active lineage used to fence stale pages. */ -lineage_id: string, turns: Array, next_cursor?: string | null, has_more: boolean, }; +lineage_id: string, +/** + * Older lineage identities that the active lineage fully adopted through + * one or more consecutive Compact rotations. Clients may retain already + * loaded turns when their prior lineage appears here, while replacing the + * cursor with this page's boundary. + */ +compact_ancestor_lineage_ids?: Array, turns: Array, next_cursor?: string | null, has_more: boolean, }; export type PendingSubmissionSummary = { submission_id: string, accepted_at_ms: number, segment_count: number, byte_len: number, }; diff --git a/web/workspace/src/lib/workspace/console/history.test.ts b/web/workspace/src/lib/workspace/console/history.test.ts index fb7674d59..47c302a95 100644 --- a/web/workspace/src/lib/workspace/console/history.test.ts +++ b/web/workspace/src/lib/workspace/console/history.test.ts @@ -8,6 +8,7 @@ import { beginConsoleHistoryRequest, type ConsoleHistoryState, conversationTurnPreviews, + conversationTurnPreviewsFromLines, emptyConsoleHistoryState, historyEntries, setConsoleHistoryTopEdge, @@ -53,10 +54,13 @@ function turn(index: number): SessionConversationTurn { function page( turns: SessionConversationTurn[], cursor: string | null, + lineageId = "lineage-a", + compactAncestorLineageIds: string[] = [], ): SessionHistoryPage { return { session_id: "session-a", - lineage_id: "lineage-a", + lineage_id: lineageId, + compact_ancestor_lineage_ids: compactAncestorLineageIds, turns, next_cursor: cursor, has_more: cursor !== null, @@ -144,6 +148,151 @@ Deno.test("refresh replaces overlapping latest turns while retaining loaded earl ); }); +Deno.test("same-lineage refresh without overlap replaces the paging boundary", () => { + let state: ConsoleHistoryState = { + ...emptyConsoleHistoryState(), + sessionId: "session-a", + lineageId: "lineage-a", + turns: [1, 2, 3, 4].map(turn), + cursor: null, + hasMore: false, + status: "loading", + requestedCursor: "__initial__", + }; + state = applyConsoleHistoryPage( + state, + page([6, 7, 8, 9, 10].map(turn), "cursor-6", "lineage-a"), + null, + ); + + assert( + state.turns.map((value) => value.turn_id).join(",") === + "u-1,u-2,u-3,u-4,u-6,u-7,u-8,u-9,u-10", + "a reconnect may preserve loaded rows while still exposing the missing gap", + ); + assert( + state.cursor === "cursor-6" && state.hasMore, + "a non-overlapping same-lineage page must replace an exhausted old boundary", + ); +}); + +Deno.test("changed lineage keeps the adopted prefix and discards the replaced branch", () => { + let state = emptyConsoleHistoryState(); + state = beginConsoleHistoryRequest(state, null)!; + state = applyConsoleHistoryPage( + state, + page([6, 7, 8, 9, 10].map(turn), "cursor-6"), + null, + ); + state = beginConsoleHistoryRequest(state, "cursor-6")!; + state = applyConsoleHistoryPage( + state, + page([1, 2, 3, 4, 5].map(turn), null), + "cursor-6", + ); + + state = beginConsoleHistoryRequest(state, null)!; + state = applyConsoleHistoryPage( + state, + page([3, 4, 5, 11, 12].map(turn), "cursor-3-new", "lineage-b"), + null, + ); + + assert( + state.turns.map((value) => value.turn_id).join(",") === + "u-1,u-2,u-3,u-4,u-5,u-11,u-12", + "rewind must preserve only the prefix adopted by the new lineage", + ); + assert( + state.cursor === "cursor-3-new" && state.hasMore, + "a changed lineage must replace the stale paging cursor", + ); +}); + +Deno.test("Compact lineage refresh preserves already loaded adopted history", () => { + let state: ConsoleHistoryState = { + ...emptyConsoleHistoryState(), + sessionId: "session-a", + lineageId: "lineage-a", + turns: Array.from({ length: 10 }, (_, index) => turn(index + 1)), + status: "loading", + requestedCursor: "__initial__", + }; + state = applyConsoleHistoryPage( + state, + page( + [11, 12, 13, 14, 15].map(turn), + "cursor-11-new", + "lineage-compact", + ["lineage-a"], + ), + null, + ); + + assert( + state.turns.map((value) => value.turn_id).join(",") === + Array.from({ length: 15 }, (_, index) => `u-${index + 1}`).join(","), + "Compact must keep the adopted prefix without requiring newest-page overlap", + ); + assert( + state.cursor === "cursor-11-new" && state.hasMore, + "Compact must replace the old-lineage cursor even when adopted turns are preserved", + ); +}); + +Deno.test("changed lineage without a stable overlap replaces retained turns", () => { + let state: ConsoleHistoryState = { + ...emptyConsoleHistoryState(), + sessionId: "session-a", + lineageId: "lineage-a", + turns: [1, 2, 3].map(turn), + cursor: "old-cursor", + hasMore: true, + status: "loading", + requestedCursor: "__initial__", + }; + state = applyConsoleHistoryPage( + state, + page([11, 12].map(turn), null, "lineage-b"), + null, + ); + assert( + state.turns.map((value) => value.turn_id).join(",") === "u-11,u-12", + "unrelated lineage content must not survive refresh", + ); + assert( + state.cursor === null && !state.hasMore, + "new lineage boundary must be authoritative", + ); +}); + +Deno.test("line previews include a current turn outside the retained five-turn page", () => { + const lines = [1, 2, 3, 4, 5, 6].flatMap((index) => [ + { + id: `user-${index}`, + entryId: `u-${index}`, + kind: "user" as const, + title: "User", + body: `question ${index}`, + source: "event" as const, + }, + { + id: `assistant-${index}`, + entryId: `a-${index}`, + kind: "assistant" as const, + title: "Assistant", + body: `done ${index}`, + source: "event" as const, + }, + ]); + assert( + conversationTurnPreviewsFromLines(lines).map((preview) => preview.turnId) + .join(",") === + "u-1,u-2,u-3,u-4,u-5,u-6", + "turn navigation must include unmatched current snapshot turns in transcript order", + ); +}); + Deno.test("turn preview uses one user line and final non-empty assistant up to three lines", () => { const preview = conversationTurnPreviews([turn(4)])[0]; assert(preview.user === "question 4", "user preview must use its first line"); diff --git a/web/workspace/src/lib/workspace/console/history.ts b/web/workspace/src/lib/workspace/console/history.ts index def1a88b5..ac994132b 100644 --- a/web/workspace/src/lib/workspace/console/history.ts +++ b/web/workspace/src/lib/workspace/console/history.ts @@ -1,3 +1,4 @@ +import type { ConsoleLine } from "./model.ts"; import type { SessionConversationTurn, SessionHistoryPage, @@ -83,9 +84,19 @@ export function applyConsoleHistoryPage( const refreshingSameSession = initial && state.sessionId === page.session_id && state.turns.length > 0; + const refreshingSameLineage = refreshingSameSession && + state.lineageId === page.lineage_id; + const refreshingCompactDescendant = refreshingSameSession && + state.lineageId !== null && + (page.compact_ancestor_lineage_ids ?? []).includes(state.lineageId); + const refreshBridgesRetained = refreshingSameLineage && incoming.some( + (turn) => state.turns.some((retained) => retained.turn_id === turn.turn_id), + ); const turns = initial - ? refreshingSameSession + ? refreshingSameLineage || refreshingCompactDescendant ? mergeRefreshedTurns(state.turns, incoming) + : refreshingSameSession + ? reconcileChangedLineage(state.turns, incoming) : dedupeTurns(incoming) : dedupeTurns([...incoming, ...state.turns]); return { @@ -93,8 +104,8 @@ export function applyConsoleHistoryPage( sessionId: page.session_id, lineageId: page.lineage_id, turns, - cursor: refreshingSameSession ? state.cursor : (page.next_cursor ?? null), - hasMore: refreshingSameSession ? state.hasMore : page.has_more, + cursor: refreshBridgesRetained ? state.cursor : (page.next_cursor ?? null), + hasMore: refreshBridgesRetained ? state.hasMore : page.has_more, status: "ready", error: null, requestedCursor: null, @@ -144,20 +155,86 @@ export function conversationTurnPreviews( .filter(isAssistantEntry) .map(entryText) .filter((text) => text.trim().length > 0); - const assistant = (assistants.at(-1) ?? "") - .split(/\r?\n/) - .map((line) => line.trim()) - .filter(Boolean) - .slice(0, 3); return { turnId: turn.turn_id, lineId: userEntry?.entry_id ?? turn.turn_id, user: firstNonEmptyLine(userEntry ? entryText(userEntry) : ""), - assistant, + assistant: previewAssistant(assistants.at(-1) ?? ""), }; }); } +export function conversationTurnPreviewsFromLines( + lines: readonly ConsoleLine[], +): ConsoleTurnPreview[] { + const previews: ConsoleTurnPreview[] = []; + let current: { + turnId: string; + lineId: string; + user: string; + assistants: string[]; + } | null = null; + for (const line of lines) { + if (line.kind === "user") { + if (current) { + previews.push({ + turnId: current.turnId, + lineId: current.lineId, + user: current.user, + assistant: previewAssistant(current.assistants.at(-1) ?? ""), + }); + } + const stableId = line.entryId ?? line.id; + current = { + turnId: stableId, + lineId: stableId, + user: firstNonEmptyLine(line.body), + assistants: [], + }; + } else if ( + line.kind === "assistant" && current && line.body.trim().length > 0 + ) { + current.assistants.push(line.body); + } + } + if (current) { + previews.push({ + turnId: current.turnId, + lineId: current.lineId, + user: current.user, + assistant: previewAssistant(current.assistants.at(-1) ?? ""), + }); + } + return previews; +} + +function previewAssistant(value: string): string[] { + return value + .split(/\r?\n/) + .map((line) => line.trim()) + .filter(Boolean) + .slice(0, 3); +} + +function reconcileChangedLineage( + retained: readonly SessionConversationTurn[], + refreshed: readonly SessionConversationTurn[], +): SessionConversationTurn[] { + const retainedPositions = new Map( + retained.map((turn, index) => [turn.turn_id, index] as const), + ); + const firstShared = refreshed.find((turn) => + retainedPositions.has(turn.turn_id) + ); + if (!firstShared) return dedupeTurns(refreshed); + + const retainedPrefix = retained.slice( + 0, + retainedPositions.get(firstShared.turn_id) ?? 0, + ); + return dedupeTurns([...retainedPrefix, ...refreshed]); +} + function mergeRefreshedTurns( retained: readonly SessionConversationTurn[], refreshed: readonly SessionConversationTurn[], diff --git a/web/workspace/src/lib/workspace/console/model.test.ts b/web/workspace/src/lib/workspace/console/model.test.ts index e271ab5a0..59c02bdab 100644 --- a/web/workspace/src/lib/workspace/console/model.test.ts +++ b/web/workspace/src/lib/workspace/console/model.test.ts @@ -9,6 +9,7 @@ import { consoleWorkerViews, createConsoleProjector, isConsoleProjectionEvent, + mergeCommittedHistoryLines, projectConsole, projectConsoleLines, projectOverviewLines, @@ -3080,3 +3081,48 @@ Deno.test("committed tool identities reconcile one live call block", () => { assertEquals(lines[0].entryId, "tool-result-entry"); assertEquals(lines[0].toolCall?.state, "done"); }); + +Deno.test("committed history preserves unmatched current rows in stable entry order", () => { + const history = [2, 3, 4, 5, 6].flatMap((turn) => { + const user = consoleLine(`history-user-${turn}`, "user"); + user.entryId = `user-${turn}`; + const assistant = consoleLine(`history-assistant-${turn}`, "assistant"); + assistant.entryId = `assistant-${turn}`; + return [user, assistant]; + }); + const current = [1, 2, 3, 4, 5, 6].flatMap((turn) => { + const user = consoleLine(`current-user-${turn}`, "user"); + user.entryId = `user-${turn}`; + const assistant = consoleLine(`current-assistant-${turn}`, "assistant"); + assistant.entryId = `assistant-${turn}`; + return [user, assistant]; + }); + + const merged = mergeCommittedHistoryLines(history, current); + assertEquals( + merged.map((line) => line.entryId), + [1, 2, 3, 4, 5, 6].flatMap((turn) => [ + `user-${turn}`, + `assistant-${turn}`, + ]), + ); + assert( + merged.every((line) => line.id.startsWith("current-")), + "the ordered current snapshot must remain authoritative for overlapping entries", + ); +}); + +Deno.test("committed history inserts current live rows after their last stable anchor", () => { + const historyUser = consoleLine("history-user", "user"); + historyUser.entryId = "user-1"; + const currentUser = consoleLine("current-user", "user"); + currentUser.entryId = "user-1"; + const live = consoleLine("live-assistant", "in_flight"); + + assertEquals( + mergeCommittedHistoryLines([historyUser], [currentUser, live]).map((line) => + line.id + ), + ["current-user", "live-assistant"], + ); +}); diff --git a/web/workspace/src/lib/workspace/console/model.ts b/web/workspace/src/lib/workspace/console/model.ts index 2a047944d..973a00736 100644 --- a/web/workspace/src/lib/workspace/console/model.ts +++ b/web/workspace/src/lib/workspace/console/model.ts @@ -286,14 +286,16 @@ export function mergeCommittedHistoryLines( history: ConsoleLine[], current: ConsoleLine[], ): ConsoleLine[] { - const committed = new Set( - history.map((line) => line.entryId).filter((id): id is string => + const currentEntryIds = new Set( + current.map((line) => line.entryId).filter((id): id is string => Boolean(id) ), ); return [ - ...history, - ...current.filter((line) => !line.entryId || !committed.has(line.entryId)), + ...history.filter((line) => + !line.entryId || !currentEntryIds.has(line.entryId) + ), + ...current, ]; } diff --git a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte index fb9c02c1f..1d766437e 100644 --- a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte +++ b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte @@ -75,6 +75,7 @@ import { applyConsoleHistoryPage, beginConsoleHistoryRequest, + conversationTurnPreviewsFromLines, emptyConsoleHistoryState, failConsoleHistoryRequest, historyEntries, @@ -82,7 +83,7 @@ shouldLoadHistoryAtTop, type ConsoleHistoryState, } from "$lib/workspace/console/history"; - import { buildTurnNavigationItems, type TurnNavigationItem } from "$lib/workspace/console/turn-navigation"; + import type { TurnNavigationItem } from "$lib/workspace/console/turn-navigation"; import type { Diagnostic, Worker } from "$lib/workspace/sidebar/types"; type Props = { @@ -290,7 +291,7 @@ mergeCommittedHistoryLines(committedHistoryLines, currentLines), ); const turnNavigationItems = $derived( - buildTurnNavigationItems(selectedHistory.turns), + conversationTurnPreviewsFromLines(lines), ); const tasks = $derived(selectedConsoleProjection.tasks); const timelineLayout = $derived( @@ -396,6 +397,7 @@ } const payload = (await response.json()) as WorkerSessionHistoryResponse; if (token !== reloadToken || mainHistoryKey(target) !== key) return; + if (pendingHistoryRefreshes.has(key)) return; if (payload.availability === "unavailable") { setHistoryState( key, @@ -410,7 +412,12 @@ await tick(); restoreHistoryAnchors(anchors); } catch (error) { - if (token !== reloadToken || mainHistoryKey(target) !== key) return; + if ( + token !== reloadToken || + mainHistoryKey(target) !== key || + pendingHistoryRefreshes.has(key) + ) + return; setHistoryState( key, failConsoleHistoryRequest( @@ -425,6 +432,13 @@ token === reloadToken && mainHistoryKey(target) === key ) { + const stale = historyStateFor(key); + setHistoryState(key, { + ...stale, + status: stale.turns.length > 0 ? "ready" : "idle", + error: null, + requestedCursor: null, + }); void requestHistoryPage(target, key, null, token); } } @@ -509,6 +523,22 @@ ); } + function fenceCurrentHistoryForRewind() { + const target = consoleTarget; + if (!target) return; + const key = mainHistoryKey(target); + const current = historyStateFor(key); + const loading = current.status === "loading"; + const reset = emptyConsoleHistoryState(); + setHistoryState(key, { + ...reset, + status: loading ? "loading" : "idle", + requestedCursor: loading ? current.requestedCursor : null, + topEdgeArmed: current.topEdgeArmed, + }); + if (loading) pendingHistoryRefreshes.add(key); + } + function refreshCurrentHistory() { const target = consoleTarget; if (!target) return; @@ -533,7 +563,9 @@ } function jumpToConversationTurn(item: TurnNavigationItem) { - const line = lines.find((candidate) => candidate.entryId === item.lineId); + const line = lines.find( + (candidate) => candidate.entryId === item.lineId || candidate.id === item.lineId, + ); const target = line ? consoleBodyElement?.querySelector( `[data-console-line-id="${cssEscape(line.id)}"]`, @@ -694,6 +726,22 @@ ? "shutdown" : workerStateFromSnapshot(consoleProjection.workerState); advanceEventObservedAtVersion(); + if ( + eventBatch.some((event) => event.event.event === "rewind_applied") + ) { + fenceCurrentHistoryForRewind(); + } + if ( + eventBatch.some((event) => + event.event.event === "snapshot" || + event.event.event === "segment_rotated" || + event.event.event === "rewind_applied" || + event.event.event === "user_message" || + event.event.event === "session_entry_committed" + ) + ) { + refreshCurrentHistory(); + } if ( initialSnapshotApplication && eventBatch.some( @@ -724,13 +772,6 @@ } else if (payload.event === "pending_submissions_changed") { pendingSubmissions = payload.data.pending; } - if ( - payload.event === "segment_rotated" || - payload.event === "user_message" || - payload.event === "session_entry_committed" - ) { - refreshCurrentHistory(); - } if (payload.event === "error") { queueObservationDiagnostic({ code: payload.data.code, @@ -1335,9 +1376,6 @@ ]; return; } - untrack(() => { - void requestHistoryPage(target, mainHistoryKey(target), null, token); - }); if ( action.kind === "subscribe_live" && targetWorker && diff --git a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/console-display.browser.spec.ts b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/console-display.browser.spec.ts index cfc3868df..c826379b0 100644 --- a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/console-display.browser.spec.ts +++ b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/console-display.browser.spec.ts @@ -6,6 +6,7 @@ import { render, screen, waitFor, + within, } from "@testing-library/svelte"; import { afterEach, beforeEach, expect, test, vi } from "vitest"; import type { @@ -112,6 +113,7 @@ function emptyHistoryPage() { page: { session_id: "session-a", lineage_id: "lineage-a", + compact_ancestor_lineage_ids: [], turns: [], next_cursor: null, has_more: false, @@ -125,6 +127,7 @@ function historyPage(indices: number[], cursor: string | null) { page: { session_id: "session-a", lineage_id: "lineage-a", + compact_ancestor_lineage_ids: [], turns: indices.map((index) => ({ turn_id: `user-${index}`, entries: [ @@ -296,7 +299,10 @@ test("applies retained empty and nonempty snapshots before showing their read-on snapshot: sessionWithUserMessage("retained message"), })); - expect(await screen.findByText("retained message")).not.toBeNull(); + expect( + await within(screen.getByRole("article", { name: "main transcript" })) + .findByText("retained message"), + ).not.toBeNull(); expect(screen.getByText("Read-only retained conversation")).not.toBeNull(); expect(screen.queryByText("No conversation to display")).toBeNull(); }); @@ -364,17 +370,107 @@ test("preserves rendered content while reconnecting", async () => { latestListener().onFrame( subscribedFrame(sessionWithUserMessage("kept message")), ); - expect(await screen.findByText("kept message")).not.toBeNull(); + expect( + await within(screen.getByRole("article", { name: "main transcript" })) + .findByText("kept message"), + ).not.toBeNull(); latestListener().onStatus?.("closed", "Workspace subscription disconnected"); expect(await screen.findByText("Conversation updates are unavailable")).not .toBeNull(); - expect(screen.getByText("kept message")).not.toBeNull(); + expect( + within(screen.getByRole("article", { name: "main transcript" })).getByText( + "kept message", + ), + ).not.toBeNull(); expect(screen.queryByText("No conversation to display")).toBeNull(); latestListener().onStatus?.("connecting", "resubscribing"); expect(await screen.findByText("Refreshing conversation")).not.toBeNull(); - expect(screen.getByText("kept message")).not.toBeNull(); + expect( + within(screen.getByRole("article", { name: "main transcript" })).getByText( + "kept message", + ), + ).not.toBeNull(); +}); + +test("reconnect snapshot refreshes lineage and removes stale sibling history", async () => { + let historyRequests = 0; + vi.stubGlobal( + "fetch", + vi.fn((input: RequestInfo | URL) => { + if (!String(input).includes("/session/history")) { + return Promise.resolve( + Response.json({ availability: "live_protocol" }), + ); + } + historyRequests += 1; + const response = historyPage([historyRequests], null); + if (historyRequests === 2) { + response.page.lineage_id = "lineage-b"; + } + return Promise.resolve(Response.json(response)); + }), + ); + + render(ConsolePage, { data: pageData() }); + await waitFor(() => expect(multiplexer.subscribe).toHaveBeenCalledOnce()); + latestListener().onFrame( + subscribedFrame(sessionWithUserMessage("lineage A current")), + ); + await waitFor(() => + expect(screen.getAllByText("question 1")).toHaveLength(2) + ); + + latestListener().onStatus?.("closed", "disconnected during rewind"); + latestListener().onStatus?.("connecting", "reconnecting"); + latestListener().onFrame( + subscribedFrame(sessionWithUserMessage("lineage B current")), + ); + + await waitFor(() => + expect(screen.getAllByText("question 2")).toHaveLength(2) + ); + expect(screen.queryByText("question 1")).toBeNull(); + expect(historyRequests).toBe(2); +}); + +test("snapshot supersedes an in-flight page before stale lineage can render", async () => { + const staleHistory = deferred(); + let historyRequests = 0; + vi.stubGlobal( + "fetch", + vi.fn((input: RequestInfo | URL) => { + if (!String(input).includes("/session/history")) { + return Promise.resolve( + Response.json({ availability: "live_protocol" }), + ); + } + historyRequests += 1; + if (historyRequests === 1) return staleHistory.promise; + const response = historyPage([2], null); + response.page.lineage_id = "lineage-b"; + return Promise.resolve(Response.json(response)); + }), + ); + + render(ConsolePage, { data: pageData() }); + await waitFor(() => expect(multiplexer.subscribe).toHaveBeenCalledOnce()); + latestListener().onFrame( + subscribedFrame(sessionWithUserMessage("lineage A current")), + ); + await waitFor(() => expect(historyRequests).toBe(1)); + + latestListener().onFrame( + subscribedFrame(sessionWithUserMessage("lineage B current")), + ); + staleHistory.resolve(Response.json(historyPage([1], null))); + + await waitFor(() => + expect(screen.getAllByText("question 2")).toHaveLength(2) + ); + expect(screen.queryByText("question 1")).toBeNull(); + expect(historyRequests).toBe(2); }); test("route changes clear prior content and fence stale Session responses", async () => { @@ -400,7 +496,10 @@ test("route changes clear prior content and fence stale Session responses", asyn latestListener().onFrame( subscribedFrame(sessionWithUserMessage("previous worker content")), ); - expect(await screen.findByText("previous worker content")).not.toBeNull(); + expect( + await within(screen.getByRole("article", { name: "main transcript" })) + .findByText("previous worker content"), + ).not.toBeNull(); latestListener().onStatus?.("closed", "reload old worker"); await fireEvent.click( @@ -408,7 +507,10 @@ test("route changes clear prior content and fence stale Session responses", asyn ); await waitFor(() => expect(firstWorkerLoads).toBe(2)); await view.rerender({ data: pageData("worker-b", "runtime-b") }); - expect(screen.queryByText("previous worker content")).toBeNull(); + expect( + within(screen.getByRole("article", { name: "main transcript" })) + .queryByText("previous worker content"), + ).toBeNull(); expect(screen.getByRole("status").textContent).toContain( "Loading conversation", ); @@ -417,7 +519,12 @@ test("route changes clear prior content and fence stale Session responses", asyn availability: "retained_snapshot", snapshot: sessionWithUserMessage("current worker content"), })); - expect(await screen.findByText("current worker content")).not.toBeNull(); + expect( + await within(screen.getByRole("article", { name: "main transcript" })) + .findByText( + "current worker content", + ), + ).not.toBeNull(); staleReload.resolve(Response.json({ availability: "unavailable", @@ -425,8 +532,15 @@ test("route changes clear prior content and fence stale Session responses", asyn })); await settleMicrotasks(); expect(screen.queryByText("stale worker error")).toBeNull(); - expect(screen.queryByText("previous worker content")).toBeNull(); - expect(screen.getByText("current worker content")).not.toBeNull(); + expect( + within(screen.getByRole("article", { name: "main transcript" })) + .queryByText("previous worker content"), + ).toBeNull(); + expect( + within(screen.getByRole("article", { name: "main transcript" })).getByText( + "current worker content", + ), + ).not.toBeNull(); }); test("loads retained conversation turns five at a time into the log and turn bar", async () => { @@ -436,6 +550,7 @@ test("loads retained conversation turns five at a time into the log and turn bar page: { session_id: "session-a", lineage_id: "lineage-a", + compact_ancestor_lineage_ids: [], turns: [8, 9, 10, 11, 12].map((index) => ({ turn_id: `user-${index}`, entries: [ @@ -466,6 +581,7 @@ test("loads retained conversation turns five at a time into the log and turn bar page: { session_id: "session-a", lineage_id: "lineage-a", + compact_ancestor_lineage_ids: [], turns: [3, 4, 5, 6, 7].map((index) => ({ turn_id: `user-${index}`, entries: [{ @@ -567,7 +683,10 @@ test("keeps live content when history fails and retries the bounded page explici subscribedFrame(sessionWithUserMessage("current response remains")), ); - expect(await screen.findByText("current response remains")).not.toBeNull(); + expect( + await within(screen.getByRole("article", { name: "main transcript" })) + .findByText("current response remains"), + ).not.toBeNull(); await fireEvent.click( await screen.findByRole("button", { name: "Earlier conversation unavailable ยท retry", @@ -577,7 +696,11 @@ test("keeps live content when history fails and retries the bounded page explici expect(screen.getAllByText("question 1")).toHaveLength(2) ); expect(historyRequests).toBe(2); - expect(screen.getByText("current response remains")).not.toBeNull(); + expect( + within(screen.getByRole("article", { name: "main transcript" })).getByText( + "current response remains", + ), + ).not.toBeNull(); expect(screen.getAllByText("Start of conversation")).toHaveLength(2); });