Name the fold in the transcript, and share the window frames with the daemon
The window opens after the session's context fold, so the daemon now says which record that fold drained through and the console mentions it at the window's edge — while the window is above it. A reader who has paged back past the fold is reading the prefix itself, and there is nothing summarized about what they are looking at, so the note goes. The window request is `after_seq: 0` with a `limit` rather than `after_seq: null`: a daemon that does not know about windows ignores the limit and replays the session, which is what it did before, so the request cannot leave a client against an older daemon with an empty transcript. The fixture reports the fold and carries it on pages, and its folds are `lcm_frontier` records with the payload the runtime actually writes. `src/lib/protocol-v2.test.ts` replays the exact frames `klbr-ipc/src/v2.rs`'s `window_frames` suite pins, through this console's parser and client. Two implementations of one contract drift in silence otherwise: a renamed field is an empty session on one side with nothing red on the other.
299 unmodified linesExpand all30030130230330430530430530630730830931031131231331431531631731831031131932032132232312 unmodified lines33633733833933233334034134234333634434533834634734834935035135235335435535692 unmodified lines44945045145245345445545645745845944044144244344446046146246346446546646744846846947047147221 unmodified lines49449549649747849849950050150296 unmodified lines299 unmodified linesExpand all};
/** * A windowed history is what the console asks for: the newest records, anchored at the session's own * last compaction, with the floor it can still serve. A `history` request pages further back. * A windowed history is what the console asks for: the newest records, bounded by the session's own fold * and by the size asked for. The mirror of the daemon's `plan_replay`: the fold is a durable * `lcm_frontier` record, and everything at or below what it drained through is carried by its summary. */function foldOf(records: V2EventRecord[]): number | null { const fold = [...records].reverse().find((record) => record.lane === "lifecycle" && (record.payload as { kind?: string }).kind === "lcm_frontier"); if (!fold) return null; const data = (fold.payload as { data?: { drained_through_seq?: number } }).data; return data?.drained_through_seq ?? fold.seq - 1;}
function windowFloor(records: V2EventRecord[], limit: number | null | undefined): number { const newest = records.at(-1)?.seq ?? 0; const fold = foldOf(records); const floorByLimit = limit && limit > 0 ? newest - limit + 1 : 1; const fold = [...records].reverse().find((record) => record.lane === "lifecycle" && (record.payload as { kind?: string }).kind === "context.compacted"); const floorByFold = fold ? fold.seq : 1; const floorByFold = fold === null ? 1 : fold + 1; return Math.max(1, floorByFold, floorByLimit);}
/**12 unmodified lines records.push({ seq, session_id: "default", lane, created_ms: at, payload } as V2EventRecord); };
for (let t = 0; t < turns; t++) { // The runtime folds its context when it fills, which is why a transcript has boundaries in it at // all: the console's window opens at the newest of these and everything above is history. // A fold: the runtime installs a context frontier when it has summarized a prefix, and the durable // record of it says which record it drained through. That record is the boundary a window opens at. if (t > 0 && t % 40 === 0) { push("lifecycle", { kind: "context.compacted", kind: "lcm_frontier", details: `folded ${t} turns of session context`, data: { before_tokens: 18422 + t * 37, after_tokens: 6910 + t * 11, retained: ["handoff invariants", "reference graph", "test receipts"] }, data: { epoch: 1 + t / 40, drained_through_seq: seq, summary: `turns 1 to ${t}: the runtime crate and its focused suite`, source_children: [`bulk-${t}`], created_ms: at, }, }); } push("input", { source: t % 5 === 0 ? "ci/watch" : "operator",92 unmodified lines const script = ws.data.mode === "bulk" ? bulkFor(ws.data.bulkTurns) : SCRIPT; const records = ws.data.mode === "gap" ? script.filter((record) => record.seq !== 18) : script; const earliest = records[0]?.seq ?? 0; const latestSeq = records.at(-1)?.seq ?? 0; const fold = foldOf(records); // No cursor at all is a live-only subscription: nothing to send, the trickle alone is the answer. if (message.after_seq === null || message.after_seq === undefined) { if (ws.data.live) scheduleLive(ws); return; }
// A client that holds nothing gets a window, not a session: the newest records, anchored at the last // compaction. A client with a cursor is asking for the gap in front of it and gets exactly that. const resuming = message.after_seq !== null && message.after_seq !== undefined; const from = resuming ? (message.after_seq as number) + 1 : windowFloor(records, message.limit); const replay = records.filter((record) => record.seq >= from); // A client with a cursor is asking for the gap in front of it and gets all of it. A client that holds // nothing is asking for what it can hold: the newest window, bounded by the fold and by its limit. const resuming = message.after_seq > 0; const from = resuming ? message.after_seq + 1 : windowFloor(records, message.limit); const replay = records.filter((record) => record.seq >= from && record.seq <= latestSeq); ws.data.nextSeq = latestSeq + 1; send(ws, { type: "sync_started", session_id: message.session_id, from_seq: replay[0]?.seq ?? from, to_seq: latestSeq, earliest_seq: earliest, from_seq: replay[0]?.seq ?? from, to_seq: latestSeq, earliest_seq: earliest, fold_seq: fold, }); for (const fixture of replay) { send(ws, { type: "event", record: { ...fixture, session_id: message.session_id } }); }21 unmodified lines const page = behind.slice(Math.max(0, behind.length - Math.max(1, message.limit))); send(ws, { type: "sync_started", session_id: message.session_id, from_seq: page[0]?.seq ?? message.before_seq, to_seq: page.at(-1)?.seq ?? message.before_seq, earliest_seq: earliest, earliest_seq: earliest, fold_seq: foldOf(records), }); for (const fixture of page) { send(ws, { type: "event", record: { ...fixture, session_id: message.session_id } }); }96 unmodified lines71 unmodified lines72737475767778767778798081808182828384858687888990301 unmodified linesExpand all392393394395392393396397398399400401402403404203 unmodified linesExpand all71 unmodified lines * the one that reports the answer. */ $: earlier = $conn.hasEarlier === true || $conn.loadingEarlier === true; /** * When the window opens on a compaction, that is the whole reason it opens there, and it is the one * thing about the boundary a reader cannot infer from the records: a turn is not obviously the first * one of anything. The preset hides lifecycle rows, so the edge says it instead. * A window that opens at a fold is not obviously the start of anything: the preset hides lifecycle * rows, so the record that folded the prefix is not in the transcript to explain it. The daemon says * which record the fold drained through, and the edge mentions it while the loaded window is above it * — a reader who has paged back past the fold is looking at the prefix itself, and there is nothing * summarized about what they are reading. */ $: windowOpensAtFold = timelineItems[0]?.lane === "lifecycle" && (timelineItems[0].data as { kind?: string }).kind === "context.compacted"; $: floorSeq = timelineItems[0]?.seq; $: foldAtWindow = floorSeq !== undefined && $conn.foldSeq !== undefined && floorSeq > $conn.foldSeq ? $conn.foldSeq : undefined; $: rows = buildRows(turns, openRuns, earlier);
/** The newest turn is the one that is still being written to. */ $: liveTurnId = turns.length > 0 ? turns[turns.length - 1].id : undefined;301 unmodified linesExpand all title="ask the daemon for the records behind this window"> <History size={13} stroke-width={1.5} /> {#if $conn.loadingEarlier}loading earlier history{:else}load earlier history{/if} </button> {#if windowOpensAtFold} <span class="mute hist-note">the loaded window starts at a context compaction</span> {#if foldAtWindow !== undefined} <span class="mute hist-note" title="the context summary carries every record through {foldAtWindow}; this transcript is what came after it"> everything before this point is carried by the context summary </span> {/if} </div> </div> {:else if row.kind === "head"}203 unmodified linesExpand all1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586import { describe, test, expect } from "bun:test";import { get } from "svelte/store";import { parseServerMsgV2 } from "./protocol-v2";import { StreamingClientV2, WINDOW_LIMIT } from "./streaming-v2";
// Mock WebSocket: this suite only needs the socket the client writes its requests to.class MockWebSocket { public static readonly OPEN = 1; public static instances: MockWebSocket[] = []; public sent: string[] = []; public readyState: number = MockWebSocket.OPEN; public url: string; public onopen: (() => void) | null = null; public onmessage: ((event: { data: string }) => void) | null = null; public onerror: (() => void) | null = null; public onclose: (() => void) | null = null; constructor(url: string) { this.url = url; MockWebSocket.instances.push(this); } send(data: string) { this.sent.push(data); } close() {}}
(globalThis as any).WebSocket = MockWebSocket;
/** * The same frames the daemon's own suite pins (`klbr-ipc/src/v2.rs`, `window_frames`), replayed through * the parser this console uses. Two implementations of one contract drift in silence otherwise: a renamed * field here is an empty session there, with nothing red on either side. */describe("window frames shared with the daemon", () => { const frames = { started: '{"type":"sync_started","session_id":"default","from_seq":26,"to_seq":31,"earliest_seq":1,"fold_seq":25}', completedNoPage: '{"type":"sync_completed","session_id":"default","latest_seq":31}', completedWithFloor: '{"type":"sync_completed","session_id":"default","latest_seq":31,"earliest_seq":1}', event: '{"type":"event","record":{"seq":26,"session_id":"default","lane":"input","payload":{"source":"operator","text":"turn 361"},"created_ms":1010}}', };
test("the daemon's range, floor and fold are read as the console reads them", () => { const started = parseServerMsgV2(frames.started); expect(started.type).toBe("sync_started"); if (started.type !== "sync_started") throw new Error("unreachable"); expect(started.from_seq).toBe(26); expect(started.to_seq).toBe(31); expect(started.earliest_seq).toBe(1); expect(started.fold_seq).toBe(25);
const bare = parseServerMsgV2(frames.completedNoPage); if (bare.type !== "sync_completed") throw new Error("unreachable"); expect(bare.earliest_seq).toBeUndefined(); const floored = parseServerMsgV2(frames.completedWithFloor); if (floored.type !== "sync_completed") throw new Error("unreachable"); expect(floored.earliest_seq).toBe(1); });
test("a client fed those frames holds a window it can page and knows where it starts", () => { const sessionId = "daemon-frames-" + Math.random().toString(36).slice(2, 7); const client = new StreamingClientV2("ws://test", sessionId); client.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 26, to_seq: 31, earliest_seq: 1, fold_seq: 25 }); client.handleServerMsg(parseServerMsgV2(frames.event)); client.handleServerMsg(parseServerMsgV2(frames.completedWithFloor));
const conn = get(client.connectionStore); // The floor is under the window, so there is history to fetch; the fold is above nothing the console // holds, so the window it has does start at the session's fold. expect(conn.hasEarlier).toBe(true); expect(conn.foldSeq).toBe(25); // The replay's range ends at the newest record, so the live cursor does too: everything the daemon // announced is what the console was sent. expect(conn.lastSeenSeq).toBe(31); expect(get(client.timelineStore).map((item) => item.seq)).toEqual([26]); });
test("the console's own requests are the shapes the daemon parses", () => { const sessionId = "daemon-requests-" + Math.random().toString(36).slice(2, 7); const client = new StreamingClientV2("ws://test", sessionId, "daemon-requests"); client.connect(); const socket = MockWebSocket.instances[MockWebSocket.instances.length - 1]; socket.readyState = MockWebSocket.OPEN; socket.onopen?.(); expect(JSON.parse(socket.sent[0] ?? "{}")).toEqual({ type: "subscribe", session_id: sessionId, after_seq: 0, limit: WINDOW_LIMIT }); });});510 unmodified linesExpand all511512513514515515516517518519520521522523524525526527528177 unmodified linesExpand all706707708709701702703710711712713714715716717583 unmodified linesExpand all510 unmodified linesExpand all | { type: "ping"; timestamp?: number | null };
// daemon -> client v2 messagesexport type V2ServerMsg = | { type: "sync_started"; session_id: string; from_seq: number; to_seq: number; earliest_seq?: number | null } | { type: "sync_started"; session_id: string; from_seq: number; to_seq: number; /** The oldest record the daemon can still serve; absent = it cannot page. */ earliest_seq?: number | null; /** The record the session's active context fold drained through, when it has one. */ fold_seq?: number | null; } | { type: "event"; record: V2EventRecord } | { type: "sync_completed"; session_id: string; latest_seq: number; earliest_seq?: number | null } | { type: "live_chunk"; session_id: string; lane: Lane; execution_id?: string | null; stream?: string | null; delta: string } | { type: "ack"; session_id: string; seq: number; client_request_id?: string | null; duplicate?: boolean | null }177 unmodified linesExpand all}
// Compatibility types and helper functions for klbr-web consumers/** The window request and the history request, and what the daemon reports about its own history. * `subscribe` with `after_seq: null` asks for the newest window it serves — at most `limit` records, * anchored at its own last compaction — rather than for the whole session; it reports the range it * chose in `sync_started` and how far back it can still serve in `sync_completed.earliest_seq`. * `subscribe` with `after_seq: 0` and a `limit` asks for the newest window it serves — at most `limit` * records, opening after the session's own context fold — rather than for the whole session. The daemon * reports the range it chose, how far back it can still serve, and the record its fold drained through in * `sync_started`; a daemon that knows nothing about windows ignores the limit and replays everything. * `history` asks for the records behind a loaded window, for a reader who pressed a button, and is * answered with the same replay shape as a subscribe. * * A range and the floor under it are announced together, in `sync_started`, and not only when the583 unmodified linesExpand all62 unmodified lines63646566676869706768697071727374526 unmodified linesExpand all60160260360460560560660760860919 unmodified lines629630631632633634635636637638639640641642643644645646647648649650651652653483 unmodified linesExpand all62 unmodified lines test("a client that holds nothing asks for a window, not for a session", () => { expect(ws.sent.length).toBeGreaterThanOrEqual(1); const subMsg = JSON.parse(ws.sent[0]); expect(subMsg.type).toBe("subscribe"); // `after_seq: null` is "this client holds nothing": the daemon answers with its newest window, // anchored at its own last compaction, and at most `limit` records of it. A session this console // could not fit in memory is not a session it should ask to replay whole. expect(subMsg.after_seq).toBeNull(); // Nothing held is `after_seq: 0` with a `limit`: the newest window, anchored at the daemon's own fold, // and at most that many records of it. Zero without a limit is what this console used to ask for, and // a daemon that does not know about windows still answers it with everything. expect(subMsg.after_seq).toBe(0); expect(subMsg.limit).toBe(WINDOW_LIMIT); expect(get(client.timelineStore).length).toBe(0); });
526 unmodified linesExpand all expect(get(reloaded.connectionStore).hasEarlier).toBeUndefined(); reloaded.connect(); await new Promise((resolve) => setTimeout(resolve, 30)); const sub = JSON.parse(MockWebSocket.instances[MockWebSocket.instances.length - 1].sent[0]); expect(sub.after_seq).toBeNull(); expect(sub.after_seq).toBe(0); expect(sub.limit).toBe(WINDOW_LIMIT); });
test("the daemon's floor is what makes the console offer to load more", () => {19 unmodified lines floorClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 102, earliest_seq: 100 }); expect(get(floorClient.connectionStore).hasEarlier).toBe(false); });
test("the fold is the daemon's fact about the session, reported with every range", () => { const sessionId = "fold-" + Math.random().toString(36).slice(2, 7); const foldClient = new StreamingClientV2("ws://test", sessionId); expect(get(foldClient.connectionStore).foldSeq).toBeUndefined();
foldClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 401, to_seq: 900, earliest_seq: 1, fold_seq: 400 }); expect(get(foldClient.connectionStore).foldSeq).toBe(400);
// A later range that mentions no fold leaving the fact alone: a session does not become unfolded. foldClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 380, to_seq: 400, earliest_seq: 1 }); expect(get(foldClient.connectionStore).foldSeq).toBe(400);
// A page behind the fold carries it too, and the view is the one that decides whether to mention it. foldClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 300, to_seq: 379, earliest_seq: 1, fold_seq: 400 }); expect(get(foldClient.connectionStore).foldSeq).toBe(400); });
test("the floor arrives with the range, not only at the end of the replay", () => { const sessionId = "floor-early-" + Math.random().toString(36).slice(2, 7); const early = new StreamingClientV2("ws://test", sessionId);
483 unmodified linesExpand all137 unmodified linesExpand all138139140141142143144145146147148149150151152213 unmodified linesExpand all366367368369363364370371372373374375376377369378379380381382127 unmodified linesExpand all51051151251351451551651751824 unmodified lines543544545546547548549550551552553554750 unmodified linesExpand all137 unmodified linesExpand all rejectedCount: number; lastRejection?: { seq: number; reason: string }; /** Records behind the loaded window that the daemon says it can still serve. */ hasEarlier?: boolean; /** * The record the session's active context fold drained through. Everything at or below it is carried by * the context summary rather than replayed, so a window that starts above it starts there *because* of * the fold. It is a fact about the session, not about one replay: the view decides whether the window * on screen is above it. */ foldSeq?: number; /** True while a press of "load earlier history" is in flight. */ loadingEarlier?: boolean;}
213 unmodified linesExpand all reconnectAttempts: 0, error: undefined, });
// What this client wants depends on what it holds. Nothing loaded: the daemon's newest window, // not its whole history. A window already on screen: only the gap ahead of its live cursor. // What this client wants depends on what it holds. Nothing loaded: the daemon's newest window, not // its whole history — `after_seq: 0` with a `limit` is that request, and a daemon that does not know // about windows ignores the limit and replays everything, which is what it already did. A window // already on screen wants only the gap ahead of its live cursor, and says so without a limit. const storesEmpty = get(this.timelineStore).length === 0; this.send({ type: "subscribe", session_id: this.sessionId, after_seq: storesEmpty ? null : this.lastSeenSeq, after_seq: storesEmpty ? 0 : this.lastSeenSeq, ...(storesEmpty ? { limit: WINDOW_LIMIT } : {}), }); };
127 unmodified linesExpand all this.lastSeenSeq = 0; this.earliestSeq = null; this.backfilling = false; this.windowStart = null; this.updateConnection({ foldSeq: undefined, hasEarlier: undefined, loadingEarlier: undefined }); this.seenSeqs.clear(); this.eventBuffer.clear(); this.activeScratchpadId = null; this.activeCellId = null;24 unmodified lines // is already reading it. if (msg.earliest_seq !== undefined && msg.earliest_seq !== null) { this.earliestSeq = msg.earliest_seq; } // A session fact, sent with every range: the window on screen may or may not be above it. if (msg.fold_seq !== undefined && msg.fold_seq !== null) { this.updateConnection({ foldSeq: msg.fold_seq }); } this.windowStart = msg.from_seq; this.updateConnection({ isSyncing: true }); this.publishHistory(); break;750 unmodified linesExpand all