fix(console): preserve exact adopted history boundaries

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<String>,
+    pub adopted_through_turn: Option<SessionConversationTurn>,
 }
 
 /// Bounded backward page over the adopted conversation lineage.

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<Vec<PositionedHistoryTurn>, 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<_>>(),
+            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";

web/workspace/src/lib/generated/protocol.ts

@@ -151,7 +151,12 @@ timestamp: number, provenance: SessionEntryProvenance, derived_from?: Array<stri
 
 export type SessionConversationTurn = { turn_id: string, entries: Array<SessionSnapshotEntry>, };
 
-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,
 /**

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",

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(

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");

web/workspace/src/lib/workspace/console/model.ts

@@ -334,9 +334,16 @@ export function mergeCommittedHistoryLines(
       }
     }
   };
-  for (const anchor of anchors) {
+  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;

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,
         },
       )));
     }),