summaryrefslogtreecommitdiffhomepage
path: root/src/features
diff options
context:
space:
mode:
Diffstat (limited to 'src/features')
-rw-r--r--src/features/chat/store.svelte.ts122
-rw-r--r--src/features/chat/store.test.ts103
-rw-r--r--src/features/chat/test-helpers.ts21
-rw-r--r--src/features/conversation-cache/cache.test.ts19
-rw-r--r--src/features/conversation-cache/cache.ts56
-rw-r--r--src/features/conversation-cache/types.ts4
-rw-r--r--src/features/layout/adapter/http.test.ts2
7 files changed, 289 insertions, 38 deletions
diff --git a/src/features/chat/store.svelte.ts b/src/features/chat/store.svelte.ts
index 01ad9f9..dab1ba2 100644
--- a/src/features/chat/store.svelte.ts
+++ b/src/features/chat/store.svelte.ts
@@ -209,7 +209,9 @@ export interface ChatStore {
* running, the server's post-subscribe replay re-asserts `generating`. The
* app store pairs this with a `chat.subscribe` for the conversation.
*/
- resync(): void;
+ resync(historyEpoch?: number): void;
+ /** Discard pre-replacement transcript state without loading an unseen conversation. */
+ invalidateHistory(historyEpoch: number): void;
dispose(): void;
}
@@ -223,6 +225,23 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
let cacheTimer = $state<CacheTimerState>(initialCacheTimerState());
const now = deps.now ?? Date.now;
let disposed = false;
+ let rerunRequested = false;
+ let historyEpoch: number | undefined;
+ let revision = 0;
+ let reloadWindow = false;
+ let earliestSeq = $state<number | null>(null);
+
+ function resetTranscript(epoch: number): void {
+ historyEpoch = epoch;
+ revision++;
+ reloadWindow = true;
+ transcript = initialState();
+ earliestSeq = null;
+ pendingEvents = [];
+ }
+ function hasEarlier(): boolean {
+ return selectHasEarlier(transcript, earliestSeq);
+ }
/** The most recent turn's id (resolves a seeded turn gap when it is stamped). */
function lastTurnIdOf(m: MetricsState): string | undefined {
@@ -313,21 +332,58 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
* 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 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();
- _error = null;
- } catch (err) {
- _error = err instanceof Error ? err.message : String(err);
- } finally {
- _pendingSync = false;
+ if (disposed) return;
+ if (_pendingSync) {
+ rerunRequested = true;
+ return;
}
+ _pendingSync = true;
+ do {
+ rerunRequested = false;
+ try {
+ const currentRevision = revision;
+ const cachedEpoch = await deps.cache.historyEpoch(deps.conversationId);
+ if (historyEpoch !== undefined && cachedEpoch !== historyEpoch)
+ await deps.cache.reset(deps.conversationId, historyEpoch);
+ const since =
+ cachedEpoch === undefined ? 0 : await deps.cache.sinceSeq(deps.conversationId);
+ const limit =
+ since === 0 ? (reloadWindow ? initialWindowSize(chatLimit) : coldLimit) : undefined;
+ const res = await deps.historySync(
+ deps.conversationId,
+ since,
+ limit === undefined ? undefined : { limit },
+ );
+ if (disposed) break;
+ if (revision !== currentRevision) {
+ rerunRequested = true;
+ continue;
+ }
+ if (
+ (cachedEpoch !== undefined && cachedEpoch !== res.historyEpoch) ||
+ (historyEpoch !== undefined && historyEpoch !== res.historyEpoch)
+ ) {
+ resetTranscript(res.historyEpoch);
+ await deps.cache.reset(deps.conversationId, res.historyEpoch);
+ if (since !== 0) {
+ rerunRequested = true;
+ continue;
+ }
+ }
+ reloadWindow = false;
+ historyEpoch = res.historyEpoch;
+ const merged = await deps.cache.commit(deps.conversationId, res.chunks, res.historyEpoch);
+ if (disposed || historyEpoch !== res.historyEpoch) continue;
+ if (since === 0 && (limit === undefined || res.chunks.length < limit))
+ earliestSeq = merged[0]?.seq ?? 0;
+ transcript = applyHistory(transcript, merged);
+ maybeTrim();
+ _error = null;
+ } catch (err) {
+ _error = err instanceof Error ? err.message : String(err);
+ }
+ } while (rerunRequested && !disposed);
+ _pendingSync = false;
}
async function syncMetrics(): Promise<void> {
@@ -351,14 +407,25 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
* newest `count` of them). Shared by `showEarlier` and the raise-refill.
*/
async function backfillOlder(oldest: number, want: number): Promise<readonly StoredChunk[]> {
+ const currentRevision = revision;
let earlier = (await deps.cache.load(deps.conversationId)).filter((c) => c.seq < oldest);
+ if (disposed || revision !== currentRevision) return [];
const oldestKnown = earlier[0]?.seq ?? oldest;
- if (earlier.length < want && oldestKnown > 1) {
+ if (earlier.length < want && (earliestSeq === null || oldestKnown > earliestSeq)) {
const res = await deps.historySync(deps.conversationId, 0, {
beforeSeq: oldestKnown,
limit: want - earlier.length,
});
- const merged = await deps.cache.commit(deps.conversationId, res.chunks);
+ if (disposed || revision !== currentRevision) return [];
+ if (res.historyEpoch !== historyEpoch) {
+ resetTranscript(res.historyEpoch);
+ await deps.cache.reset(deps.conversationId, res.historyEpoch);
+ await syncTail(initialWindowSize(chatLimit));
+ return [];
+ }
+ const merged = await deps.cache.commit(deps.conversationId, res.chunks, res.historyEpoch);
+ if (disposed || revision !== currentRevision) return [];
+ if (res.chunks.length < want - earlier.length) earliestSeq = merged[0]?.seq ?? 0;
earlier = merged.filter((c) => c.seq < oldest);
}
return earlier;
@@ -368,15 +435,15 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
* Refill toward the fresh-load window after a limit RAISE: pull older
* history (cache first, then server) so the loaded set grows to match what a
* fresh `load()` would show at the new limit. No-op when already at the
- * origin (seq 1) or already within the window. `restoreEarlier` re-derives
+ * known history boundary or already within the window. `restoreEarlier` re-derives
* the window start at apply time, so a delta landing during the await can't
* corrupt the merge. NOT gated (refilling prepends above the viewport; the
* caller preserves scroll position).
*/
async function refill(): Promise<void> {
if (disposed) return;
+ if (!hasEarlier()) return;
const oldest = transcript.committed[0]?.seq ?? transcript.hiddenBeforeSeq;
- if (oldest <= 1) return;
const want = initialWindowSize(chatLimit) - transcript.committed.length;
if (want <= 0) return;
try {
@@ -418,7 +485,7 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
return _model;
},
get hasEarlier(): boolean {
- return selectHasEarlier(transcript);
+ return hasEarlier();
},
get thinkingKeyBase(): number {
return transcript.hiddenThinkingCount;
@@ -524,8 +591,11 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
// 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 currentRevision = revision;
+ const cachedEpoch = await deps.cache.historyEpoch(deps.conversationId);
const cached = await deps.cache.load(deps.conversationId);
- if (cached.length > 0) {
+ if (!disposed && revision === currentRevision && cachedEpoch !== undefined) {
+ historyEpoch = cachedEpoch;
transcript = windowTranscript(applyHistory(transcript, cached), windowSize);
}
await syncTail(windowSize);
@@ -536,7 +606,7 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
async showEarlier(): Promise<void> {
if (disposed) return;
const oldest = transcript.committed[0]?.seq ?? transcript.hiddenBeforeSeq;
- if (oldest <= 1) return;
+ if (!hasEarlier()) return;
const want = unloadCount(chatLimit);
try {
const earlier = await backfillOlder(oldest, want);
@@ -547,8 +617,14 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
}
},
- resync(): void {
+ invalidateHistory(epoch): void {
+ if (!disposed && (historyEpoch === undefined || epoch > historyEpoch)) resetTranscript(epoch);
+ },
+
+ resync(epoch?: number): void {
if (disposed) return;
+ if (epoch !== undefined && (historyEpoch === undefined || epoch > historyEpoch))
+ resetTranscript(epoch);
// A turn may have sealed while we were disconnected (missed `turn-sealed`):
// clear the now-stale spinner BEFORE re-subscribing, so a finished turn
// doesn't spin forever. A still-running turn's replay re-asserts it.
diff --git a/src/features/chat/store.test.ts b/src/features/chat/store.test.ts
index 8604107..bb1e304 100644
--- a/src/features/chat/store.test.ts
+++ b/src/features/chat/store.test.ts
@@ -1,5 +1,8 @@
import type { AgentEvent, ImageInput, StepId, StoredChunk } from "@dispatch/wire";
+import { IDBFactory, IDBKeyRange } from "fake-indexeddb";
import { describe, expect, it, vi } from "vitest";
+import { createIdbChunkStore } from "../../adapters/idb";
+import { createConversationCache } from "../conversation-cache";
import { createChatStore } from "./store.svelte";
import {
createFakeCache,
@@ -528,7 +531,7 @@ describe("createChatStore", () => {
store.dispose();
});
- it("overlapping syncs are guarded", async () => {
+ it("overlapping syncs coalesce into a sequential rerun", async () => {
const transport = createFakeTransport();
const _historySync = createFakeHistorySync();
const metricsSync = createFakeMetricsSync();
@@ -546,7 +549,7 @@ describe("createChatStore", () => {
if (callCount === 1) {
await firstSyncPromise;
}
- return { chunks: [], latestSeq: sinceSeq };
+ return { chunks: [], latestSeq: sinceSeq, historyEpoch: 0 };
};
const store = createChatStore({
@@ -573,7 +576,7 @@ describe("createChatStore", () => {
// Release the first sync
resolveFirstSync?.();
- await new Promise((r) => setTimeout(r, 10));
+ await vi.waitFor(() => expect(callCount).toBe(2));
store.dispose();
});
@@ -1645,6 +1648,7 @@ describe("createChatStore", () => {
});
await store.load(); // only 50 chunks → all loaded, window starts at seq 1
expect(store.chunks).toHaveLength(50);
+ await store.showEarlier(); // Probe server exhaustion; seq 1 alone is not a boundary.
expect(store.hasEarlier).toBe(false);
const callsAfterLoad = historySync.calls.length;
@@ -1892,3 +1896,96 @@ describe("createChatStore — cacheTimer", () => {
busy.dispose();
});
});
+
+describe("history epochs", () => {
+ it.each([
+ "tail",
+ "backfill",
+ ])("recovers missed compaction on %s and exhausts non-1 history", async (path) => {
+ const history = createFakeHistorySync();
+ const cache = createFakeCache();
+ history.returnChunks = Array.from({ length: 20 }, (_, i) => makeStoredChunk(i + 1));
+ const store = createChatStore({
+ conversationId: CONV_ID,
+ transport: createFakeTransport().impl,
+ historySync: history.impl,
+ metricsSync: createFakeMetricsSync().impl,
+ cache: cache.impl,
+ chatLimit: 10,
+ });
+ await store.load();
+ history.historyEpoch = 1;
+ history.returnChunks = [makeStoredChunk(101), makeStoredChunk(102)];
+ if (path === "tail") {
+ store.resync();
+ await vi.waitFor(() => expect(store.chunks[0]?.seq).toBe(101));
+ } else await store.showEarlier();
+ expect(store.chunks.map((c) => c.seq)).toEqual([101, 102]);
+ expect((await cache.impl.load(CONV_ID)).map((c) => c.seq)).toEqual([101, 102]);
+ expect(history.calls.at(-1)).toMatchObject({ sinceSeq: 0, window: { limit: 7 } });
+ expect(store.hasEarlier).toBe(false);
+ const calls = history.calls.length;
+ await store.showEarlier();
+ expect(history.calls).toHaveLength(calls);
+ store.dispose();
+ });
+});
+
+it("a compaction broadcast during an in-flight fetch discards the old response", async () => {
+ const history = createFakeHistorySync();
+ history.returnChunks = [makeStoredChunk(101)];
+ let resolve!: (res: import("@dispatch/transport-contract").ConversationHistoryResponse) => void;
+ const first = new Promise<import("@dispatch/transport-contract").ConversationHistoryResponse>(
+ (r) => {
+ resolve = r;
+ },
+ );
+ const sync = vi
+ .fn<import("./ports").HistorySync>()
+ .mockImplementationOnce(() => first)
+ .mockImplementation(history.impl);
+ const cache = createFakeCache();
+ const store = createChatStore({
+ conversationId: CONV_ID,
+ transport: createFakeTransport().impl,
+ historySync: sync,
+ metricsSync: createFakeMetricsSync().impl,
+ cache: cache.impl,
+ });
+ const loading = store.load();
+ await vi.waitFor(() => expect(sync).toHaveBeenCalledTimes(1));
+ history.historyEpoch = 1;
+ store.resync(1);
+ resolve({ chunks: [makeStoredChunk(1)], latestSeq: 1, historyEpoch: 0 });
+ await loading;
+ expect(sync).toHaveBeenCalledTimes(2);
+ expect(store.chunks.map((c) => c.seq)).toEqual([101]);
+ expect((await cache.impl.load(CONV_ID)).map((c) => c.seq)).toEqual([101]);
+ store.dispose();
+});
+
+it("reloads unknown legacy cache epochs rather than merging pre-upgrade rows", async () => {
+ vi.stubGlobal("IDBKeyRange", IDBKeyRange);
+ const storage = createIdbChunkStore({ indexedDB: new IDBFactory() });
+ const cache = createConversationCache(storage);
+ await storage.append(CONV_ID, [makeStoredChunk(1)]);
+ const history = createFakeHistorySync();
+ history.returnChunks = [makeStoredChunk(101)];
+ const store = createChatStore({
+ conversationId: CONV_ID,
+ transport: createFakeTransport().impl,
+ historySync: history.impl,
+ metricsSync: createFakeMetricsSync().impl,
+ cache,
+ });
+ try {
+ await store.load();
+ expect(history.calls[0]).toMatchObject({ sinceSeq: 0 });
+ expect(store.chunks.map((c) => c.seq)).toEqual([101]);
+ expect(await cache.historyEpoch(CONV_ID)).toBe(0);
+ expect((await cache.load(CONV_ID)).map((c) => c.seq)).toEqual([101]);
+ } finally {
+ store.dispose();
+ vi.unstubAllGlobals();
+ }
+});
diff --git a/src/features/chat/test-helpers.ts b/src/features/chat/test-helpers.ts
index c99d1f4..c1b020b 100644
--- a/src/features/chat/test-helpers.ts
+++ b/src/features/chat/test-helpers.ts
@@ -43,14 +43,22 @@ export interface FakeHistorySync {
readonly calls: Array<{ conversationId: string; sinceSeq: number; window?: HistoryWindow }>;
/** Set the chunks to return on the next call. */
returnChunks: readonly StoredChunk[];
+ historyEpoch: number;
readonly impl: HistorySync;
}
export function createFakeHistorySync(): FakeHistorySync {
const calls: Array<{ conversationId: string; sinceSeq: number; window?: HistoryWindow }> = [];
let returnChunks: readonly StoredChunk[] = [];
+ let historyEpoch = 0;
return {
calls,
+ get historyEpoch() {
+ return historyEpoch;
+ },
+ set historyEpoch(v: number) {
+ historyEpoch = v;
+ },
get returnChunks() {
return returnChunks;
},
@@ -72,7 +80,7 @@ export function createFakeHistorySync(): FakeHistorySync {
chunks = chunks.slice(-window.limit);
}
const latestSeq = chunks.length > 0 ? Math.max(...chunks.map((c) => c.seq)) : sinceSeq;
- return { chunks, latestSeq };
+ return { chunks, latestSeq, historyEpoch };
},
};
}
@@ -122,13 +130,22 @@ export interface FakeCache {
export function createFakeCache(): FakeCache {
const store = new Map<string, StoredChunk[]>();
+ const epochs = new Map<string, number>();
return {
store,
impl: {
+ async historyEpoch(id) {
+ return epochs.get(id) ?? 0;
+ },
+ async reset(id, epoch) {
+ store.delete(id);
+ epochs.set(id, epoch);
+ },
async load(conversationId) {
return store.get(conversationId) ?? [];
},
- async commit(conversationId, incoming) {
+ async commit(conversationId, incoming, epoch) {
+ if (epoch !== undefined) epochs.set(conversationId, epoch);
const existing = store.get(conversationId) ?? [];
const seen = new Set(existing.map((c) => c.seq));
const toAppend = incoming.filter((c) => !seen.has(c.seq));
diff --git a/src/features/conversation-cache/cache.test.ts b/src/features/conversation-cache/cache.test.ts
index ce1c60c..4306d51 100644
--- a/src/features/conversation-cache/cache.test.ts
+++ b/src/features/conversation-cache/cache.test.ts
@@ -15,8 +15,16 @@ const chunk = (seq: number, role: "user" | "assistant" = "user"): StoredChunk =>
*/
function createFakeStore(): ConversationChunkStore {
const store = new Map<string, StoredChunk[]>();
+ const epochs = new Map<string, number>();
return {
+ async historyEpoch(id) {
+ return epochs.get(id);
+ },
+ async reset(id, epoch) {
+ store.delete(id);
+ epochs.set(id, epoch);
+ },
async load(conversationId) {
return store.get(conversationId) ?? [];
},
@@ -33,6 +41,7 @@ function createFakeStore(): ConversationChunkStore {
async delete(conversationId) {
store.delete(conversationId);
+ epochs.delete(conversationId);
},
async index() {
@@ -195,3 +204,13 @@ describe("cache.delete", () => {
expect(result).toEqual([]);
});
});
+
+it("serializes invalidation and rejects stale epoch commits", async () => {
+ const store = createFakeStore();
+ const cache = createConversationCache(store);
+ await store.append("c", [chunk(1)]); // Legacy rows have no trusted epoch.
+ expect(await cache.commit("c", [chunk(101)], 1)).toEqual([chunk(101)]);
+ await Promise.all([cache.commit("c", [chunk(102)], 1), cache.reset("c", 2)]);
+ expect(await cache.commit("c", [chunk(103)], 1)).toEqual([]);
+ expect(await store.historyEpoch("c")).toBe(2);
+});
diff --git a/src/features/conversation-cache/cache.ts b/src/features/conversation-cache/cache.ts
index 4953944..431a2a7 100644
--- a/src/features/conversation-cache/cache.ts
+++ b/src/features/conversation-cache/cache.ts
@@ -3,6 +3,9 @@ import { nextSinceSeq, reconcileCache, selectEvictions } from "./logic";
import type { ConversationChunkStore } from "./types";
export interface ConversationCache {
+ historyEpoch(conversationId: string): Promise<number | undefined>;
+ reset(conversationId: string, historyEpoch: number): Promise<void>;
+
/** Load all cached chunks for a conversation. */
load(conversationId: string): Promise<readonly StoredChunk[]>;
@@ -10,7 +13,11 @@ export interface ConversationCache {
* Load + reconcile + append new chunks.
* Returns the merged cache (the new authoritative cache for this conversation).
*/
- commit(conversationId: string, incoming: readonly StoredChunk[]): Promise<readonly StoredChunk[]>;
+ commit(
+ conversationId: string,
+ incoming: readonly StoredChunk[],
+ historyEpoch?: number,
+ ): Promise<readonly StoredChunk[]>;
/** Return the `?sinceSeq=` cursor for the next incremental sync. */
sinceSeq(conversationId: string): Promise<number>;
@@ -43,21 +50,52 @@ export function createConversationCache(
): ConversationCache {
const maxChunks = opts?.maxChunks ?? DEFAULT_MAX_CHUNKS;
+ // Serialize mutation per conversation, including invalidation, so an old fetch
+ // cannot resurrect rows after a compaction broadcast has invalidated them.
+ const pending = new Map<string, Promise<unknown>>();
+ function exclusive<T>(id: string, run: () => Promise<T>): Promise<T> {
+ const next = (pending.get(id) ?? Promise.resolve()).catch(() => {}).then(run);
+ pending.set(id, next);
+ void next
+ .finally(() => {
+ if (pending.get(id) === next) pending.delete(id);
+ })
+ .catch(() => {});
+ return next;
+ }
return {
+ historyEpoch: async (id) => {
+ await pending.get(id);
+ return store.historyEpoch(id);
+ },
+ reset: (id, epoch) =>
+ exclusive(id, async () => {
+ const current = await store.historyEpoch(id);
+ if (current === undefined || current < epoch) await store.reset(id, epoch);
+ }),
async load(conversationId) {
+ await pending.get(conversationId);
return store.load(conversationId);
},
- async commit(conversationId, incoming) {
- const cached = await store.load(conversationId);
- const { merged, toAppend } = reconcileCache(cached, incoming);
- if (toAppend.length > 0) {
- await store.append(conversationId, toAppend);
- }
- return merged;
+ commit(conversationId, incoming, epoch) {
+ return exclusive(conversationId, async () => {
+ if (epoch !== undefined) {
+ const current = await store.historyEpoch(conversationId);
+ if (current !== undefined && current > epoch) return store.load(conversationId);
+ if (current !== epoch) await store.reset(conversationId, epoch);
+ }
+ const cached = await store.load(conversationId);
+ const { merged, toAppend } = reconcileCache(cached, incoming);
+ if (toAppend.length > 0) {
+ await store.append(conversationId, toAppend);
+ }
+ return merged;
+ });
},
async sinceSeq(conversationId) {
+ await pending.get(conversationId);
const cached = await store.load(conversationId);
return nextSinceSeq(cached);
},
@@ -72,7 +110,7 @@ export function createConversationCache(
},
async delete(conversationId) {
- await store.delete(conversationId);
+ await exclusive(conversationId, () => store.delete(conversationId));
},
};
}
diff --git a/src/features/conversation-cache/types.ts b/src/features/conversation-cache/types.ts
index 345ab71..a273f34 100644
--- a/src/features/conversation-cache/types.ts
+++ b/src/features/conversation-cache/types.ts
@@ -17,6 +17,10 @@ export interface ConversationCacheIndexEntry {
* All methods MUST be idempotent on `seq`: re-appending an existing seq is a no-op.
*/
export interface ConversationChunkStore {
+ historyEpoch(conversationId: string): Promise<number | undefined>;
+ /** Atomically replace cached rows with empty history at this epoch. */
+ reset(conversationId: string, historyEpoch: number): Promise<void>;
+
/** Load all cached chunks for a conversation, seq-ordered. Returns [] if absent. */
load(conversationId: string): Promise<readonly StoredChunk[]>;
diff --git a/src/features/layout/adapter/http.test.ts b/src/features/layout/adapter/http.test.ts
index cc7d7e9..30e194a 100644
--- a/src/features/layout/adapter/http.test.ts
+++ b/src/features/layout/adapter/http.test.ts
@@ -77,7 +77,7 @@ describe("createLayoutHttp — getTree", () => {
it("a malformed 200 (SPA-fallback JSON without a tree) degrades to notFound", async () => {
const http = createLayoutHttp(
"http://backend.test",
- fetchReturning(200, { chunks: [], latestSeq: 0 }),
+ fetchReturning(200, { chunks: [], historyEpoch: 0, latestSeq: 0 }),
);
const result = await http.getTree("ws-a");
expect(result).toMatchObject({ ok: false, notFound: true });
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-25 23:46:53 +0000