From 709f3c5530b1b2050174ee040d405736542d429f Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 30 Sep 2026 11:16:23 +0900 Subject: [PATCH] fix(console): reconcile history around lineage anchors --- crates/protocol/src/lib.rs | 14 ++++ crates/protocol/src/typescript.rs | 9 +-- crates/session-store/src/public_snapshot.rs | 42 ++++++++++- web/workspace/src/lib/generated/protocol.ts | 9 ++- .../src/lib/workspace/console/history.test.ts | 45 ++++++++++++ .../src/lib/workspace/console/history.ts | 26 ++++++- .../src/lib/workspace/console/model.test.ts | 30 ++++++++ .../src/lib/workspace/console/model.ts | 59 ++++++++++++++-- .../console/console-display.browser.spec.ts | 70 ++++++++++++++++++- 9 files changed, 288 insertions(+), 16 deletions(-) diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 851bead6e..298dccc52 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -842,6 +842,16 @@ pub struct SessionConversationTurn { pub entries: Vec, } +/// The immediate parent lineage and the stable real-user boundary adopted from it. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] +pub struct SessionHistoryLineageBoundary { + pub lineage_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub adopted_through_turn_id: Option, +} + /// Bounded backward page over the adopted conversation lineage. /// /// Turns remain chronological within a page. `next_cursor` is an opaque, @@ -859,6 +869,10 @@ pub struct SessionHistoryPage { /// cursor with this page's boundary. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub compact_ancestor_lineage_ids: Vec, + /// Immediate parent relationship used to retain only the adopted prefix + /// across fork/rewind refreshes that have no newest-page overlap. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent_lineage: Option, pub turns: Vec, #[serde(default, skip_serializing_if = "Option::is_none")] pub next_cursor: Option, diff --git a/crates/protocol/src/typescript.rs b/crates/protocol/src/typescript.rs index e9a725148..d0a051dae 100644 --- a/crates/protocol/src/typescript.rs +++ b/crates/protocol/src/typescript.rs @@ -12,10 +12,10 @@ use crate::{ PasteArtifactAvailability, PasteArtifactMediaType, PasteArtifactRef, PendingSubmissionSummary, PendingSubmissionsSnapshot, Permission, RewindSummary, RewindTarget, RewindTargetId, RunResult, ScopeRule, Segment, SessionContentPart, SessionConversationTurn, SessionEntryProvenance, - SessionHistoryPage, SessionMessageRole, SessionSnapshot, SessionSnapshotEntry, - SessionSnapshotEntryData, SessionToolAttachment, SubmissionDisposition, SymlinkPolicy, - ToolResultDisposition, TurnResult, UploadedFileAvailability, UploadedFileRef, WorkerBusyState, - WorkerCommandAcknowledgement, WorkerCommandDisposition, WorkerCommandEnvelope, + SessionHistoryLineageBoundary, SessionHistoryPage, SessionMessageRole, SessionSnapshot, + SessionSnapshotEntry, SessionSnapshotEntryData, SessionToolAttachment, SubmissionDisposition, + SymlinkPolicy, ToolResultDisposition, TurnResult, UploadedFileAvailability, UploadedFileRef, + WorkerBusyState, WorkerCommandAcknowledgement, WorkerCommandDisposition, WorkerCommandEnvelope, WorkerCommandKind, WorkerEvent, WorkerMaintenanceState, WorkerRunState, WorkerState, WorkerStateSnapshot, WorkerStatus, subscription::{ @@ -106,6 +106,7 @@ pub fn generated_protocol_types() -> String { push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); diff --git a/crates/session-store/src/public_snapshot.rs b/crates/session-store/src/public_snapshot.rs index 1b9af6333..6c8919571 100644 --- a/crates/session-store/src/public_snapshot.rs +++ b/crates/session-store/src/public_snapshot.rs @@ -4,8 +4,8 @@ use base64::{ }; use protocol::{ Segment, SessionContentPart, SessionConversationTurn, SessionEntryProvenance, - SessionHistoryPage, SessionMessageRole, SessionSnapshot, SessionSnapshotEntry, - SessionSnapshotEntryData, SessionToolAttachment, + SessionHistoryLineageBoundary, SessionHistoryPage, SessionMessageRole, SessionSnapshot, + SessionSnapshotEntry, SessionSnapshotEntryData, SessionToolAttachment, }; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; @@ -261,6 +261,28 @@ pub fn read_retained_session_history_page( } else { None }; + let parent_lineage = if 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 + }, + )?; + 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()), + }) + } else { + None + }; let mut turns = read_history_turns_backward( &store, active.session_id, @@ -302,6 +324,7 @@ pub fn read_retained_session_history_page( session_id: active.session_id.to_string(), lineage_id, compact_ancestor_lineage_ids, + parent_lineage, turns: turns .into_iter() .map(|positioned| positioned.turn) @@ -1735,6 +1758,21 @@ mod tests { .all(|entry| entry.entry_id != "assistant-1-unadopted") })); assert!(page.turns.iter().all(|turn| turn.turn_id != "user-2")); + let store = WorkerSessionStore::open_read_only(aggregate_root.join("session")).unwrap(); + let (_, source_lineage_id, _, _) = load_adopted_lineage( + &store, + session_id, + source_segment, + RetainedHistoryReadLimits::default(), + ) + .unwrap(); + assert_eq!( + page.parent_lineage, + Some(SessionHistoryLineageBoundary { + lineage_id: source_lineage_id, + adopted_through_turn_id: Some("user-1".to_string()), + }) + ); } #[test] diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index 2fc6f64a3..1d7d29673 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -151,6 +151,8 @@ timestamp: number, provenance: SessionEntryProvenance, derived_from?: Array, }; +export type SessionHistoryLineageBoundary = { lineage_id: string, adopted_through_turn_id?: string | null, }; + export type SessionHistoryPage = { session_id: string, /** * Stable identity of the adopted active lineage used to fence stale pages. @@ -162,7 +164,12 @@ lineage_id: string, * 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, }; +compact_ancestor_lineage_ids?: Array, +/** + * Immediate parent relationship used to retain only the adopted prefix + * across fork/rewind refreshes that have no newest-page overlap. + */ +parent_lineage?: SessionHistoryLineageBoundary | null, 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 47c302a95..beb857fea 100644 --- a/web/workspace/src/lib/workspace/console/history.test.ts +++ b/web/workspace/src/lib/workspace/console/history.test.ts @@ -56,11 +56,16 @@ function page( cursor: string | null, lineageId = "lineage-a", compactAncestorLineageIds: string[] = [], + parentLineage?: { + lineage_id: string; + adopted_through_turn_id?: string | null; + }, ): SessionHistoryPage { return { session_id: "session-a", lineage_id: lineageId, compact_ancestor_lineage_ids: compactAncestorLineageIds, + parent_lineage: parentLineage, turns, next_cursor: cursor, has_more: cursor !== null, @@ -240,6 +245,46 @@ Deno.test("Compact lineage refresh preserves already loaded adopted history", () ); }); +Deno.test("fork lineage boundary preserves an adopted prefix without page overlap", () => { + let state: ConsoleHistoryState = { + ...emptyConsoleHistoryState(), + sessionId: "session-a", + lineageId: "lineage-a", + turns: Array.from({ length: 50 }, (_, index) => turn(index + 1)), + cursor: null, + hasMore: false, + status: "loading", + requestedCursor: "__initial__", + }; + state = applyConsoleHistoryPage( + state, + page( + [101, 102, 103, 104, 105].map(turn), + "cursor-101", + "lineage-fork", + [], + { + lineage_id: "lineage-a", + adopted_through_turn_id: "u-25", + }, + ), + null, + ); + + assert( + state.turns.map((value) => value.turn_id).join(",") === + [ + ...Array.from({ length: 25 }, (_, index) => `u-${index + 1}`), + ...Array.from({ length: 5 }, (_, index) => `u-${index + 101}`), + ].join(","), + "fork refresh must retain only the authoritative adopted prefix", + ); + assert( + state.cursor === "cursor-101" && state.hasMore, + "fork refresh must replace the old paging boundary", + ); +}); + Deno.test("changed lineage without a stable overlap replaces retained turns", () => { let state: ConsoleHistoryState = { ...emptyConsoleHistoryState(), diff --git a/web/workspace/src/lib/workspace/console/history.ts b/web/workspace/src/lib/workspace/console/history.ts index ac994132b..adf0f148e 100644 --- a/web/workspace/src/lib/workspace/console/history.ts +++ b/web/workspace/src/lib/workspace/console/history.ts @@ -86,6 +86,9 @@ export function applyConsoleHistoryPage( state.turns.length > 0; const refreshingSameLineage = refreshingSameSession && state.lineageId === page.lineage_id; + const refreshingParentLineage = refreshingSameSession && + state.lineageId !== null && + page.parent_lineage?.lineage_id === state.lineageId; const refreshingCompactDescendant = refreshingSameSession && state.lineageId !== null && (page.compact_ancestor_lineage_ids ?? []).includes(state.lineageId); @@ -93,7 +96,15 @@ export function applyConsoleHistoryPage( (turn) => state.turns.some((retained) => retained.turn_id === turn.turn_id), ); const turns = initial - ? refreshingSameLineage || refreshingCompactDescendant + ? refreshingSameLineage + ? mergeRefreshedTurns(state.turns, incoming) + : refreshingParentLineage + ? reconcileParentLineage( + state.turns, + incoming, + page.parent_lineage?.adopted_through_turn_id ?? null, + ) + : refreshingCompactDescendant ? mergeRefreshedTurns(state.turns, incoming) : refreshingSameSession ? reconcileChangedLineage(state.turns, incoming) @@ -216,6 +227,19 @@ function previewAssistant(value: string): string[] { .slice(0, 3); } +function reconcileParentLineage( + retained: readonly SessionConversationTurn[], + refreshed: readonly SessionConversationTurn[], + adoptedThroughTurnId: string | null, +): SessionConversationTurn[] { + if (adoptedThroughTurnId === null) return dedupeTurns(refreshed); + const boundary = retained.findIndex((turn) => + turn.turn_id === adoptedThroughTurnId + ); + if (boundary < 0) return dedupeTurns(refreshed); + return dedupeTurns([...retained.slice(0, boundary + 1), ...refreshed]); +} + function reconcileChangedLineage( 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 59c02bdab..fccfc53d0 100644 --- a/web/workspace/src/lib/workspace/console/model.test.ts +++ b/web/workspace/src/lib/workspace/console/model.test.ts @@ -3112,6 +3112,36 @@ Deno.test("committed history preserves unmatched current rows in stable entry or ); }); +Deno.test("committed history places omitted retained rows between snapshot anchors", () => { + const makeLine = (prefix: string, turn: number) => { + const line = consoleLine(`${prefix}-${turn}`, "user"); + line.entryId = `user-${turn}`; + return line; + }; + const history = [2, 3, 4, 5, 6].map((turn) => makeLine("history", turn)); + const current = [1, 2, 6].map((turn) => makeLine("current", turn)); + + assertEquals( + mergeCommittedHistoryLines(history, current).map((line) => line.entryId), + [1, 2, 3, 4, 5, 6].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"); + line.entryId = `user-${turn}`; + return line; + }; + const history = [2, 3, 4, 5, 6].map((turn) => makeLine("history", turn)); + const current = [1, 2, 7].map((turn) => makeLine("current", turn)); + + assertEquals( + mergeCommittedHistoryLines(history, current).map((line) => line.entryId), + [1, 2, 3, 4, 5, 6, 7].map((turn) => `user-${turn}`), + ); +}); + Deno.test("committed history inserts current live rows after their last stable anchor", () => { const historyUser = consoleLine("history-user", "user"); historyUser.entryId = "user-1"; diff --git a/web/workspace/src/lib/workspace/console/model.ts b/web/workspace/src/lib/workspace/console/model.ts index 973a00736..716089bb5 100644 --- a/web/workspace/src/lib/workspace/console/model.ts +++ b/web/workspace/src/lib/workspace/console/model.ts @@ -291,12 +291,59 @@ export function mergeCommittedHistoryLines( Boolean(id) ), ); - return [ - ...history.filter((line) => - !line.entryId || !currentEntryIds.has(line.entryId) - ), - ...current, - ]; + const historyPositions = new Map(); + history.forEach((line, index) => { + if (line.entryId && !historyPositions.has(line.entryId)) { + historyPositions.set(line.entryId, index); + } + }); + + // Use stable entries that occur in the same order in both projections as + // anchors. History-only rows are inserted in their retained interval, while + // current-only rows retain the snapshot's authoritative order. This handles + // Compact snapshots that preserve only a non-contiguous subset of context. + const anchors: Array<{ history: number; current: number }> = []; + let lastHistoryIndex = -1; + current.forEach((line, currentIndex) => { + const historyIndex = line.entryId + ? historyPositions.get(line.entryId) + : undefined; + if (historyIndex !== undefined && historyIndex > lastHistoryIndex) { + anchors.push({ history: historyIndex, current: currentIndex }); + lastHistoryIndex = historyIndex; + } + }); + + if (anchors.length === 0) { + return [ + ...history.filter((line) => + !line.entryId || !currentEntryIds.has(line.entryId) + ), + ...current, + ]; + } + + const merged: ConsoleLine[] = []; + let historyCursor = 0; + let currentCursor = 0; + const appendHistoryInterval = (end: number) => { + for (; historyCursor < end; historyCursor += 1) { + const line = history[historyCursor]!; + if (!line.entryId || !currentEntryIds.has(line.entryId)) { + merged.push(line); + } + } + }; + for (const anchor of anchors) { + appendHistoryInterval(anchor.history); + merged.push(...current.slice(currentCursor, anchor.current)); + merged.push(current[anchor.current]!); + historyCursor = anchor.history + 1; + currentCursor = anchor.current + 1; + } + appendHistoryInterval(history.length); + merged.push(...current.slice(currentCursor)); + return merged; } export function projectSessionHistoryEntries( 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 c826379b0..9b8bded7c 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 @@ -121,13 +121,22 @@ function emptyHistoryPage() { }; } -function historyPage(indices: number[], cursor: string | null) { +function historyPage( + indices: number[], + cursor: string | null, + lineageId = "lineage-a", + parentLineage?: { + lineage_id: string; + adopted_through_turn_id?: string | null; + }, +) { return { availability: "page", page: { session_id: "session-a", - lineage_id: "lineage-a", + lineage_id: lineageId, compact_ancestor_lineage_ids: [], + parent_lineage: parentLineage, turns: indices.map((index) => ({ turn_id: `user-${index}`, entries: [ @@ -473,6 +482,63 @@ test("snapshot supersedes an in-flight page before stale lineage can render", as expect(historyRequests).toBe(2); }); +test("fork refresh reconciles an in-flight older page to the adopted prefix", async () => { + const staleOlder = 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 Promise.resolve( + Response.json(historyPage([46, 47, 48, 49, 50], "older-46")), + ); + } + if (historyRequests === 2) return staleOlder.promise; + return Promise.resolve(Response.json(historyPage( + [101, 102, 103, 104, 105], + "older-101", + "lineage-fork", + { + lineage_id: "lineage-a", + adopted_through_turn_id: "user-48", + }, + ))); + }), + ); + + render(ConsolePage, { data: pageData() }); + await waitFor(() => expect(multiplexer.subscribe).toHaveBeenCalledOnce()); + latestListener().onFrame(subscribedFrame()); + await waitFor(() => + expect(screen.getAllByText("question 50")).toHaveLength(2) + ); + + const transcriptScroller = screen.getByLabelText("main transcript") + .parentElement as HTMLElement; + transcriptScroller.scrollTop = 0; + await fireEvent.scroll(transcriptScroller); + await waitFor(() => expect(historyRequests).toBe(2)); + + latestListener().onFrame( + subscribedFrame(sessionWithUserMessage("fork current")), + ); + staleOlder.resolve(Response.json(historyPage([41, 42, 43, 44, 45], null))); + + await waitFor(() => + expect(screen.getAllByText("question 101")).toHaveLength(2) + ); + expect(screen.getAllByText("question 48")).toHaveLength(2); + expect(screen.queryByText("question 49")).toBeNull(); + expect(screen.getAllByText("question 41")).toHaveLength(2); + expect(historyRequests).toBe(3); +}); + test("route changes clear prior content and fence stale Session responses", async () => { const staleReload = deferred(); const current = deferred();