Ask the daemon for a window instead of caching the transcript
The cache existed to make a reload cheap, and it could not: a 4800-record transcript is ~10 MB against a 5 MB sessionStorage quota, so the biggest sessions it was written for were the ones it failed on. It is gone, along with the remembered cursor and the diagnostics snapshot that were only there to avoid replaying records it held, and the client now asks for what it can hold: `subscribe` with `after_seq: null` and a `limit` is a window request, and the daemon answers with the newest window of at most that many records, anchored at its own last compaction, reporting its range and its floor in `sync_started` and `sync_completed`. `{ type: "history", before_seq, limit }` asks for the records behind a loaded window; the answer is the same replay shape, its records prepend into the timeline where their sequence puts them, and the live cursor does not move for them. Deduplication is a set of seen sequences instead of a scan of the transcript per record, which is what makes a retried or overlapping page idempotent. A daemon that says nothing about its floor offers no paging, so a client against today's daemon degrades to what it does now rather than to a button that does nothing. The fixture serves the same contract, with a compaction every forty turns of the bulk session, so the behaviour is reachable in a browser: at 400 turns the console opens on 481 records instead of 4809.
299 unmodified linesExpand all3003013023033043053063073083093103113123133143153163173183198 unmodified lines32832933033133233333433533633733833934034134234334486 unmodified lines43143243343441443543641741841943743843944044144244344444544644742244844945045145242745345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149250 unmodified lines54354454554654754854955055127 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. */function windowFloor(records: V2EventRecord[], limit: number | null | undefined): number { const newest = records.at(-1)?.seq ?? 0; 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; return Math.max(1, floorByFold, floorByLimit);}
/** * A session large enough to be worth virtualizing: `?mode=bulk` replays this instead of the * hand-written script. It uses the same record shapes, many more of them, and cell outputs of wildly * different heights — so dynamic measurement is exercised rather than assumed, and the console has a * transcript where the difference between "render everything" and "render the fold" is measurable.8 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. if (t > 0 && t % 40 === 0) { push("lifecycle", { kind: "context.compacted", 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"] }, }); } push("input", { source: t % 5 === 0 ? "ci/watch" : "operator", author: "dawn", text: `turn ${t + 1}: ${t % 3 === 0 ? "check the runtime crate and report only what the compiler agreed with." : "keep going; the previous step left a loose end in the scheduler."}`,86 unmodified lines send(ws, { type: "sync_completed", session_id: message.session_id, latest_seq: 0 }); return; }
const afterSeq = message.after_seq ?? 0; 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 replay = records.filter((record) => record.seq > afterSeq); const latestSeq = replay.at(-1)?.seq ?? afterSeq; ws.data.nextSeq = Math.max(records.at(-1)?.seq ?? 0, afterSeq) + 1; const earliest = records[0]?.seq ?? 0; const latestSeq = records.at(-1)?.seq ?? 0;
// 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); ws.data.nextSeq = latestSeq + 1; send(ws, { type: "sync_started", session_id: message.session_id, from_seq: afterSeq + 1, to_seq: latestSeq, from_seq: replay[0]?.seq ?? from, to_seq: latestSeq, earliest_seq: earliest, }); for (const fixture of replay) { send(ws, { type: "event", record: { ...fixture, session_id: message.session_id } }); } send(ws, { type: "sync_completed", session_id: message.session_id, latest_seq: latestSeq }); send(ws, { type: "sync_completed", session_id: message.session_id, latest_seq: latestSeq, earliest_seq: earliest, });
if (ws.data.mode === "error") { send(ws, { type: "error", code: "fixture", message: "fixture mode requested a post-sync daemon error" }); } else if (ws.data.live) { scheduleLive(ws); }}
/** * The answer to "load earlier history": the records behind a loaded window, in the same replay shape a * subscribe uses. It does not touch the live trickle — reading the past is not what starts the future. */function history(ws: Bun.ServerWebSocket<SocketData>, message: Extract<V2ClientMsg, { type: "history" }>): void { 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 behind = records.filter((record) => record.seq < message.before_seq); 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, }); for (const fixture of page) { send(ws, { type: "event", record: { ...fixture, session_id: message.session_id } }); } send(ws, { type: "sync_completed", session_id: message.session_id, latest_seq: records.at(-1)?.seq ?? 0, earliest_seq: earliest, });}
function parseClientMessage(raw: string): V2ClientMsg | null { try { const value: unknown = JSON.parse(raw); return value !== null && typeof value === "object" && "type" in value ? value as V2ClientMsg : null;50 unmodified lines return; } switch (message.type) { case "subscribe": sync(ws, message); break; case "history": history(ws, message); break; case "ping": send(ws, { type: "pong", timestamp: message.timestamp }); break; case "send_input": { const record: V2EventRecord = { seq: ws.data.nextSeq++, session_id: message.session_id, lane: "input", created_ms: Date.now(),27 unmodified lines494 unmodified linesExpand all4954964974984994995005015025035046 unmodified lines511512513514514515516516517518519520521174 unmodified linesExpand all696697698699700701702703704705706707708709710711712713577 unmodified linesExpand all494 unmodified linesExpand all | { seq: number; session_id: string; lane: "lifecycle"; payload: LifecyclePayload; created_ms: number };
// client -> daemon v2 messagesexport type V2ClientMsg = | { type: "subscribe"; session_id: string; after_seq?: number | null } | { type: "subscribe"; session_id: string; after_seq?: number | null; limit?: number | null } | { type: "history"; session_id: string; before_seq: number; limit: number } | { type: "send_input"; session_id: string; source: string; text: string; client_request_id?: string | null } | { type: "interrupt"; session_id: string; execution_id?: string | null } | { type: "reset_kernel"; session_id: string } | { type: "compact"; session_id: string }6 unmodified lines | { type: "ping"; timestamp?: number | null };
// daemon -> client v2 messagesexport type V2ServerMsg = | { type: "sync_started"; session_id: string; from_seq: number; to_seq: number } | { type: "sync_started"; session_id: string; from_seq: number; to_seq: number; earliest_seq?: number | null } | { type: "event"; record: V2EventRecord } | { type: "sync_completed"; session_id: string; latest_seq: number } | { 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 } | { type: "pong"; timestamp?: number | null } | { type: "operator_ack"; action: string; session_id?: string | null; success: boolean; message?: string | null }174 unmodified linesExpand all return window.process(record.seq, record);}
// 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`. * `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 the * replay finishes: a console that learns there is history behind its window only at the end would grow * a control at the top of a transcript the reader is already reading, and move the page under them. */export type ClientMsgV2 = V2ClientMsg;export type ServerMsgV2 = | V2ServerMsg | { type: "event"; seq: number; session_id: string; lane: Lane; payload: LanePayload; created_ms: number };577 unmodified linesExpand all1233456738 unmodified lines4647484950515253545556575859606162636465666768697071727374505152535 unmodified lines596061628863646566929394676869707172737475432 unmodified linesExpand all50850951051153553651251351453954054154251554451651751851952061 unmodified lines582583584585614615616617586587588589590591592593594595596597598599600601602603604605606607608609610611612619613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644621622623624625626627645646647648649650651630631652653654655656633634635657658659660661662663664665666667668669670671638639640641672673674675676677678679680681682683644645646647648649650684685686687688689690691692693694695696697652653698699700655656657658659660661701702703704705665666667668706707708709710711712670671672673674675676677678713714715716717718719720721722723680681682724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763180 unmodified linesExpand all9449459469478718728739489499509518758769529539549558788798808818828838849569579589599609619629638878888898908918929649659669679689699708948958968979719729739749758999009019029039049059769779789799809819829839849079089099109119129139149159169179189199209859869879889899909919929939949959969979989991000119 unmodified linesExpand allimport { describe, test, expect, beforeEach } from "bun:test";import { get } from "svelte/store";import { StreamingClientV2 } from "./streaming-v2";import { StreamingClientV2, WINDOW_LIMIT, HISTORY_PAGE } from "./streaming-v2";import type { ServerMsgV2 } from "./protocol-v2";import fixture from "../../tests/fixtures/lifecycle_diagnostics.json";
// Mock WebSocket38 unmodified lines}
(globalThis as any).WebSocket = MockWebSocket;
const mockStorage = new Map<string, string>();/** Writes per key, so a test can assert how OFTEN the cache is re-serialized, not just that it is. */const mockWrites = new Map<string, number>();const pagehideListeners: Array<() => void> = [];
(globalThis as any).window = { sessionStorage: { getItem: (k: string) => mockStorage.get(k) ?? null, setItem: (k: string, v: string) => { mockWrites.set(k, (mockWrites.get(k) ?? 0) + 1); mockStorage.set(k, v); }, removeItem: (k: string) => mockStorage.delete(k), clear: () => mockStorage.clear(), }, addEventListener: (type: string, listener: () => void) => { if (type === "pagehide") pagehideListeners.push(listener); },};
/** The page going away is the one cache write that cannot wait for a timer. */function pagehide() { for (const listener of pagehideListeners) listener();}
describe("StreamingClientV2 sequence cursor & lane algebra", () => { let client: StreamingClientV2; let ws: MockWebSocket;
5 unmodified lines await new Promise((resolve) => setTimeout(resolve, 30)); ws = MockWebSocket.instances[MockWebSocket.instances.length - 1]; });
test("subscribes on open with initial cursor", () => { 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"); // With nothing cached, the client asks for the durable history from its start rather than for a // live-only subscription. A session with records cached resumes from the newest of them instead. expect(subMsg.after_seq).toBe(0); // `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(); expect(subMsg.limit).toBe(WINDOW_LIMIT); expect(get(client.timelineStore).length).toBe(0); });
test("in-order event ingestion updates cursor and stores", () => {432 unmodified linesExpand all expect(get(client.connectionStore).isSyncing).toBe(false); expect(get(client.connectionStore).lastSeenSeq).toBe(10); });
test("page reload with empty stores does not suppress events even if cursor > 0 in storage", () => { // Set a saved cursor in sessionStorage test("a page that starts with nothing accepts whatever the daemon replays for it", () => { const sessionId = "reload-test-session"; const serverId = "local-server"; if (typeof window !== "undefined" && window.sessionStorage) { window.sessionStorage.setItem(`${serverId}:${sessionId}:cursor`, "100"); }
const reloadedClient = new StreamingClientV2("ws://127.0.0.1:8765", sessionId, serverId); // Stores are empty, so cursor must be 0 or never suppress events // Nothing is remembered between pages, so there is nothing a fresh page could suppress a replay with. expect(get(reloadedClient.timelineStore).length).toBe(0);
// Incoming events 1 and 3 arrive reloadedClient.handleServerMsg({61 unmodified lines client_request_id: reqId, }); });
test("nonempty stale cache after write failure does not leave desynced cursor", () => { const sessionId = "quota-test-" + Math.random().toString(36).slice(2, 7); const serverId = "test-server"; const cacheClient = new StreamingClientV2("ws://test", sessionId, serverId); test("nothing is remembered between loads: a reload asks the daemon again", async () => { const sessionId = "reload-" + Math.random().toString(36).slice(2, 7); const first = new StreamingClientV2("ws://test", sessionId, "reload-server"); first.handleServerMsg({ type: "event", record: { seq: 1, session_id: sessionId, lane: "input", payload: { source: "operator", text: "hello" }, created_ms: 1000 }, }); first.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 1, earliest_seq: 1 }); expect(get(first.timelineStore).length).toBe(1);
// A second client for the same session is a page that was reloaded. It holds nothing, and it says // so: the window is the daemon's to hand out, and a remembered cursor could only be ahead of the // records a fresh page does not have. const reloaded = new StreamingClientV2("ws://test", sessionId, "reload-server"); expect(get(reloaded.timelineStore).length).toBe(0); 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.limit).toBe(WINDOW_LIMIT); });
test("the daemon's floor is what makes the console offer to load more", () => { const sessionId = "floor-" + Math.random().toString(36).slice(2, 7); const floorClient = new StreamingClientV2("ws://test", sessionId); for (const seq of [100, 101, 102]) { cacheClient.handleServerMsg({ floorClient.handleServerMsg({ type: "event", record: { seq, session_id: sessionId, lane: "input", payload: { source: "operator", text: `w${seq}` }, created_ms: 1000 + seq }, }); }
// A daemon that says nothing about its history gets no load-more row: the console cannot page what // it cannot measure, and a button that does nothing is worse than no button. floorClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 102 }); expect(get(floorClient.connectionStore).hasEarlier).toBe(false);
// One that names a floor below the window gets one. floorClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 102, earliest_seq: 1 }); expect(get(floorClient.connectionStore).hasEarlier).toBe(true);
// A floor at the window means the window is the whole session. floorClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 102, earliest_seq: 100 }); expect(get(floorClient.connectionStore).hasEarlier).toBe(false); });
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);
// A replay states what it is sending and what is under it in the same breath, so the row that // offers the rest of the history is there before the reader is placed, not after. early.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 401, to_seq: 900, earliest_seq: 1 }); expect(get(early.connectionStore).isSyncing).toBe(true); expect(get(early.connectionStore).hasEarlier).toBe(true);
early.handleServerMsg({ type: "event", record: { seq: 1, session_id: sessionId, lane: "input", payload: { source: "operator", text: "event 1" }, created_ms: 1000, }, record: { seq: 401, session_id: sessionId, lane: "input", payload: { source: "operator", text: "first in the window" }, created_ms: 1401 }, }); // A floor at the window's own start is not history to fetch. early.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 401, earliest_seq: 401 }); expect(get(early.connectionStore).hasEarlier).toBe(false); });
// The cache is written when the stream pauses or the page goes away, not per event. pagehide(); test("load earlier sends a history request for the records behind the window", () => { const sessionId = "page-" + Math.random().toString(36).slice(2, 7); const pageClient = new StreamingClientV2("ws://test", sessionId, "page-server"); pageClient.connect();
// Simulate quota error on timeline key while cursor key was set to 2 if (typeof window !== "undefined" && window.sessionStorage) { window.sessionStorage.setItem(`${serverId}:${sessionId}:cursor`, "2"); // Nothing loaded, no floor known: there is nothing to ask for yet. expect(pageClient.loadEarlier()).toBe(false);
pageClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 500, to_seq: 501, earliest_seq: 1 }); for (const seq of [500, 501]) { pageClient.handleServerMsg({ type: "event", record: { seq, session_id: sessionId, lane: "input", payload: { source: "operator", text: `w${seq}` }, created_ms: 1000 + seq }, }); } // While the window is arriving, a page behind a floor that is still moving is refused. expect(pageClient.loadEarlier()).toBe(false); pageClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 501, earliest_seq: 1 }); expect(get(pageClient.connectionStore).isSyncing).toBe(false);
const restored = new StreamingClientV2("ws://test", sessionId, serverId); const restoredSeqs = get(restored.timelineStore).map((t) => t.seq); expect(restoredSeqs).toEqual([1]); expect(get(restored.connectionStore).lastSeenSeq).toBe(1); const ws = MockWebSocket.instances[MockWebSocket.instances.length - 1]; ws.readyState = MockWebSocket.OPEN; ws.sent = []; expect(pageClient.loadEarlier()).toBe(true); const req = JSON.parse(ws.sent[ws.sent.length - 1]); expect(req).toEqual({ type: "history", session_id: sessionId, before_seq: 500, limit: HISTORY_PAGE }); expect(get(pageClient.connectionStore).loadingEarlier).toBe(true);
// One page at a time: a second press while the first is in flight is refused rather than queued. expect(pageClient.loadEarlier()).toBe(false); });
test("a burst of events re-serializes the transcript once, not once per event", () => { const sessionId = "coalesce-" + Math.random().toString(36).slice(2, 7); const serverId = "coalesce-server"; const timelineKey = `${serverId}:${sessionId}:timeline`; const cursorKey = `${serverId}:${sessionId}:cursor`; const client = new StreamingClientV2("ws://test", sessionId, serverId); const events = 40; test("a history page prepends, keeps the live cursor, and a re-delivered page is not doubled", () => { const sessionId = "prepend-" + Math.random().toString(36).slice(2, 7); const histClient = new StreamingClientV2("ws://test", sessionId); for (const seq of [500, 501, 502]) { histClient.handleServerMsg({ type: "event", record: { seq, session_id: sessionId, lane: "input", payload: { source: "operator", text: `w${seq}` }, created_ms: 1000 + seq }, }); } const cursorBefore = get(histClient.connectionStore).lastSeenSeq; expect(cursorBefore).toBe(502);
// The daemon answers a history request with the same replay shape it uses for a subscribe. histClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 498, to_seq: 499 }); for (let seq = 1; seq <= events; seq++) { client.handleServerMsg({ for (const seq of [498, 499]) { histClient.handleServerMsg({ type: "event", record: { seq, session_id: sessionId, lane: "input", payload: { source: "operator", text: `burst ${seq}` }, created_ms: 1000 + seq, }, record: { seq, session_id: sessionId, lane: "input", payload: { source: "operator", text: `w${seq}` }, created_ms: 1000 + seq }, }); } histClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 502, earliest_seq: 1 });
// Every record is in the view, and the cache has not been re-serialized once per record: ingesting // a session must not cost a full serialize of the session per record. expect(get(client.timelineStore).length).toBe(events); expect(mockWrites.get(timelineKey) ?? 0).toBe(0); const seqs = get(histClient.timelineStore).map((t) => t.seq); expect(seqs).toEqual([498, 499, 500, 501, 502]); // A page behind the window is history, not an advance: the live cursor is where the live stream // left it, whatever the daemon said its latest record is. expect(get(histClient.connectionStore).lastSeenSeq).toBe(502); expect(get(histClient.connectionStore).loadingEarlier).toBe(false);
pagehide(); expect(mockWrites.get(timelineKey)).toBe(1); const cached = JSON.parse(mockStorage.get(timelineKey)!); expect(cached.length).toBe(events); expect(cached.map((t: { seq: number }) => t.seq)).toEqual( Array.from({ length: events }, (_, i) => i + 1) ); // Records and the cursor they cover are one write, so the cursor cannot claim more than the cache holds. expect(mockStorage.get(cursorKey)).toBe(String(events)); // The same page again: every record is already held, so the timeline is untouched. A retried or // overlapping page must be idempotent, and a set of seen sequences is what makes it so. for (const seq of [498, 499]) { histClient.handleServerMsg({ type: "event", record: { seq, session_id: sessionId, lane: "input", payload: { source: "operator", text: `w${seq}` }, created_ms: 1000 + seq }, }); } expect(get(histClient.timelineStore).map((t) => t.seq)).toEqual([498, 499, 500, 501, 502]); });
const reloaded = new StreamingClientV2("ws://test", sessionId, serverId); expect(get(reloaded.timelineStore).length).toBe(events); expect(get(reloaded.connectionStore).lastSeenSeq).toBe(events); test("history arriving mid-stream is never merged into the live cell it arrives beside", () => { const sessionId = "live-history-" + Math.random().toString(36).slice(2, 7); const mixClient = new StreamingClientV2("ws://test", sessionId); // a live execution streaming into its entry mixClient.handleServerMsg({ type: "event", record: { seq: 900, session_id: sessionId, lane: "cell", payload: { execution_id: "cell-live", code: "cargo test", status: "running", stream: "stdout", text: "live" }, created_ms: 9000, }, }); const live = get(mixClient.cellStore).find((e) => e.id === "cell-live"); expect(live?.status).toBe("running");
// The reader pages back, and the daemon replays a finished execution from before the window. mixClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 100, to_seq: 101 }); mixClient.handleServerMsg({ type: "event", record: { seq: 101, session_id: sessionId, lane: "cell", payload: { execution_id: "cell-old", code: "cargo fmt", status: "ok", stream: "result", text: "42", exit_code: 0 }, created_ms: 1010, }, }); mixClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 900, earliest_seq: 1 });
// The old record is its own cell, and the streaming one is exactly as it was. const ids = get(mixClient.cellStore).map((e) => e.id).sort(); expect(ids).toEqual(["cell-live", "cell-old"]); expect(get(mixClient.cellStore).find((e) => e.id === "cell-live")?.status).toBe("running"); expect(get(mixClient.timelineStore).map((t) => t.seq)).toEqual([101, 900]); });
test("lifecycle diagnostics update healthStore, workItemsStore, and candidateStore while strictly preserving lane separation", () => { const sessionId = "diag-test-" + Math.random().toString(36).slice(2, 7);180 unmodified linesExpand all expect(get(client.scratchpadStore).length).toBe(0); expect(get(client.cellStore).length).toBe(0); });
test("page reload recovers health, work items, and candidate diagnostics consistent with retained cursor", () => { const sessionId = "reload-diag-" + Math.random().toString(36).slice(2, 7); const client1 = new StreamingClientV2("ws://test", sessionId); test("a replayed window is what rebuilds the diagnostics a page lost", async () => { const sessionId = "window-diag-" + Math.random().toString(36).slice(2, 7); const windowClient = new StreamingClientV2("ws://test", sessionId);
// Ingest event with full diagnostic fixture client1.handleServerMsg({ // This is what a fresh page receives: the newest window of the session, lifecycle records included. windowClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 9, to_seq: 12 }); windowClient.handleServerMsg({ type: "event", record: { seq: 10, session_id: sessionId, lane: "lifecycle", payload: fixture.host_payload, created_ms: 1010, }, record: { seq: 10, session_id: sessionId, lane: "lifecycle", payload: fixture.host_payload, created_ms: 1010 }, }); windowClient.handleServerMsg({ type: "event", record: { seq: 12, session_id: sessionId, lane: "input", payload: { source: "operator", text: "and then?" }, created_ms: 1012 }, }); windowClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 12, earliest_seq: 9 });
// Check client1 stores expect(get(client1.healthStore).length).toBe(2); expect(get(client1.workItemsStore).length).toBe(1); expect(get(client1.candidateStore)?.candidate_id).toBe("c1"); expect(get(client1.connectionStore).lastSeenSeq).toBe(10); expect(get(client1.speechStore).length).toBe(0); const health = get(windowClient.healthStore); expect(health.length).toBe(2); expect(health[0].id).toBe("guidance"); expect(health[0].status).toBe("degraded"); expect(health[1].id).toBe("context_assembly"); expect(health[1].status).toBe("healthy");
// Simulate page reload: the page goes away (which is what makes the cache durable) and // a new client is instantiated for the same sessionId pagehide(); const reloaded = new StreamingClientV2("ws://test", sessionId); const work = get(windowClient.workItemsStore); expect(work.length).toBe(1); expect(work[0].id).toBe("repair1"); expect(work[0].status).toBe("pending");
// Reloaded client must recover diagnostics AND cursor const reloadHealth = get(reloaded.healthStore); expect(reloadHealth.length).toBe(2); expect(reloadHealth[0].id).toBe("guidance"); expect(reloadHealth[0].status).toBe("degraded"); expect(reloadHealth[1].id).toBe("context_assembly"); expect(reloadHealth[1].status).toBe("healthy"); expect(get(windowClient.candidateStore)?.candidate_id).toBe("c1"); expect(get(windowClient.candidateStore)?.epoch).toBe(2); expect(get(windowClient.connectionStore).lastSeenSeq).toBe(12); expect(get(windowClient.speechStore).length).toBe(0); // The daemon named a floor of 9 below a window that starts at 10, so there is a record behind it // and the console offers to fetch it. Sequences are sparse, so that floor is a bound rather than a // promise: a page that comes back empty is what ends the offering. expect(get(windowClient.connectionStore).hasEarlier).toBe(true);
const reloadWork = get(reloaded.workItemsStore); expect(reloadWork.length).toBe(1); expect(reloadWork[0].id).toBe("repair1"); expect(reloadWork[0].status).toBe("pending");
const reloadCand = get(reloaded.candidateStore); expect(reloadCand).not.toBeNull(); expect(reloadCand?.candidate_id).toBe("c1"); expect(reloadCand?.epoch).toBe(2);
// Cursor must be 10 expect(get(reloaded.connectionStore).lastSeenSeq).toBe(10); // Speech must be 0 expect(get(reloaded.speechStore).length).toBe(0); // The reader presses: the daemon answers with an empty page, because there is no record 9 either. // A page the console asked for that brings back nothing is the end of the history, and the offering // retires. Without that, the button would be a promise to return the same nothing forever. windowClient.connect(); await new Promise((resolve) => setTimeout(resolve, 30)); const socket = MockWebSocket.instances[MockWebSocket.instances.length - 1]; socket.readyState = MockWebSocket.OPEN; expect(windowClient.loadEarlier()).toBe(true); windowClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 9, to_seq: 9 }); windowClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 12, earliest_seq: 9 }); expect(get(windowClient.connectionStore).hasEarlier).toBe(false); expect(windowClient.loadEarlier()).toBe(false); });
test("incremental lifecycle events (capability.health, work.updated, candidate.updated) update stores cleanly", () => { const sessionId = "incremental-" + Math.random().toString(36).slice(2, 7);119 unmodified linesExpand all135 unmodified linesExpand all13613713813914014114214314414514214314414514614714814915015115215315415515615735 unmodified lines19319419519618919719819920020120220320420520620720820921021119119219319419519619721221321421527 unmodified lines2432442452462332342352362372382392472482492502512523 unmodified lines25625725825925125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636726026126226326436937037137237337437537626526626726826927027137938038138238338438538638738838939039127227327427527627727839439539639739839940040140240327940540628028128228340840941041141241341441541641728428528628728828929029129229329429529629729842042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149229930030130249749849950030330430530630730830931048 unmodified lines359360361362557363364365559560366367368564369370371372373374107 unmodified linesExpand all48248348448568048648748848968549049149249369049449549649769549849950069950170170270370470570670770870950250350450550650750850951071551151251351412 unmodified lines52752852953053153253353453553653753853954054154254354454554654754854955055155255355474355555655755855956056156256356456556656756856966 unmodified lines63663763863982164082382482582682782882964164264364464564664764864965065165265365483383465565665765865966066166266366466566666766814 unmodified lines683684685686860687688689690137 unmodified linesExpand all828829830831832833834835836837100683883984084184284384470 unmodified lines915916917918108691992092192265 unmodified lines9889899909919929931161994995996997998109 unmodified linesExpand all1108110911101111111211131280128111141115111611171118111966 unmodified lines118611871188118913561357119011911359119211931194136311951196119713671198119912001371120112021203137313741375137613771204120512061207120812091379138013811382121012111212121340 unmodified lines12541255125612571258125912601261126212632 unmodified lines1266126712681269144114421443127012711272127317 unmodified lines135 unmodified linesExpand all error?: string; reconnectAttempts: number; rejectedCount: number; lastRejection?: { seq: number; reason: string }; /** Records behind the loaded window that the daemon says it can still serve. */ hasEarlier?: boolean; /** True while a press of "load earlier history" is in flight. */ loadingEarlier?: boolean;}
/** How long the stream must be quiet before the transcript is re-serialized into its cache. */const CACHE_IDLE_MS = 400;/** A stream that never goes quiet still gets a cache write this often. */const CACHE_MAX_WAIT_MS = 3000;/** * How many records a fresh console asks for. The daemon answers with the newest window of at most this * size, anchored at its own last compaction, and says how far back its history goes. The console never * asks for a whole session: a transcript is read from its live end. */export const WINDOW_LIMIT = 2000;/** One press of "load earlier history" asks for this many records behind the window. */export const HISTORY_PAGE = 500;
export class StreamingClientV2 { private ws: WebSocket | null = null; private url: string;35 unmodified lines private activeScratchpadId: string | null = null; private activeCellId: string | null = null; private activeSpeechId: string | null = null; private pendingRequests = new Map<string, { text: string; source: string; timestamp: number }>(); private timelineSaveFailed = false; /** * Every seq the timeline holds, in either direction. A replay, a reconnect and a press of "load * earlier history" can each deliver a record this console already has, and this is what tells them * apart without walking the transcript. */ private readonly seenSeqs = new Set<number>(); /** The oldest record the daemon can still serve, when it says so. Nothing said = it cannot page. */ private earliestSeq: number | null = null; /** True between asking for a history page and its `sync_completed`. */ private backfilling = false; /** The floor the last history page asked behind, so an empty answer can be told from a full one. */ private beforeBackfillFloor = 0; /** The start of the newest range the daemon announced, until its first record lands. */ private windowStart: number | null = null;
// The cache write is the one cost in this client that grows with the session, so it is coalesced // rather than issued per event. See `scheduleCacheWrite`. private pendingCache: { items: TimelineItem[]; seq: number } | null = null; private pendingCacheSince = 0; private cacheTimer: ReturnType<typeof setTimeout> | null = null; private cacheFlushBound = false;
constructor( url: string = "ws://127.0.0.1:8765", sessionId: string = "default", serverId?: string27 unmodified lines this.diagnosticsStore = writable<DiagnosticItem[]>([]); this.operatorAckStore = writable<{ action: string; success: boolean; message: string | null; at: number } | null>(null); this.sessionStatusStore = writable<Record<string, unknown> | null>(null);
// On reload, either rehydrate the view from cache or re-request history from cursor 0 // If local stores are empty, cursor is initialized to 0 to never suppress replayed events const rehydrated = this.rehydrateFromCache(); if (!rehydrated) { this.lastSeenSeq = 0; this.updateConnection({ lastSeenSeq: 0 }); } // Nothing is remembered between loads. The window is the daemon's to hand out, and a cursor // remembered across a reload could only be ahead of the records it claims to cover. }
private deriveServerId(url: string): string { try {3 unmodified lines return url.replace(/[^a-zA-Z0-9_-]/g, "_") || "default"; } }
private get cursorKey(): string { return `${this.serverId}:${this.sessionId}:cursor`; }
private get timelineKey(): string { return `${this.serverId}:${this.sessionId}:timeline`; }
private get diagnosticsKey(): string { return `${this.serverId}:${this.sessionId}:diagnostics`; }
private saveDiagnosticsSnapshot(seq?: number) { try { if (typeof window !== "undefined" && window.sessionStorage) { const snapshot: DiagnosticSnapshot = { seq: seq ?? this.lastSeenSeq, health: get(this.healthStore), work_items: get(this.workItemsStore), candidate: get(this.candidateStore), orientation: get(this.orientationStore), target: get(this.targetStore), context: get(this.contextStore), output: get(this.outputStore), posture: get(this.postureStore), adoption: get(this.adoptionStore), saved_ms: Date.now(), }; window.sessionStorage.setItem(this.diagnosticsKey, JSON.stringify(snapshot)); } } catch { // ignore } }
private loadDiagnosticsSnapshot(maxSeq?: number): boolean { try { if (typeof window !== "undefined" && window.sessionStorage) { const raw = window.sessionStorage.getItem(this.diagnosticsKey); if (raw) { const snapshot: DiagnosticSnapshot = JSON.parse(raw); if (snapshot && typeof snapshot === "object") { if (maxSeq === undefined || snapshot.seq <= maxSeq || maxSeq === 0) { if (Array.isArray(snapshot.health)) { this.healthStore.set(snapshot.health); } if (Array.isArray(snapshot.work_items)) { this.workItemsStore.set(snapshot.work_items); } if (snapshot.candidate !== undefined) { this.candidateStore.set(snapshot.candidate); } if (snapshot.orientation !== undefined) { this.orientationStore.set(snapshot.orientation); } this.restoreLinkedSnapshot(snapshot); return true; } } } } } catch { // ignore } return false; }
private loadCursor(): number { try { if (typeof window !== "undefined" && window.sessionStorage) { const val = window.sessionStorage.getItem(this.cursorKey); if (val) { const num = parseInt(val, 10); if (!Number.isNaN(num) && num >= 0) return num; } } } catch { // sessionStorage inaccessible or private mode } return 0; }
private saveCursor(seq: number) { try { if (typeof window !== "undefined" && window.sessionStorage) { if (!this.timelineSaveFailed) { window.sessionStorage.setItem(this.cursorKey, seq.toString()); } } } catch { // ignore } }
private saveTimelineCache(items: TimelineItem[]): boolean { try { if (typeof window !== "undefined" && window.sessionStorage) { window.sessionStorage.setItem(this.timelineKey, JSON.stringify(items)); this.timelineSaveFailed = false; return true; } } catch { this.timelineSaveFailed = true; return false; } return false; }
/** * Serializing the transcript is the one cost in this client that grows with the session, and writing it * per event made ingesting a session quadratic: a replay of a few thousand records spent most of its * time re-serializing records it had already cached. So a write is coalesced — one lands once the * stream goes quiet, and a stream that never goes quiet still gets one every `CACHE_MAX_WAIT_MS`. * * The records and the cursor they cover are one record, written together in `flushCacheWrite`. The * cursor is therefore never persisted ahead of the records it claims, which is the ordering the reload * path depends on: `rehydrateFromCache` resumes from `min(savedCursor, newest cached record)`./** * The daemon is the only holder of a session's history; this console keeps a window of it and nothing * else. No cache, no remembered cursor, no second copy to keep in step: the window *is* the local * state, and asking the daemon again is what a fresh page does. */ private scheduleCacheWrite(items: TimelineItem[], seq: number) { this.pendingCache = { items, seq }; if (this.pendingCacheSince === 0) this.pendingCacheSince = Date.now(); const waited = Date.now() - this.pendingCacheSince; const delay = Math.max(0, Math.min(CACHE_IDLE_MS, CACHE_MAX_WAIT_MS - waited)); if (this.cacheTimer !== null) clearTimeout(this.cacheTimer); this.cacheTimer = setTimeout(() => this.flushCacheWrite(), delay); this.bindCacheFlush(); private windowFloor(): number { // Until the first record of a range lands, the range's announced start is the floor: the daemon // said what it is sending, and a console that waited for the records would grow the row that // offers the rest of the history over a reader who is already reading the transcript. return get(this.timelineStore)[0]?.seq ?? this.windowStart ?? 0; }
/** Write what is pending now, records before cursor, and drop it whatever the outcome. */ private flushCacheWrite() { if (this.cacheTimer !== null) { clearTimeout(this.cacheTimer); this.cacheTimer = null; } this.pendingCacheSince = 0; const pending = this.pendingCache; this.pendingCache = null; if (!pending || pending.items.length === 0) return; if (this.saveTimelineCache(pending.items)) { this.saveCursor(pending.seq); } /** What the console knows about the history behind its window, for the row that asks for more. */ private publishHistory(patch: Partial<ConnectionStatus> = {}) { const floor = this.windowFloor(); const hasEarlier = this.earliestSeq !== null && floor > this.earliestSeq; this.updateConnection({ hasEarlier, loadingEarlier: this.backfilling, ...patch }); }
/** Drop a pending write without making it: the caller is about to invalidate what it holds. */ private cancelCacheWrite() { if (this.cacheTimer !== null) { clearTimeout(this.cacheTimer); this.cacheTimer = null; } this.pendingCache = null; this.pendingCacheSince = 0; }
/** * A coalesced write has one deadline it cannot miss: the page is going away and there is no next timer. * Bound once, on the first write that is actually pending. * Ask for the records behind the loaded window. This is the reader pressing a button, not a client * catching up, so it moves neither the live cursor nor the floor: the answer is ingested as history * and the loaded window grows downward. */ private bindCacheFlush() { if (this.cacheFlushBound || typeof window === "undefined" || typeof window.addEventListener !== "function") { return; } this.cacheFlushBound = true; const flush = () => this.flushCacheWrite(); window.addEventListener("pagehide", flush); window.addEventListener("visibilitychange", () => { if (typeof document !== "undefined" && document.visibilityState === "hidden") flush(); }); public loadEarlier(limit: number = HISTORY_PAGE): boolean { if (this.backfilling) return false; // The window is still arriving: a page behind a floor that is still moving would ask for records // the console is about to receive anyway. if (get(this.connectionStore).isSyncing) return false; if (!this.ws || this.ws.readyState !== 1) return false; const floor = this.windowFloor(); if (this.earliestSeq === null || floor <= this.earliestSeq) return false; this.backfilling = true; this.beforeBackfillFloor = floor; this.updateConnection({ loadingEarlier: true }); this.send({ type: "history", session_id: this.sessionId, before_seq: floor, limit }); return true; }
private rehydrateFromCache(): boolean { try { if (typeof window !== "undefined" && window.sessionStorage) { const raw = window.sessionStorage.getItem(this.timelineKey); if (raw) { const items: TimelineItem[] = JSON.parse(raw); if (Array.isArray(items) && items.length > 0) { this.timelineStore.set(items); const sp: ScratchpadEntry[] = []; const cells: CellExecutionItem[] = []; const speech: SpeechEntry[] = []; let replayedHealth: CapabilityHealthObservation[] | null = null; let replayedWork: WorkItemDiagnostic[] | null = null; let replayedCandidate: CandidateRevisionDiagnostic | null = null; let replayedOrientation: OrientationDiagnostic | null = null; let replayedLinked: ReturnType<typeof extractLinkedDiagnostics> = {}; let candidateSeen = false; let orientationSeen = false;
for (const item of items) { if (item.lane === "scratchpad") sp.push(item.data); else if (item.lane === "cell") cells.push(item.data); else if (item.lane === "speech") speech.push(item.data); else if (item.lane === "lifecycle") { const le = item.data as LifecycleEntry; if (le.health) replayedHealth = le.health; if (le.work_items) replayedWork = le.work_items; if (le.candidate !== undefined) { replayedCandidate = le.candidate; candidateSeen = true; } if (le.orientation !== undefined) { replayedOrientation = le.orientation; orientationSeen = true; } const linked = extractLinkedDiagnostics(le.data); for (const key of ["target", "context", "output", "posture", "adoption"] as const) { if (linked[key] !== undefined) replayedLinked[key] = linked[key]; } } } this.scratchpadStore.set(sp); this.cellStore.set(cells); this.speechStore.set(speech);
const savedSeq = this.loadCursor(); const maxSeqInItems = items.reduce( (max, it) => (it.seq !== undefined ? Math.max(max, it.seq) : max), 0 ); const effectiveSeq = Math.min(savedSeq, maxSeqInItems); this.lastSeenSeq = effectiveSeq; this.saveCursor(effectiveSeq); this.updateConnection({ lastSeenSeq: effectiveSeq });
const restoredFromSnapshot = this.loadDiagnosticsSnapshot(effectiveSeq); if (!restoredFromSnapshot) { if (replayedHealth) this.healthStore.set(replayedHealth); if (replayedWork) this.workItemsStore.set(replayedWork); if (candidateSeen) this.candidateStore.set(replayedCandidate); if (orientationSeen) this.orientationStore.set(replayedOrientation); this.restoreLinkedSnapshot(replayedLinked); } return true; } } } } catch { // ignore } return false; }
public connect(url?: string, sessionId?: string) { if (url) this.url = url; if (sessionId && sessionId !== this.sessionId) { (this as any).sessionId = sessionId; const rehydrated = this.rehydrateFromCache(); if (!rehydrated) { this.lastSeenSeq = 0; } this.lastSeenSeq = 0; this.earliestSeq = null; this.seenSeqs.clear(); this.timelineStore.set([]); this.eventBuffer.clear(); this.activeScratchpadId = null; this.activeCellId = null; }48 unmodified lines reconnectAttempts: 0, error: undefined, });
// An empty local cache needs the durable history, not a live-only subscription. // 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. const storesEmpty = get(this.timelineStore).length === 0; const resumeSeq = storesEmpty ? 0 : this.lastSeenSeq;
this.send({ type: "subscribe", session_id: this.sessionId, after_seq: resumeSeq, after_seq: storesEmpty ? null : this.lastSeenSeq, ...(storesEmpty ? { limit: WINDOW_LIMIT } : {}), }); };
this.ws.onmessage = (event: MessageEvent) => {107 unmodified linesExpand all }
public setHealth(health: CapabilityHealthObservation[]) { this.healthStore.set(health.map(normalizeHealthObservation)); this.saveDiagnosticsSnapshot(); }
public setWorkItems(items: WorkItemDiagnostic[]) { this.workItemsStore.set(items.map(normalizeWorkItem)); this.saveDiagnosticsSnapshot(); }
public setCandidate(candidate: CandidateRevisionDiagnostic | null) { this.candidateStore.set(candidate ? normalizeCandidate(candidate) : null); this.saveDiagnosticsSnapshot(); }
public setOrientation(orientation: OrientationDiagnostic | null) { this.orientationStore.set(orientation ? normalizeOrientation(orientation) : null); this.saveDiagnosticsSnapshot(); }
public clearSessionState() { this.cancelCacheWrite(); this.lastSeenSeq = 0; this.saveCursor(0); try { if (typeof window !== "undefined" && window.sessionStorage) { window.sessionStorage.removeItem(this.timelineKey); window.sessionStorage.removeItem(this.diagnosticsKey); } } catch { // ignore } this.earliestSeq = null; this.backfilling = false; this.windowStart = null; this.seenSeqs.clear(); this.eventBuffer.clear(); this.activeScratchpadId = null; this.activeCellId = null; this.activeSpeechId = null; this.pendingRequests.clear(); this.timelineSaveFailed = false; this.scratchpadStore.set([]); this.cellStore.set([]); this.speechStore.set([]); this.timelineStore.set([]);12 unmodified lines
public handleServerMsg(msg: ServerMsgV2) { switch (msg.type) { case "sync_started": { // The range and the floor it sits above arrive together, so the row that offers the records // behind the window is in the first frame of the replay rather than appearing over a reader who // is already reading it. if (msg.earliest_seq !== undefined && msg.earliest_seq !== null) { this.earliestSeq = msg.earliest_seq; } this.windowStart = msg.from_seq; this.updateConnection({ isSyncing: true }); this.publishHistory(); break; }
case "sync_completed": { // Drain any buffered events up to latest watermark this.drainBuffer(); const wasBackfilling = this.backfilling; this.backfilling = false; // The daemon is the one that knows how far back its history goes. Silence here means it cannot // serve anything behind the window, and a console that guessed otherwise would offer a button // that does nothing. if (msg.earliest_seq !== undefined && msg.earliest_seq !== null) { this.earliestSeq = msg.earliest_seq; } // A window of its own is not an advance: a page behind it must never move the live cursor. if (msg.latest_seq > this.lastSeenSeq) { if (!wasBackfilling && msg.latest_seq > this.lastSeenSeq) { this.setLastSeenSeq(msg.latest_seq); } // Sequence numbers are sparse, so a floor below the window is a bound on where history could // be, not a promise that any of it exists. A page that brought back nothing is the answer: the // button offered a press that would return the same nothing every time. if (wasBackfilling && this.earliestSeq !== null && get(this.timelineStore)[0]?.seq === this.beforeBackfillFloor) { this.earliestSeq = get(this.timelineStore)[0]?.seq ?? 0; } this.updateConnection({ isSyncing: false }); this.publishHistory(); break; }
case "event": {66 unmodified lines
// Idempotent sequence cursor processor supporting sparse monotonic sequences private processSequencedRecord(record: LaneEventRecord) { const seq = record.seq; const timeline = get(this.timelineStore);
// Deduplication: drop only if local timeline already represents this sequence if (seq <= this.lastSeenSeq && timeline.some((t) => t.seq === seq)) { this.send({ type: "ack", session_id: this.sessionId, seq: this.lastSeenSeq, }); // Deduplication: drop only if the timeline already represents this sequence. Membership is kept as // a set because a replay, a reconnect and a history page all re-deliver records, and a scan of the // transcript per record is the cost this console has spent the most time removing. if (this.seenSeqs.has(seq)) { this.acknowledge(); return; }
// A record behind the loaded window is history: the reader asked for it, and it can never be an // advance. It goes into the timeline where its sequence puts it (see `updateTimelineItem`) and the // live cursor stays where it was. const floor = get(this.timelineStore)[0]?.seq; const history = floor !== undefined && seq < floor; this.applyEvent(record, history ? "history" : "live"); this.applyEvent(record); if (seq > this.lastSeenSeq) { this.seenSeqs.add(seq); if (!history && seq > this.lastSeenSeq) { this.setLastSeenSeq(seq); } this.drainBuffer(); if (history) this.publishHistory(); this.acknowledge(); }
private acknowledge() { this.send({ type: "ack", session_id: this.sessionId, seq: this.lastSeenSeq,14 unmodified lines }
private setLastSeenSeq(seq: number) { this.lastSeenSeq = seq; this.scheduleCacheWrite(get(this.timelineStore), seq); this.updateConnection({ lastSeenSeq: seq }); }
private finishActiveScratchpad() {137 unmodified linesExpand all } }
// Ingest sequenced events into individual stores & unified timeline /** * `mode` is the difference between a record the session just produced and a record the reader asked * to see from behind the window. Only the live stream has an entry a delta is currently streaming * into, so only a live record may be merged into one: history arriving while a cell streams must * never be mistaken for the cell that is streaming. */ private applyEvent(record: LaneEventRecord) { private applyEvent(record: LaneEventRecord, mode: "live" | "history" = "live") { const { seq, lane, payload, created_ms } = record; const merging = mode === "live";
switch (lane) { case "scratchpad": { const sp = payload as ScratchpadPayload;70 unmodified lines const parsed = JSON.parse(rawText); const orient = normalizeOrientation(parsed); if (orient) { this.orientationStore.set(orient); this.saveDiagnosticsSnapshot(seq); break; } } catch {} }65 unmodified lines } const truncated = Boolean(cp.truncated || (cp.truncated_bytes && cp.truncated_bytes > 0)); const command = cp.code || cp.command;
const activeCell = merging ? this.activeCellId : null; this.cellStore.update((entries) => { const idx = entries.findIndex((e) => e.id === execId || e.id === this.activeCellId); const idx = entries.findIndex((e) => e.id === execId || e.id === activeCell); let entry: CellExecutionItem;
if (idx >= 0) { const prev = entries[idx];109 unmodified linesExpand all created_ms: created_ms || Date.now(), receipt, };
const activeSpeech = merging ? this.activeSpeechId : null; this.speechStore.update((entries) => { if (this.activeSpeechId) { const activeIdx = entries.findIndex((e) => e.id === this.activeSpeechId); if (activeSpeech) { const activeIdx = entries.findIndex((e) => e.id === activeSpeech); if (activeIdx >= 0) { const prev = entries[activeIdx]; const prevId = prev.id; const updated: SpeechEntry = {66 unmodified lines const entryId = `life_${seq}`; const extracted = extractDiagnostics(lp); const linked = extractLinkedDiagnostics(lp.data);
let diagnosticsChanged = false; if (extracted.health !== undefined) {if (extracted.health !== undefined) { this.healthStore.set(extracted.health); diagnosticsChanged = true; } if (extracted.work_items !== undefined) { this.workItemsStore.set(extracted.work_items); diagnosticsChanged = true; } if (extracted.candidate !== undefined) { this.candidateStore.set(extracted.candidate); diagnosticsChanged = true; } if (extracted.orientation !== undefined) { this.orientationStore.set(extracted.orientation); diagnosticsChanged = true; } // Diagnostics live in memory for the life of the page: they are re-derived from the window on // load, so there is nothing about them worth writing down. if (linked.target !== undefined) this.targetStore.set(linked.target), diagnosticsChanged = true; if (linked.context !== undefined) this.contextStore.set(linked.context), diagnosticsChanged = true; if (linked.output !== undefined) this.outputStore.set(linked.output), diagnosticsChanged = true; if (linked.posture !== undefined) this.postureStore.set(linked.posture), diagnosticsChanged = true; if (linked.adoption !== undefined) this.adoptionStore.set(linked.adoption), diagnosticsChanged = true; if (linked.target !== undefined) this.targetStore.set(linked.target); if (linked.context !== undefined) this.contextStore.set(linked.context); if (linked.output !== undefined) this.outputStore.set(linked.output); if (linked.posture !== undefined) this.postureStore.set(linked.posture); if (linked.adoption !== undefined) this.adoptionStore.set(linked.adoption);
if (diagnosticsChanged) { this.saveDiagnosticsSnapshot(seq); }
const diagItem: DiagnosticItem = { id: entryId, seq, created_ms: created_ms || Date.now(),40 unmodified lines if (idx >= 0) { next = [...timeline]; next[idx] = item; } else { // A record with no place yet goes where its sequence says, so a history page prepends itself // and a live append lands at the end without either path having to know which it is. next = [...timeline, item]; next.sort((a, b) => { if (a.seq !== undefined && b.seq !== undefined) { return a.seq - b.seq;2 unmodified lines }); } return next; }); // An append is what makes the cache stale, and it must not be what makes the ingest slow: the // array is handed to the coalescing writer, which serializes it once the stream pauses. this.scheduleCacheWrite(next, this.lastSeenSeq); }
private updateConnection(patch: Partial<ConnectionStatus>) { this.connectionStore.update((prev) => ({ ...prev, ...patch }));17 unmodified lines