diff options
| author | Adam Malczewski <[email protected]> | 2026-06-12 19:00:29 +0900 |
|---|---|---|
| committer | Adam Malczewski <[email protected]> | 2026-06-12 19:00:29 +0900 |
| commit | d66585333ee5764700c67a81eaec015b0026f8f1 (patch) | |
| tree | 6e1ac455c2ecbf3c442fce9f73fdaed8fb71fade /src | |
| parent | 1764e3e5dff836255d121a933dd92542368346f9 (diff) | |
| download | dispatch-web-d66585333ee5764700c67a81eaec015b0026f8f1.tar.gz dispatch-web-d66585333ee5764700c67a81eaec015b0026f8f1.zip | |
feat(chat): consume CR-5 history windowing — server-windowed cold loads + show-earlier backfill
Re-pinned [email protected]>0.10.0 + [email protected]>0.6.1 (reply
frontend-history-windowing-handoff.md); re-mirrored both .dispatch references.
- HistorySync port gains optional { limit?, beforeSeq? } (CR-5 params); the
app's createHistorySync appends them to GET /conversations/:id.
- COLD-cache fresh load now fetches ?sinceSeq=0&limit=<floor(0.75xL)> — a huge
conversation no longer ships whole to show 192 chunks. A warm-cache tail sync
stays unwindowed (windowing a tail that outgrew the limit would leave a
silent seq gap behind the cache).
- hasEarlier now derives from the [email protected] CONTRACT (1-based gap-free seqs):
loaded window starting above seq 1 => older history exists — covering both
locally-trimmed AND server-windowed transcripts (the watermark stays as the
merge floor only).
- showEarlier(): local cache first; when the cache doesn't reach far enough
back, backfills the missing older run via ?beforeSeq=<oldestKnown>&limit=
and persists it (next page-in is local). latestSeq windowed-read caveat is
satisfied structurally (tail cursor derives from the cache's max seq).
- live-probe: +6 CR-5 checks (seq origin, newest-k ascending, short-chat
exactness, beforeSeq paging, 400 validation x2). NOT yet run live — backend
was down at commit time; run pending.
- backend-handoff.md: CR-5 RESOLVED, pins/mirrors current. 602 tests green x2.
Diffstat (limited to 'src')
| -rw-r--r-- | src/app/store.svelte.ts | 15 | ||||
| -rw-r--r-- | src/core/chunks/trim.test.ts | 30 | ||||
| -rw-r--r-- | src/core/chunks/trim.ts | 52 | ||||
| -rw-r--r-- | src/features/chat/index.ts | 2 | ||||
| -rw-r--r-- | src/features/chat/ports.ts | 21 | ||||
| -rw-r--r-- | src/features/chat/store.svelte.ts | 59 | ||||
| -rw-r--r-- | src/features/chat/store.test.ts | 75 | ||||
| -rw-r--r-- | src/features/chat/test-helpers.ts | 23 |
8 files changed, 218 insertions, 59 deletions
diff --git a/src/app/store.svelte.ts b/src/app/store.svelte.ts index 379805f..999f2be 100644 --- a/src/app/store.svelte.ts +++ b/src/app/store.svelte.ts @@ -25,7 +25,7 @@ import { subscribe as protocolSubscribe, unsubscribe as protocolUnsubscribe, } from "../core/protocol"; -import type { ChatStore, MetricsSync } from "../features/chat"; +import type { ChatStore, HistorySync, MetricsSync } from "../features/chat"; import { createChatStore } from "../features/chat"; import type { ConversationCache } from "../features/conversation-cache"; import { createConversationCache } from "../features/conversation-cache"; @@ -111,12 +111,13 @@ export interface CreateAppStoreOptions { localStorage?: Storage; } -function createHistorySync( - httpBase: string, - fetchImpl: typeof fetch, -): (conversationId: string, sinceSeq: number) => Promise<ConversationHistoryResponse> { - return async (conversationId: string, sinceSeq: number) => { - const url = `${httpBase}/conversations/${encodeURIComponent(conversationId)}?sinceSeq=${sinceSeq}`; +function createHistorySync(httpBase: string, fetchImpl: typeof fetch): HistorySync { + return async (conversationId, sinceSeq, window) => { + let url = `${httpBase}/conversations/${encodeURIComponent(conversationId)}?sinceSeq=${sinceSeq}`; + // CR-5 windowing ([email protected]): both must be positive + // integers when present (the server 400s otherwise; callers guarantee it). + if (window?.limit !== undefined) url += `&limit=${window.limit}`; + if (window?.beforeSeq !== undefined) url += `&beforeSeq=${window.beforeSeq}`; const res = await fetchImpl(url); if (!res.ok) { throw new Error(`History sync failed: ${res.status}`); diff --git a/src/core/chunks/trim.test.ts b/src/core/chunks/trim.test.ts index 091b646..7914f35 100644 --- a/src/core/chunks/trim.test.ts +++ b/src/core/chunks/trim.test.ts @@ -177,28 +177,43 @@ describe("restoreEarlier", () => { expect(selectHasEarlier(restored)).toBe(true); }); - it("clears the watermark when the restore exhausts known earlier history", () => { + it("restoring down to seq 1 reaches the contractual origin (hasEarlier clears)", () => { const windowed = windowTranscript(stateWith(chunks(1, 100)), 75); // hidden: 1..25 const restored = restoreEarlier(windowed, chunks(1, 100), 64); expect(restored.committed).toHaveLength(100); expect(restored.committed[0]?.seq).toBe(1); - expect(restored.hiddenBeforeSeq).toBe(0); + expect(restored.hiddenBeforeSeq).toBe(1); // floor at the origin — inert expect(restored.hiddenThinkingCount).toBe(0); expect(selectHasEarlier(restored)).toBe(false); }); - it("clears the watermark when nothing is actually below it", () => { + it("is the identity when nothing older is known locally (server may still hold more)", () => { const windowed = windowTranscript(stateWith(chunks(50, 200)), 75); const restored = restoreEarlier(windowed, [], 64); - expect(restored.hiddenBeforeSeq).toBe(0); - expect(restored.committed).toEqual(windowed.committed); + expect(restored).toBe(windowed); + // seqs are 1-based gap-free: window starts at 126 ⇒ older chunks DO exist. + expect(selectHasEarlier(restored)).toBe(true); }); - it("is the identity when nothing is hidden", () => { + it("is the identity when the window already starts at seq 1", () => { const state = stateWith(chunks(1, 10)); expect(restoreEarlier(state, chunks(1, 10), 5)).toBe(state); }); + it("works on a server-windowed transcript (no local watermark)", () => { + // A cold-cache fresh load with `?limit=` commits a suffix (seq 809..1000) + // with hiddenBeforeSeq still 0 — hasEarlier derives from seq > 1, and a + // backfilled run merges below it. + const state = stateWith(chunks(809, 1000)); + expect(state.hiddenBeforeSeq).toBe(0); + expect(selectHasEarlier(state)).toBe(true); + const restored = restoreEarlier(state, chunks(745, 808), 64); + expect(restored.committed[0]?.seq).toBe(745); + expect(restored.committed).toHaveLength(192 + 64); + expect(restored.hiddenBeforeSeq).toBe(745); + expect(selectHasEarlier(restored)).toBe(true); + }); + it("decrements the hidden thinking count by the restored thinking chunks", () => { const committed = [chunk(1, "thinking"), chunk(2), chunk(3, "thinking"), ...chunks(4, 12)]; const trimmed = trimTranscript(stateWith(committed), 10); // drops 3: seqs 1..3 (2 thinking) @@ -213,6 +228,7 @@ describe("restoreEarlier", () => { const trimmed = trimTranscript(stateWith(original), 100); const restored = restoreEarlier(trimmed, original, 1000); expect(restored.committed).toEqual(original); - expect(restored.hiddenBeforeSeq).toBe(0); + expect(restored.hiddenBeforeSeq).toBe(1); + expect(selectHasEarlier(restored)).toBe(false); }); }); diff --git a/src/core/chunks/trim.ts b/src/core/chunks/trim.ts index 1733027..94065b3 100644 --- a/src/core/chunks/trim.ts +++ b/src/core/chunks/trim.ts @@ -110,40 +110,52 @@ export function windowTranscript(state: TranscriptState, maxCommitted: number): } /** - * Page earlier (unloaded) history back in — the "Show earlier messages" action. + * The oldest LOADED seq — the start of the transcript's loaded window. Usually + * `committed[0].seq`; falls back to the watermark when a trim emptied the + * committed list (all-provisional overflow). 0 = window start unknown/origin. + */ +function oldestLoadedSeq(state: TranscriptState): number { + return state.committed[0]?.seq ?? state.hiddenBeforeSeq; +} + +/** + * Page earlier history back in — the "Show earlier messages" action. * - * `earlier` must be ALL locally-known chunks below the watermark (typically the - * full cached conversation; chunks at/above the watermark are ignored). The - * newest `count` of them are merged back in front of `committed` and the - * watermark lowers to the new oldest loaded seq — or clears to 0 when this - * restore exhausts the known earlier history (nothing left to offer). + * `earlier` is every locally-known chunk older than the loaded window + * (typically the full cached conversation, possibly extended by a CR-5 + * `?beforeSeq=` backfill; chunks at/inside the window are ignored). The newest + * `count` of them are merged back in front of `committed`, and the watermark + * follows the new window start so history merges still can't resurrect what + * remains unloaded. Identity when the window already starts at seq 1 (the + * contractual origin) or nothing older is known locally. */ export function restoreEarlier( state: TranscriptState, earlier: readonly StoredChunk[], count: number, ): TranscriptState { - if (state.hiddenBeforeSeq <= 0) return state; - const below = earlier.filter((c) => c.seq < state.hiddenBeforeSeq).sort((a, b) => a.seq - b.seq); - if (below.length === 0) { - // Nothing is actually hidden below the watermark: clear it so the - // "Show earlier" affordance disappears. - return { ...state, hiddenBeforeSeq: 0, hiddenThinkingCount: 0 }; - } + const oldest = oldestLoadedSeq(state); + if (oldest <= 1) return state; + const below = earlier.filter((c) => c.seq < oldest).sort((a, b) => a.seq - b.seq); + if (below.length === 0) return state; const keep = below.slice(-Math.max(1, count)); - const exhausted = keep.length === below.length; const firstKept = keep[0]; return { ...state, committed: [...keep, ...state.committed], - hiddenBeforeSeq: exhausted || firstKept === undefined ? 0 : firstKept.seq, - hiddenThinkingCount: exhausted - ? 0 - : Math.max(0, state.hiddenThinkingCount - countThinking(keep)), + hiddenBeforeSeq: firstKept?.seq ?? state.hiddenBeforeSeq, + hiddenThinkingCount: Math.max(0, state.hiddenThinkingCount - countThinking(keep)), }; } -/** Whether unloaded earlier history exists to offer ("Show earlier messages"). */ +/** + * Whether earlier history exists below the loaded window — drives the + * "Show earlier messages" affordance. Derived from the [email protected] CONTRACT + * that per-conversation seqs are 1-based and gap-free: a loaded window that + * starts above seq 1 means older chunks exist (locally cached or server-side), + * whether the window came from a local trim or a server-windowed (`?limit=`) + * fresh load. + */ export function selectHasEarlier(state: TranscriptState): boolean { - return state.hiddenBeforeSeq > 0; + return oldestLoadedSeq(state) > 1; } diff --git a/src/features/chat/index.ts b/src/features/chat/index.ts index 18ed693..139a64f 100644 --- a/src/features/chat/index.ts +++ b/src/features/chat/index.ts @@ -1,7 +1,7 @@ export type { RenderedChunk, RenderGroup, ToolBatchEntry } from "../../core/chunks"; export { groupRenderedChunks } from "../../core/chunks"; export type { TurnMetricsEntry } from "../../core/metrics"; -export type { ChatTransport, HistorySync, MetricsSync } from "./ports"; +export type { ChatTransport, HistorySync, HistoryWindow, MetricsSync } from "./ports"; export type { ChatStore, ChatStoreDependencies } from "./store.svelte"; export { createChatStore } from "./store.svelte"; export { default as ChatView } from "./ui/ChatView.svelte"; diff --git a/src/features/chat/ports.ts b/src/features/chat/ports.ts index e28ebf6..f8c665f 100644 --- a/src/features/chat/ports.ts +++ b/src/features/chat/ports.ts @@ -9,10 +9,29 @@ export interface ChatTransport { send(msg: ChatSendMessage): void; } -/** Injected history-sync port — fetches incremental history from the server. */ +/** + * Optional windowing for a history fetch ([email protected], CR-5). + * Both must be POSITIVE integers when present (the server 400s otherwise). + */ +export interface HistoryWindow { + /** Return only the NEWEST `limit` chunks of the selection (still ascending). */ + readonly limit?: number; + /** Exclusive upper bound: only chunks with `seq < beforeSeq` (backfill paging). */ + readonly beforeSeq?: number; +} + +/** + * Injected history-sync port — fetches incremental history from the server + * (`GET /conversations/:id?sinceSeq=&beforeSeq=&limit=`). NOTE the contract + * caveat: on a windowed/backfill read the response's `latestSeq` describes the + * returned window, not the conversation's high-water mark — never regress a + * tail cursor from it (the FE's cursor comes from the cache's max seq, which + * satisfies this naturally). + */ export type HistorySync = ( conversationId: string, sinceSeq: number, + window?: HistoryWindow, ) => Promise<ConversationHistoryResponse>; /** Injected metrics-sync port — fetches persisted per-turn metrics from the server. */ diff --git a/src/features/chat/store.svelte.ts b/src/features/chat/store.svelte.ts index 5ca28af..e74980d 100644 --- a/src/features/chat/store.svelte.ts +++ b/src/features/chat/store.svelte.ts @@ -92,10 +92,10 @@ export interface ChatStore { setModel(model: string): void; load(): Promise<void>; /** - * Page one unload-unit (`ceil(limit/4)`) of earlier history back in from the - * local cache — the "Show earlier messages" action. (When the backend ships - * CR-5 `?beforeSeq=`, this can fall through to the server once the cache is - * exhausted.) + * Page one unload-unit (`ceil(limit/4)`) of earlier history back in — the + * "Show earlier messages" action. Local cache first; when the cache doesn't + * reach far enough back (a server-windowed fresh load), the missing older + * run is fetched via CR-5 `?beforeSeq=&limit=` and persisted to the cache. */ showEarlier(): Promise<void>; /** @@ -129,12 +129,21 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { transcript = trimTranscript(transcript, chatLimit); } - async function syncTail(): Promise<void> { + /** + * Pull `seq > cache-cursor` from the server and fold it in. `coldLimit`, when + * given AND the cache is empty (a truly fresh browser), windows the fetch to + * the newest N chunks (CR-5 `?limit=`) so a huge conversation doesn't ship + * whole. It is deliberately NOT applied to a warm-cache tail: windowing a + * tail that grew past N while we were away would leave a silent seq GAP + * between the cache and the fetched window. + */ + async function syncTail(coldLimit?: number): Promise<void> { if (disposed || _pendingSync) return; _pendingSync = true; try { const since = await deps.cache.sinceSeq(deps.conversationId); - const res = await deps.historySync(deps.conversationId, since); + const window = since === 0 && coldLimit !== undefined ? { limit: coldLimit } : undefined; + const res = await deps.historySync(deps.conversationId, since, window); const merged = await deps.cache.commit(deps.conversationId, res.chunks); transcript = applyHistory(transcript, merged); maybeTrim(); @@ -227,24 +236,48 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { async load(): Promise<void> { // Fresh load shows only the newest 75% of the limit — headroom before the - // first trim. Window the cached slice SYNCHRONOUSLY with its apply (no - // render in between), and again after the tail sync (a cold cache means - // syncTail pulled the whole history in one response). + // first trim. A warm cache is windowed locally (synchronously with its + // apply — no render in between); a COLD cache passes the window to the + // server instead (CR-5 `?limit=`), so a huge conversation never ships + // whole. The post-sync window re-asserts the cap either way. const windowSize = initialWindowSize(chatLimit); const cached = await deps.cache.load(deps.conversationId); if (cached.length > 0) { transcript = windowTranscript(applyHistory(transcript, cached), windowSize); } - await syncTail(); + await syncTail(windowSize); transcript = windowTranscript(transcript, windowSize); await syncMetrics(); }, async showEarlier(): Promise<void> { if (disposed) return; - if (!selectHasEarlier(transcript)) return; - const cached = await deps.cache.load(deps.conversationId); - transcript = restoreEarlier(transcript, cached, unloadCount(chatLimit)); + const oldest = transcript.committed[0]?.seq ?? transcript.hiddenBeforeSeq; + if (oldest <= 1) return; + const want = unloadCount(chatLimit); + try { + let earlier = (await deps.cache.load(deps.conversationId)).filter((c) => c.seq < oldest); + // The local cache may not reach far enough back (a server-windowed + // fresh load cached only the window): page the missing OLDER run in + // from the server (CR-5 `?beforeSeq=&limit=`) and persist it, so the + // next page-in is local. Seqs are gap-free, so the fetched run is + // contiguous with what we hold. NOTE: the backfill response's + // `latestSeq` is a window cursor — never fed to the tail cursor + // (ours derives from the cache's max seq). + const oldestKnown = earlier[0]?.seq ?? oldest; + if (earlier.length < want && oldestKnown > 1) { + const res = await deps.historySync(deps.conversationId, 0, { + beforeSeq: oldestKnown, + limit: want - earlier.length, + }); + const merged = await deps.cache.commit(deps.conversationId, res.chunks); + earlier = merged.filter((c) => c.seq < oldest); + } + transcript = restoreEarlier(transcript, earlier, want); + _error = null; + } catch (err) { + _error = err instanceof Error ? err.message : String(err); + } }, resync(): void { diff --git a/src/features/chat/store.test.ts b/src/features/chat/store.test.ts index 5c798d6..3232009 100644 --- a/src/features/chat/store.test.ts +++ b/src/features/chat/store.test.ts @@ -1054,12 +1054,12 @@ describe("createChatStore", () => { store.dispose(); }); - it("chat limit: a cold cache (fresh browser) windows the full server history to 75%", async () => { + it("chat limit: a cold cache (fresh browser) asks the SERVER for the 75% window (CR-5 ?limit=)", async () => { const transport = createFakeTransport(); const historySync = createFakeHistorySync(); const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); - // Backend has no limit param yet (CR-5): sinceSeq=0 returns EVERYTHING. + // The server holds 500 chunks; the windowed fetch returns the newest 75. historySync.returnChunks = Array.from({ length: 500 }, (_, i) => makeStoredChunk(i + 1)); const store = createChatStore({ @@ -1073,12 +1073,77 @@ describe("createChatStore", () => { await store.load(); + // The cold-cache initial sync carried the window (`?sinceSeq=0&limit=75`). + expect(historySync.calls[0]?.sinceSeq).toBe(0); + expect(historySync.calls[0]?.window).toEqual({ limit: 75 }); + expect(store.chunks).toHaveLength(75); expect(store.chunks[0]?.seq).toBe(426); + // hasEarlier derives from the 1-based gap-free seq contract (426 > 1) — + // no local watermark was ever set. + expect(store.hasEarlier).toBe(true); + // Only the window was shipped + cached (the point of CR-5). + const cached = await cache.impl.load(CONV_ID); + expect(cached).toHaveLength(75); + + store.dispose(); + }); + + it("chat limit: a warm cache syncs the tail UNWINDOWED (no seq gap behind the cache)", async () => { + const transport = createFakeTransport(); + const historySync = createFakeHistorySync(); + const metricsSync = createFakeMetricsSync(); + const cache = createFakeCache(); + await cache.impl.commit(CONV_ID, [makeStoredChunk(1), makeStoredChunk(2)]); + historySync.returnChunks = [makeStoredChunk(3)]; + + const store = createChatStore({ + conversationId: CONV_ID, + transport: transport.impl, + historySync: historySync.impl, + metricsSync: metricsSync.impl, + cache: cache.impl, + chatLimit: 100, + }); + + await store.load(); + + expect(historySync.calls[0]?.sinceSeq).toBe(2); + expect(historySync.calls[0]?.window).toBeUndefined(); + + store.dispose(); + }); + + it("chat limit: showEarlier backfills from the server when the cache is too shallow (CR-5 ?beforeSeq=)", async () => { + const transport = createFakeTransport(); + const historySync = createFakeHistorySync(); + const metricsSync = createFakeMetricsSync(); + const cache = createFakeCache(); + historySync.returnChunks = Array.from({ length: 500 }, (_, i) => makeStoredChunk(i + 1)); + + const store = createChatStore({ + conversationId: CONV_ID, + transport: transport.impl, + historySync: historySync.impl, + metricsSync: metricsSync.impl, + cache: cache.impl, + chatLimit: 100, + }); + + await store.load(); // server-windowed: loaded + cached = 426..500 + expect(store.chunks[0]?.seq).toBe(426); + + await store.showEarlier(); + + // Nothing below 426 was cached → fetched the missing run from the server. + const backfill = historySync.calls[1]; + expect(backfill?.window).toEqual({ beforeSeq: 426, limit: 25 }); + expect(store.chunks).toHaveLength(100); + expect(store.chunks[0]?.seq).toBe(401); expect(store.hasEarlier).toBe(true); - // The full history is still CACHED locally (show-earlier pages from it). + // The backfilled run is persisted: the NEXT page-in is cache-local. const cached = await cache.impl.load(CONV_ID); - expect(cached).toHaveLength(500); + expect(cached).toHaveLength(100); store.dispose(); }); @@ -1109,6 +1174,8 @@ describe("createChatStore", () => { expect(store.chunks).toHaveLength(100); expect(store.chunks[0]?.seq).toBe(401); expect(store.hasEarlier).toBe(true); + // The cache reached deep enough — no server backfill was needed. + expect(historySync.calls).toHaveLength(1); store.dispose(); }); diff --git a/src/features/chat/test-helpers.ts b/src/features/chat/test-helpers.ts index 07dad26..6bb98a1 100644 --- a/src/features/chat/test-helpers.ts +++ b/src/features/chat/test-helpers.ts @@ -1,6 +1,6 @@ import type { StoredChunk } from "@dispatch/wire"; import type { ConversationCache } from "../conversation-cache"; -import type { ChatTransport, HistorySync, MetricsSync } from "./ports"; +import type { ChatTransport, HistorySync, HistoryWindow, MetricsSync } from "./ports"; export interface FakeTransport { readonly sent: import("@dispatch/transport-contract").ChatSendMessage[]; @@ -20,14 +20,14 @@ export function createFakeTransport(): FakeTransport { } export interface FakeHistorySync { - readonly calls: Array<{ conversationId: string; sinceSeq: number }>; + readonly calls: Array<{ conversationId: string; sinceSeq: number; window?: HistoryWindow }>; /** Set the chunks to return on the next call. */ returnChunks: readonly StoredChunk[]; readonly impl: HistorySync; } export function createFakeHistorySync(): FakeHistorySync { - const calls: Array<{ conversationId: string; sinceSeq: number }> = []; + const calls: Array<{ conversationId: string; sinceSeq: number; window?: HistoryWindow }> = []; let returnChunks: readonly StoredChunk[] = []; return { calls, @@ -37,9 +37,20 @@ export function createFakeHistorySync(): FakeHistorySync { set returnChunks(v: readonly StoredChunk[]) { returnChunks = v; }, - impl: async (conversationId, sinceSeq) => { - calls.push({ conversationId, sinceSeq }); - const chunks = returnChunks; + impl: async (conversationId, sinceSeq, window) => { + calls.push({ conversationId, sinceSeq, ...(window !== undefined ? { window } : {}) }); + // Apply the CR-5 WINDOW semantics (`beforeSeq` bound, then newest-`limit`) + // so store tests exercise the real windowed flows. `sinceSeq` filtering is + // deliberately NOT applied — tests set `returnChunks` to the slice they + // mean the server to hold past the cursor. + let chunks = returnChunks; + const before = window?.beforeSeq; + if (before !== undefined) { + chunks = chunks.filter((c) => c.seq < before); + } + if (window?.limit !== undefined && chunks.length > window.limit) { + chunks = chunks.slice(-window.limit); + } const latestSeq = chunks.length > 0 ? Math.max(...chunks.map((c) => c.seq)) : sinceSeq; return { chunks, latestSeq }; }, |
