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

fix(console): reconcile history around lineage anchors

Browse Source
This commit is contained in:
Hare
2026-09-30 11:16:23 +09:00
parent 4976b6fd00
commit 709f3c5530
9 changed files with 288 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/protocol/src/lib.rs
+14
View File
@@ -842,6 +842,16 @@ pub struct SessionConversationTurn {
pub entries: Vec<SessionSnapshotEntry>,
}
/// 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<String>,
}
/// 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<String>,
/// 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<SessionHistoryLineageBoundary>,
pub turns: Vec<SessionConversationTurn>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub next_cursor: Option<String>,
crates/protocol/src/typescript.rs
+5 -4
View File
@@ -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::<SessionSnapshotEntryData>(&cfg, &mut output);
push_decl::<SessionSnapshotEntry>(&cfg, &mut output);
push_decl::<SessionConversationTurn>(&cfg, &mut output);
push_decl::<SessionHistoryLineageBoundary>(&cfg, &mut output);
push_decl::<SessionHistoryPage>(&cfg, &mut output);
push_decl::<PendingSubmissionSummary>(&cfg, &mut output);
push_decl::<PendingSubmissionsSnapshot>(&cfg, &mut output);
crates/session-store/src/public_snapshot.rs
+40 -2
View File
@@ -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]
web/workspace/src/lib/generated/protocol.ts
+8 -1
View File
@@ -151,6 +151,8 @@ 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 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<string>, turns: Array<SessionConversationTurn>, next_cursor?: string | null, has_more: boolean, };
compact_ancestor_lineage_ids?: Array<string>,
/**
* 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<SessionConversationTurn>, next_cursor?: string | null, has_more: boolean, };
export type PendingSubmissionSummary = { submission_id: string, accepted_at_ms: number, segment_count: number, byte_len: number, };
web/workspace/src/lib/workspace/console/history.test.ts
+45
View File
@@ -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(),
web/workspace/src/lib/workspace/console/history.ts
+25 -1
View File
@@ -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[],
web/workspace/src/lib/workspace/console/model.test.ts
+30
View File
@@ -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";
web/workspace/src/lib/workspace/console/model.ts
+53 -6
View File
@@ -291,12 +291,59 @@ export function mergeCommittedHistoryLines(
Boolean(id)
),
);
return [
...history.filter((line) =>
!line.entryId || !currentEntryIds.has(line.entryId)
),
...current,
];
const historyPositions = new Map<string, number>();
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(
web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/console-display.browser.spec.ts
+68 -2
View File
@@ -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<Response>();
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<Response>();
const current = deferred<Response>();