From c73e20be17490dd257e43effb8f2d5ede3c69949 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 30 Sep 2026 11:40:45 +0900 Subject: [PATCH] fix(console): preserve exact adopted history boundaries --- crates/protocol/src/lib.rs | 4 +- crates/session-store/src/public_snapshot.rs | 122 +++++++++++++----- web/workspace/src/lib/generated/protocol.ts | 7 +- .../src/lib/workspace/console/history.test.ts | 12 +- .../src/lib/workspace/console/history.ts | 17 ++- .../src/lib/workspace/console/model.test.ts | 15 +++ .../src/lib/workspace/console/model.ts | 13 +- .../console/console-display.browser.spec.ts | 6 +- 8 files changed, 151 insertions(+), 45 deletions(-) diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 298dccc52..6acbddf61 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -848,8 +848,10 @@ pub struct SessionConversationTurn { #[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] pub struct SessionHistoryLineageBoundary { pub lineage_id: String, + /// Canonical public form of the final user-visible turn at the exact + /// provider boundary. Entries after that boundary are excluded. #[serde(default, skip_serializing_if = "Option::is_none")] - pub adopted_through_turn_id: Option, + pub adopted_through_turn: Option, } /// Bounded backward page over the adopted conversation lineage. diff --git a/crates/session-store/src/public_snapshot.rs b/crates/session-store/src/public_snapshot.rs index 6c8919571..a3f75fb15 100644 --- a/crates/session-store/src/public_snapshot.rs +++ b/crates/session-store/src/public_snapshot.rs @@ -261,24 +261,25 @@ pub fn read_retained_session_history_page( } else { None }; - let parent_lineage = if lineage.len() > 1 { + let mut scanned_bytes = metadata_bytes; + let mut scanned_entries = metadata_entries; + let parent_lineage = if cursor.is_none() && lineage.len() > 1 { let parent_turns = read_history_turns_backward( &store, active.session_id, &lineage[1..], None, 1, - RetainedHistoryReadLimits { - max_scan_bytes: limits.max_scan_bytes.saturating_sub(metadata_bytes), - max_entries: limits.max_entries.saturating_sub(metadata_entries), - ..limits - }, + limits, + &mut scanned_bytes, + &mut scanned_entries, )?; Some(SessionHistoryLineageBoundary { lineage_id: lineage_identity(active.session_id, &lineage[1..]), - adopted_through_turn_id: parent_turns - .first() - .map(|positioned| positioned.turn.turn_id.clone()), + adopted_through_turn: parent_turns + .into_iter() + .next() + .map(|positioned| positioned.turn), }) } else { None @@ -289,11 +290,9 @@ pub fn read_retained_session_history_page( &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 - }, + limits, + &mut scanned_bytes, + &mut scanned_entries, )?; let has_more = turns.len() > page_limit; if has_more { @@ -641,9 +640,9 @@ fn read_history_turns_backward( start: Option<(HistoryTurnPosition, &str)>, turn_limit: usize, limits: RetainedHistoryReadLimits, + scanned_bytes: &mut u64, + scanned_entries: &mut usize, ) -> Result, RetainedHistoryReadError> { - let mut scanned_bytes = 0_u64; - let mut scanned_entries = 0_usize; let mut turns = Vec::new(); let mut pending_entries = Vec::new(); let mut seen_entries = HashSet::new(); @@ -676,7 +675,7 @@ fn read_history_turns_backward( let record_end = position .record_end_offset .ok_or(RetainedHistoryReadError::InvalidCursor)?; - let remaining = limits.max_scan_bytes.saturating_sub(scanned_bytes); + let remaining = limits.max_scan_bytes.saturating_sub(*scanned_bytes); let record = store .read_log_record_range_read_only_bounded( session_id, @@ -686,7 +685,7 @@ fn read_history_turns_backward( remaining, ) .map_err(map_history_cursor_store_error)?; - scanned_bytes = scanned_bytes + *scanned_bytes = scanned_bytes .checked_add(record.end_offset - record.start_offset) .ok_or(RetainedHistoryReadError::ResourceLimit)?; if record.start_offset != position.before_offset @@ -696,10 +695,10 @@ fn read_history_turns_backward( { return Err(RetainedHistoryReadError::InvalidCursor); } - scanned_entries = scanned_entries + *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 { + if *scanned_entries > limits.max_entries { return Err(RetainedHistoryReadError::ResourceLimit); } } @@ -738,8 +737,8 @@ fn read_history_turns_backward( session_id, segment, limits, - &mut scanned_bytes, - &mut scanned_entries, + scanned_bytes, + scanned_entries, )?; let before_offset = if starting_here { start_position @@ -756,20 +755,20 @@ fn read_history_turns_backward( .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 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 + *scanned_bytes = scanned_bytes .checked_add(bytes) .ok_or(RetainedHistoryReadError::ResourceLimit)?; let Some(record) = record else { break; }; - scanned_entries = scanned_entries + *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 { + if *scanned_entries > limits.max_entries { return Err(RetainedHistoryReadError::ResourceLimit); } if matches!(record.entry, LogEntry::AnnotatedSegmentStart { .. }) { @@ -1766,12 +1765,17 @@ mod tests { RetainedHistoryReadLimits::default(), ) .unwrap(); + let parent = page.parent_lineage.as_ref().unwrap(); + assert_eq!(parent.lineage_id, source_lineage_id); + let adopted_turn = parent.adopted_through_turn.as_ref().unwrap(); + assert_eq!(adopted_turn.turn_id, "user-1"); assert_eq!( - page.parent_lineage, - Some(SessionHistoryLineageBoundary { - lineage_id: source_lineage_id, - adopted_through_turn_id: Some("user-1".to_string()), - }) + adopted_turn + .entries + .iter() + .map(|entry| entry.entry_id.as_str()) + .collect::>(), + vec!["user-1", "assistant-1-boundary"] ); } @@ -2264,6 +2268,62 @@ mod tests { assert!(page.has_more); } + #[test] + fn retained_history_shares_one_entry_budget_with_parent_boundary_lookup() { + let worker_name = "worker-shared-parent-budget"; + 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..=5 { + 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, + )]; + for turn in 6..=10 { + append_turn(&mut active_log, turn); + } + let (_root, aggregate_root) = persist_history_fixture( + worker_name, + session_id, + active_segment, + vec![(source_segment, source_log), (active_segment, active_log)], + ); + + let error = read_retained_session_history_page( + &aggregate_root, + worker_name, + None, + Some(1), + RetainedHistoryReadLimits { + max_entries: 40, + ..RetainedHistoryReadLimits::default() + }, + ) + .unwrap_err(); + assert_eq!(error, RetainedHistoryReadError::ResourceLimit); + + let page = read_retained_session_history_page( + &aggregate_root, + worker_name, + None, + Some(1), + RetainedHistoryReadLimits { + max_entries: 100, + ..RetainedHistoryReadLimits::default() + }, + ) + .unwrap(); + assert_eq!(turn_ids(&page), vec!["user-10"]); + } + #[test] fn retained_history_enforces_page_scan_and_response_limits_without_splitting_turns() { let worker_name = "worker-history-limits"; diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index 1d7d29673..1d9b5ed1b 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -151,7 +151,12 @@ timestamp: number, provenance: SessionEntryProvenance, derived_from?: Array, }; -export type SessionHistoryLineageBoundary = { lineage_id: string, adopted_through_turn_id?: string | null, }; +export type SessionHistoryLineageBoundary = { lineage_id: string, +/** + * Canonical public form of the final user-visible turn at the exact + * provider boundary. Entries after that boundary are excluded. + */ +adopted_through_turn?: SessionConversationTurn | null, }; export type SessionHistoryPage = { session_id: string, /** diff --git a/web/workspace/src/lib/workspace/console/history.test.ts b/web/workspace/src/lib/workspace/console/history.test.ts index beb857fea..da5343067 100644 --- a/web/workspace/src/lib/workspace/console/history.test.ts +++ b/web/workspace/src/lib/workspace/console/history.test.ts @@ -58,7 +58,7 @@ function page( compactAncestorLineageIds: string[] = [], parentLineage?: { lineage_id: string; - adopted_through_turn_id?: string | null; + adopted_through_turn?: SessionConversationTurn | null; }, ): SessionHistoryPage { return { @@ -256,6 +256,10 @@ Deno.test("fork lineage boundary preserves an adopted prefix without page overla status: "loading", requestedCursor: "__initial__", }; + state.turns[24]!.entries.push( + message("a-unadopted-25", "assistant", "stale sibling response"), + ); + const adoptedBoundary = turn(25); state = applyConsoleHistoryPage( state, page( @@ -265,7 +269,7 @@ Deno.test("fork lineage boundary preserves an adopted prefix without page overla [], { lineage_id: "lineage-a", - adopted_through_turn_id: "u-25", + adopted_through_turn: adoptedBoundary, }, ), null, @@ -279,6 +283,10 @@ Deno.test("fork lineage boundary preserves an adopted prefix without page overla ].join(","), "fork refresh must retain only the authoritative adopted prefix", ); + assert( + historyEntries(state).every((entry) => entry.entry_id !== "a-unadopted-25"), + "the canonical provider boundary must trim a sibling entry in the same visible turn", + ); assert( state.cursor === "cursor-101" && state.hasMore, "fork refresh must replace the old paging boundary", diff --git a/web/workspace/src/lib/workspace/console/history.ts b/web/workspace/src/lib/workspace/console/history.ts index adf0f148e..12bd98845 100644 --- a/web/workspace/src/lib/workspace/console/history.ts +++ b/web/workspace/src/lib/workspace/console/history.ts @@ -102,7 +102,7 @@ export function applyConsoleHistoryPage( ? reconcileParentLineage( state.turns, incoming, - page.parent_lineage?.adopted_through_turn_id ?? null, + page.parent_lineage?.adopted_through_turn ?? null, ) : refreshingCompactDescendant ? mergeRefreshedTurns(state.turns, incoming) @@ -230,14 +230,21 @@ function previewAssistant(value: string): string[] { function reconcileParentLineage( retained: readonly SessionConversationTurn[], refreshed: readonly SessionConversationTurn[], - adoptedThroughTurnId: string | null, + adoptedThroughTurn: SessionConversationTurn | null, ): SessionConversationTurn[] { - if (adoptedThroughTurnId === null) return dedupeTurns(refreshed); + if (adoptedThroughTurn === null) return dedupeTurns(refreshed); const boundary = retained.findIndex((turn) => - turn.turn_id === adoptedThroughTurnId + turn.turn_id === adoptedThroughTurn.turn_id ); if (boundary < 0) return dedupeTurns(refreshed); - return dedupeTurns([...retained.slice(0, boundary + 1), ...refreshed]); + const refreshedIncludesBoundary = refreshed.some((turn) => + turn.turn_id === adoptedThroughTurn.turn_id + ); + return dedupeTurns([ + ...retained.slice(0, boundary), + ...(refreshedIncludesBoundary ? [] : [adoptedThroughTurn]), + ...refreshed, + ]); } function reconcileChangedLineage( diff --git a/web/workspace/src/lib/workspace/console/model.test.ts b/web/workspace/src/lib/workspace/console/model.test.ts index fccfc53d0..8ae2a0cc3 100644 --- a/web/workspace/src/lib/workspace/console/model.test.ts +++ b/web/workspace/src/lib/workspace/console/model.test.ts @@ -3127,6 +3127,21 @@ Deno.test("committed history places omitted retained rows between snapshot ancho ); }); +Deno.test("committed history places current context before an interior first anchor", () => { + const makeLine = (prefix: string, turn: number) => { + const line = consoleLine(`${prefix}-${turn}`, "user"); + line.entryId = `user-${turn}`; + return line; + }; + const history = [6, 7, 8, 9, 10].map((turn) => makeLine("history", turn)); + const current = [1, 8, 10].map((turn) => makeLine("current", turn)); + + assertEquals( + mergeCommittedHistoryLines(history, current).map((line) => line.entryId), + [1, 6, 7, 8, 9, 10].map((turn) => `user-${turn}`), + ); +}); + Deno.test("committed history places a retained suffix before newer current-only rows", () => { const makeLine = (prefix: string, turn: number) => { const line = consoleLine(`${prefix}-${turn}`, "user"); diff --git a/web/workspace/src/lib/workspace/console/model.ts b/web/workspace/src/lib/workspace/console/model.ts index 716089bb5..6521513d5 100644 --- a/web/workspace/src/lib/workspace/console/model.ts +++ b/web/workspace/src/lib/workspace/console/model.ts @@ -334,9 +334,16 @@ export function mergeCommittedHistoryLines( } } }; - for (const anchor of anchors) { - appendHistoryInterval(anchor.history); - merged.push(...current.slice(currentCursor, anchor.current)); + for (const [index, anchor] of anchors.entries()) { + if (index === 0) { + // Current rows before the first shared entry can be older retained + // context that the newest history page starts after. + merged.push(...current.slice(currentCursor, anchor.current)); + appendHistoryInterval(anchor.history); + } else { + appendHistoryInterval(anchor.history); + merged.push(...current.slice(currentCursor, anchor.current)); + } merged.push(current[anchor.current]!); historyCursor = anchor.history + 1; currentCursor = anchor.current + 1; 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 9b8bded7c..9d0a51b69 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 @@ -11,6 +11,7 @@ import { import { afterEach, beforeEach, expect, test, vi } from "vitest"; import type { Event as ProtocolEvent, + SessionConversationTurn, SessionSnapshot, } from "$lib/generated/protocol"; import type { Worker } from "$lib/workspace/sidebar/types"; @@ -127,7 +128,7 @@ function historyPage( lineageId = "lineage-a", parentLineage?: { lineage_id: string; - adopted_through_turn_id?: string | null; + adopted_through_turn?: SessionConversationTurn | null; }, ) { return { @@ -506,7 +507,8 @@ test("fork refresh reconciles an in-flight older page to the adopted prefix", as "lineage-fork", { lineage_id: "lineage-a", - adopted_through_turn_id: "user-48", + adopted_through_turn: historyPage([48], null).page + .turns[0] as SessionConversationTurn, }, ))); }),