diff --git a/src/app-server/query/cache.ts b/src/app-server/query/cache.ts index a40d102f..3d0b1bc1 100644 --- a/src/app-server/query/cache.ts +++ b/src/app-server/query/cache.ts @@ -55,7 +55,6 @@ interface AppServerQueryOptions { readonly staleTime: number; } -type ThreadListKind = "active" | "archived"; interface MetadataResourceSnapshot { readonly value: T; readonly probe: DiagnosticProbeResult; @@ -78,6 +77,7 @@ export class AppServerQueryCache { private activeThreadCursorKnown = false; private activeThreadCursor: string | null = null; private activeThreadRevision = 0; + private readonly threadListEpochs: Record = { active: 0, archived: 0 }; private readonly metadataResourceFetches = new Map>>(); private readonly metadataNotificationRefreshes = new Map(); private generation = 0; @@ -96,6 +96,8 @@ export class AppServerQueryCache { this.activeThreadCursorKnown = false; this.activeThreadCursor = null; this.activeThreadRevision = 0; + this.threadListEpochs.active = 0; + this.threadListEpochs.archived = 0; this.metadataResourceFetches.clear(); this.metadataNotificationRefreshes.clear(); this.client.clear(); @@ -179,6 +181,30 @@ export class AppServerQueryCache { return this.fetchArchivedThreads({ force: true }); } + applyThreadListMutations(mutations: readonly ThreadListMutation[]): void { + this.assertUsable(); + if (!appServerQueryContextIsComplete(this.context) || mutations.length === 0) return; + for (const kind of ["active", "archived"] as const) { + const relevant = mutations.filter((mutation) => mutation.list === kind); + if (relevant.length === 0) continue; + + const key = this.threadListQueryKey(kind); + const before = this.threadListSnapshot(kind); + const wasFetching = this.client.getQueryState(key)?.fetchStatus === "fetching"; + this.threadListEpochs[kind] += 1; + const after = relevant.reduce(applyThreadListMutation, before); + if (after) this.storeThreadList(kind, after); + + const partialSnapshotCreated = before === null && after !== null; + const refreshRequested = relevant.some((mutation) => mutation.kind === "refresh"); + if (before !== null || wasFetching || partialSnapshotCreated || refreshRequested) { + void this.fetchThreadList(kind, { force: true }).catch(() => { + // Query observers retain refresh failures while the event projection remains last-known-good state. + }); + } + } + } + private threadListSnapshot(kind: ThreadListKind): readonly Thread[] | null { if (!appServerQueryContextIsComplete(this.context)) return null; const threads = this.client.getQueryData(this.threadListQueryKey(kind)); @@ -357,13 +383,21 @@ export class AppServerQueryCache { queryKey: this.threadListQueryKey(kind), queryFn: async (): Promise => { if (kind === "active") { - const revision = this.activeThreadRevision; - const page = await this.runWithClient((client) => readThreadPage(client, this.context.vaultPath, { archived: false })); - if (this.activeThreadRevision !== revision) return this.activeThreadsSnapshot() ?? []; - this.rememberActiveThreadCursor(page.nextCursor); - return cloneThreads(page.threads); + for (;;) { + const epoch = this.threadListEpochs.active; + const revision = this.activeThreadRevision; + const page = await this.runWithClient((client) => readThreadPage(client, this.context.vaultPath, { archived: false })); + if (this.threadListEpochs.active !== epoch) continue; + if (this.activeThreadRevision !== revision) return this.activeThreadsSnapshot() ?? []; + this.rememberActiveThreadCursor(page.nextCursor); + return cloneThreads(page.threads); + } + } + for (;;) { + const epoch = this.threadListEpochs.archived; + const threads = await this.runWithClient((client) => listThreads(client, this.context.vaultPath, { archived: true })); + if (this.threadListEpochs.archived === epoch) return cloneThreads(threads); } - return cloneThreads(await this.runWithClient((client) => listThreads(client, this.context.vaultPath, { archived: true }))); }, staleTime: THREAD_LIST_STALE_TIME_MS, }; diff --git a/src/app-server/query/resource-store.ts b/src/app-server/query/resource-store.ts index 3f07ed4c..215b5b0d 100644 --- a/src/app-server/query/resource-store.ts +++ b/src/app-server/query/resource-store.ts @@ -13,6 +13,7 @@ import { createAppServerContextLease, } from "./keys"; import type { ObservedResultListener } from "./observed-result"; +import type { ThreadListMutation } from "./thread-list-mutation"; export interface AppServerResourceStoreOptions { cacheFactory?: (context: AppServerQueryContextIdentity) => AppServerQueryCache; @@ -122,6 +123,10 @@ export class AppServerResourceStore { return this.runForCurrentContext((cache) => cache.refreshArchivedThreads()); } + applyThreadListMutations(mutations: readonly ThreadListMutation[]): void { + this.currentCache().applyThreadListMutations(mutations); + } + observeActiveThreadsResult(listener: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void { return this.observeCurrentContext( (cache, contextListener, observeOptions) => cache.observeActiveThreadsResult(contextListener, observeOptions), diff --git a/src/app-server/query/thread-list-mutation.ts b/src/app-server/query/thread-list-mutation.ts new file mode 100644 index 00000000..2e6c1ec9 --- /dev/null +++ b/src/app-server/query/thread-list-mutation.ts @@ -0,0 +1,31 @@ +import type { Thread } from "../../domain/threads/model"; + +export type ThreadListKind = "active" | "archived"; + +export type ThreadListMutation = + | { readonly kind: "upsert"; readonly list: ThreadListKind; readonly thread: Thread } + | { readonly kind: "remove"; readonly list: ThreadListKind; readonly threadId: string } + | { + readonly kind: "update"; + readonly list: ThreadListKind; + readonly threadId: string; + readonly changes: Partial>; + } + | { readonly kind: "refresh"; readonly list: ThreadListKind }; + +export function applyThreadListMutation(snapshot: readonly Thread[] | null, mutation: ThreadListMutation): readonly Thread[] | null { + switch (mutation.kind) { + case "refresh": + return snapshot; + case "upsert": { + if (!snapshot) return [mutation.thread]; + const index = snapshot.findIndex((thread) => thread.id === mutation.thread.id); + if (index < 0) return [mutation.thread, ...snapshot]; + return snapshot.map((thread) => (thread.id === mutation.thread.id ? mutation.thread : thread)); + } + case "remove": + return snapshot?.filter((thread) => thread.id !== mutation.threadId) ?? null; + case "update": + return snapshot?.map((thread) => (thread.id === mutation.threadId ? { ...thread, ...mutation.changes } : thread)) ?? null; + } +} diff --git a/src/features/threads/catalog/thread-catalog.ts b/src/features/threads/catalog/thread-catalog.ts index 6f20e116..645fdb75 100644 --- a/src/features/threads/catalog/thread-catalog.ts +++ b/src/features/threads/catalog/thread-catalog.ts @@ -1,5 +1,6 @@ import type { AppServerQueryContextIdentity } from "../../../app-server/query/keys"; -import type { ObservedResult, ObservedResultListener } from "../../../app-server/query/observed-result"; +import type { ObservedResultListener } from "../../../app-server/query/observed-result"; +import type { ThreadListMutation } from "../../../app-server/query/thread-list-mutation"; import type { Thread } from "../../../domain/threads/model"; type ThreadListObserver = ObservedResultListener; @@ -14,35 +15,11 @@ interface ThreadCatalogStore { loadMoreActiveThreads(): Promise; refreshActiveThreads(): Promise; refreshArchivedThreads(): Promise; + applyThreadListMutations(mutations: readonly ThreadListMutation[]): void; observeActiveThreadsResult(observer: ThreadListObserver, options?: { emitCurrent?: boolean }): () => void; observeArchivedThreadsResult(observer: ThreadListObserver, options?: { emitCurrent?: boolean }): () => void; } -type ThreadCatalogEventStore = Pick< - ThreadCatalogStore, - "activeThreadsSnapshot" | "archivedThreadsSnapshot" | "refreshActiveThreads" | "refreshArchivedThreads" ->; - -interface PendingThreadUpsert { - readonly thread: Thread; - readonly acknowledgedBy: (thread: Thread) => boolean; -} - -type PendingThreadUpserts = Map; -type PendingThreadRemovals = Set; - -interface PendingThreadListFacts { - readonly upserts: PendingThreadUpserts; - readonly removals: PendingThreadRemovals; -} - -interface ThreadCatalogFacts { - readonly active: PendingThreadListFacts; - readonly archived: PendingThreadListFacts; - activeObservedResult: ObservedResult | null; - archivedObservedResult: ObservedResult | null; -} - type ThreadCatalogEventObserver = (event: ThreadCatalogEvent) => void; export interface ThreadCatalogOptions { @@ -90,313 +67,83 @@ export interface ThreadCatalog extends ThreadCatalogPaginatedActiveReader, ThreadCatalogArchivedReader, ThreadCatalogEventSink, - ThreadCatalogConnectionEventSink { - clear(): void; -} + ThreadCatalogConnectionEventSink {} export function createThreadCatalog(options: ThreadCatalogOptions): ThreadCatalog { - let factsState: { contextKey: string; facts: ThreadCatalogFacts } | null = null; - const activeObservers = new Set(); - const archivedObservers = new Set(); const { store } = options; - const currentFacts = (): ThreadCatalogFacts => { - const contextKey = store.contextKey(); - if (factsState?.contextKey === contextKey) return factsState.facts; - const facts = pendingThreadCatalogFacts(); - factsState = { contextKey, facts }; - return facts; - }; - const activeRawSnapshot = (): readonly Thread[] | null => store.activeThreadsSnapshot(); - const archivedRawSnapshot = (): readonly Thread[] | null => store.archivedThreadsSnapshot(); - const publish = (kind: ThreadListKind): void => { - const facts = currentFacts(); - const raw = kind === "active" ? activeRawSnapshot() : archivedRawSnapshot(); - const value = threadListProjection(raw, kind === "active" ? facts.active : facts.archived); - const observedResult = kind === "active" ? facts.activeObservedResult : facts.archivedObservedResult; - const result: ObservedResult = { - ...(observedResult ?? { error: null, isFetching: false }), - value, - }; - for (const observer of kind === "active" ? activeObservers : archivedObservers) observer(result); - }; - const applyToFacts = ( - eventStore: ThreadCatalogEventStore, - facts: ThreadCatalogFacts, - event: ThreadCatalogEvent, - publishChanges: boolean, - ): void => { - const changed = applyThreadCatalogEvent(eventStore, facts, event); - if (publishChanges) { - if (changed.active) publish("active"); - if (changed.archived) publish("archived"); - options.onEventApplied?.(event); - } - }; const apply = (event: ThreadCatalogEvent): void => { - applyToFacts(store, currentFacts(), event, true); + store.applyThreadListMutations(threadListMutationsForEvent(store, event)); + options.onEventApplied?.(event); }; return { apply, applyConnectionEvent: (context, event) => { - const capturedContextKey = store.contextKeyFor(context); - if (capturedContextKey !== store.contextKey()) return; - applyToFacts(store, currentFacts(), event, true); + if (store.contextKeyFor(context) !== store.contextKey()) return; + apply(event); }, - clear: () => { - factsState = null; - }, - activeSnapshot: () => threadListProjection(activeRawSnapshot(), currentFacts().active), - loadActive: () => loadThreadList(store.fetchAllActiveThreads(), currentFacts().active), - refreshActive: () => loadThreadList(store.refreshActiveThreads(), currentFacts().active), + activeSnapshot: () => store.activeThreadsSnapshot(), + loadActive: () => store.fetchAllActiveThreads(), + refreshActive: () => store.refreshActiveThreads(), hasMoreActive: () => store.hasMoreActiveThreads(), - loadMoreActive: () => loadThreadList(store.loadMoreActiveThreads(), currentFacts().active), - observeActive: (observer, observeOptions) => { - activeObservers.add(observer); - const unsubscribe = store.observeActiveThreadsResult((result) => { - const facts = currentFacts(); - facts.activeObservedResult = result; - if (result.value) acknowledgeThreadListSnapshot(facts.active, result.value); - observer({ - ...result, - value: threadListProjection(result.value, facts.active), - }); - }, observeOptions); - return () => { - activeObservers.delete(observer); - unsubscribe(); - }; - }, - archivedSnapshot: () => threadListProjection(archivedRawSnapshot(), currentFacts().archived), - refreshArchived: () => loadThreadList(store.refreshArchivedThreads(), currentFacts().archived), - observeArchived: (observer, observeOptions) => { - archivedObservers.add(observer); - const unsubscribe = store.observeArchivedThreadsResult((result) => { - const facts = currentFacts(); - facts.archivedObservedResult = result; - if (result.value) acknowledgeThreadListSnapshot(facts.archived, result.value); - observer({ - ...result, - value: threadListProjection(result.value, facts.archived), - }); - }, observeOptions); - return () => { - archivedObservers.delete(observer); - unsubscribe(); - }; - }, + loadMoreActive: () => store.loadMoreActiveThreads(), + observeActive: (observer, observeOptions) => store.observeActiveThreadsResult(observer, observeOptions), + archivedSnapshot: () => store.archivedThreadsSnapshot(), + refreshArchived: () => store.refreshArchivedThreads(), + observeArchived: (observer, observeOptions) => store.observeArchivedThreadsResult(observer, observeOptions), }; } -type ThreadListKind = "active" | "archived"; - -function pendingThreadCatalogFacts(): ThreadCatalogFacts { - return { - active: pendingThreadListFacts(), - archived: pendingThreadListFacts(), - activeObservedResult: null, - archivedObservedResult: null, - }; -} - -function applyThreadCatalogEvent( - store: ThreadCatalogEventStore, - facts: ThreadCatalogFacts, - event: ThreadCatalogEvent, -): { active: boolean; archived: boolean } { - const { active: activeFacts, archived: archivedFacts } = facts; +function threadListMutationsForEvent(store: ThreadCatalogStore, event: ThreadCatalogEvent): ThreadListMutation[] { switch (event.type) { case "thread-started": - upsertActiveThread(activeFacts, event.thread, acknowledgeByThreadId); - return { active: true, archived: false }; case "thread-forked": - upsertActiveThread(activeFacts, event.thread, acknowledgedByThreadVersion(event.thread)); - return { active: true, archived: false }; + return [{ kind: "upsert", list: "active", thread: { ...event.thread, archived: false } }]; case "thread-touched": - applyThreadTouchedEvent(store, facts, event.threadId, event.recencyAt); - return { active: true, archived: false }; + return [ + { + kind: "update", + list: "active", + threadId: event.threadId, + changes: event.recencyAt === undefined ? {} : { recencyAt: event.recencyAt }, + }, + ]; case "thread-renamed": - applyThreadRenamedEvent(store, facts, event.threadId, event.name); - return { active: true, archived: true }; - case "thread-archived": - applyThreadArchivedEvent(store, facts, event.threadId); - return { active: true, archived: true }; + return [ + { kind: "update", list: "active", threadId: event.threadId, changes: { name: event.name } }, + { kind: "update", list: "archived", threadId: event.threadId, changes: { name: event.name } }, + ]; + case "thread-archived": { + const thread = threadById(store.activeThreadsSnapshot(), event.threadId); + return [ + { kind: "remove", list: "active", threadId: event.threadId }, + ...(thread + ? [{ kind: "upsert", list: "archived", thread: { ...thread, archived: true } } satisfies ThreadListMutation] + : [{ kind: "refresh", list: "archived" } satisfies ThreadListMutation]), + ]; + } case "thread-deleted": - applyThreadDeletedEvent(activeFacts, archivedFacts, event.threadId); - return { active: true, archived: true }; + return [ + { kind: "remove", list: "active", threadId: event.threadId }, + { kind: "remove", list: "archived", threadId: event.threadId }, + ]; case "thread-restored": - applyThreadRestoredEvent(activeFacts, archivedFacts, event.thread); - return { active: true, archived: true }; - case "thread-unarchived": - applyThreadUnarchivedEvent(store, facts, event.threadId); - return { active: true, archived: true }; - } -} - -async function loadThreadList(threadsPromise: Promise, facts: PendingThreadListFacts): Promise { - const threads = await threadsPromise; - acknowledgeThreadListSnapshot(facts, threads); - return threadListProjection(threads, facts) ?? []; -} - -function upsertActiveThread(activeFacts: PendingThreadListFacts, thread: Thread, acknowledgedBy: (thread: Thread) => boolean): void { - rememberPendingThreadUpsert(activeFacts, thread, acknowledgedBy); -} - -function applyThreadTouchedEvent( - store: ThreadCatalogEventStore, - facts: ThreadCatalogFacts, - threadId: string, - recencyAt: number | null | undefined, -): void { - const activeFacts = facts.active; - const existingFact = activeFacts.upserts.get(threadId)?.thread ?? null; - let touchedThread = existingFact ? touchedActiveThread(existingFact, recencyAt) : null; - const currentThread = - threadListProjection(store.activeThreadsSnapshot(), activeFacts)?.find((thread) => thread.id === threadId) ?? touchedThread; - if (currentThread) touchedThread = touchedActiveThread(currentThread, recencyAt); - if (!touchedThread) return; - const promotedThread = touchedThread; - rememberPendingThreadUpsert( - activeFacts, - promotedThread, - recencyAt === undefined ? acknowledgeByThreadId : acknowledgedByRecency(recencyAt), - ); -} - -function applyThreadRenamedEvent(store: ThreadCatalogEventStore, facts: ThreadCatalogFacts, threadId: string, name: string | null): void { - const { active: activeFacts, archived: archivedFacts } = facts; - const activeThread = threadListProjection(store.activeThreadsSnapshot(), activeFacts)?.find((thread) => thread.id === threadId) ?? null; - if (activeThread) rememberPendingThreadUpsert(activeFacts, { ...activeThread, name }, acknowledgedByName(name)); - const archivedThread = - threadListProjection(store.archivedThreadsSnapshot(), archivedFacts)?.find((thread) => thread.id === threadId) ?? null; - if (archivedThread) rememberPendingThreadUpsert(archivedFacts, { ...archivedThread, name }, acknowledgedByName(name)); -} - -function applyThreadArchivedEvent(store: ThreadCatalogEventStore, facts: ThreadCatalogFacts, threadId: string): void { - const { active: activeFacts, archived: archivedFacts } = facts; - const archivedThread = threadListProjection(store.activeThreadsSnapshot(), activeFacts)?.find((thread) => thread.id === threadId); - rememberPendingThreadRemoval(activeFacts, threadId); - if (archivedThread) { - const pendingArchivedThread = { ...archivedThread, archived: true }; - rememberPendingThreadUpsert(archivedFacts, pendingArchivedThread, acknowledgeByThreadId); - } else { - refreshArchivedThreadsAfterUnknownArchive(store, archivedFacts); - } -} - -function applyThreadDeletedEvent(activeFacts: PendingThreadListFacts, archivedFacts: PendingThreadListFacts, threadId: string): void { - rememberPendingThreadRemoval(activeFacts, threadId); - rememberPendingThreadRemoval(archivedFacts, threadId); -} - -function applyThreadRestoredEvent(activeFacts: PendingThreadListFacts, archivedFacts: PendingThreadListFacts, thread: Thread): void { - upsertActiveThread(activeFacts, { ...thread, archived: false }, acknowledgeByThreadId); - rememberPendingThreadRemoval(archivedFacts, thread.id); -} - -function applyThreadUnarchivedEvent(store: ThreadCatalogEventStore, facts: ThreadCatalogFacts, threadId: string): void { - const { active: activeFacts, archived: archivedFacts } = facts; - const restoredThread = - threadListProjection(store.archivedThreadsSnapshot(), archivedFacts)?.find((thread) => thread.id === threadId) ?? null; - rememberPendingThreadRemoval(archivedFacts, threadId); - if (restoredThread) { - upsertActiveThread(activeFacts, { ...restoredThread, archived: false }, acknowledgeByThreadId); - return; - } - refreshThreadListsAfterUnknownUnarchive(store, activeFacts, archivedFacts); -} - -function threadListProjection(snapshot: readonly Thread[] | null, facts: PendingThreadListFacts): readonly Thread[] | null { - if (!snapshot && facts.upserts.size === 0 && facts.removals.size === 0) return null; - const threads = (snapshot ?? []) - .filter((thread) => !facts.removals.has(thread.id)) - .map((thread) => facts.upserts.get(thread.id)?.thread ?? thread); - const snapshotThreadIds = new Set(threads.map((thread) => thread.id)); - const pendingThreads = Array.from(facts.upserts.values()) - .map((entry) => entry.thread) - .filter((thread) => !facts.removals.has(thread.id) && !snapshotThreadIds.has(thread.id)); - return pendingThreads.length === 0 ? threads : [...pendingThreads.reverse(), ...threads]; -} - -function pendingThreadListFacts(): PendingThreadListFacts { - return { - upserts: new Map(), - removals: new Set(), - }; -} - -function acknowledgeThreadListSnapshot(facts: PendingThreadListFacts, threads: readonly Thread[]): void { - const threadIds = new Set(threads.map((thread) => thread.id)); - for (const thread of threads) { - const upsert = facts.upserts.get(thread.id); - if (upsert?.acknowledgedBy(thread)) facts.upserts.delete(thread.id); - } - for (const threadId of [...facts.removals]) { - if (!threadIds.has(threadId)) facts.removals.delete(threadId); - } -} - -function rememberPendingThreadUpsert(facts: PendingThreadListFacts, thread: Thread, acknowledgedBy: (thread: Thread) => boolean): void { - facts.removals.delete(thread.id); - facts.upserts.delete(thread.id); - facts.upserts.set(thread.id, { thread, acknowledgedBy }); -} - -function rememberPendingThreadRemoval(facts: PendingThreadListFacts, threadId: string): void { - facts.upserts.delete(threadId); - facts.removals.add(threadId); -} - -function touchedActiveThread(thread: Thread, recencyAt: number | null | undefined): Thread { - return recencyAt === undefined ? thread : { ...thread, recencyAt }; -} - -function acknowledgeByThreadId(): boolean { - return true; -} - -function acknowledgedByName(name: string | null): (thread: Thread) => boolean { - return (thread) => thread.name === name; -} - -function acknowledgedByThreadVersion(reference: Thread): (thread: Thread) => boolean { - return (thread) => - thread.updatedAt > reference.updatedAt || - (thread.updatedAt === reference.updatedAt && - thread.preview === reference.preview && - thread.name === reference.name && - thread.recencyAt === reference.recencyAt); -} - -function acknowledgedByRecency(recencyAt: number | null): (thread: Thread) => boolean { - return (thread) => thread.recencyAt === recencyAt; -} - -function refreshArchivedThreadsAfterUnknownArchive(store: ThreadCatalogEventStore, archivedFacts: PendingThreadListFacts): void { - // A force refresh can join an older in-flight archived request. Run one more - // refresh afterward so an archive recorded during that request is not lost. - void refreshArchivedThreadsTwice(store, archivedFacts); -} - -async function refreshArchivedThreadsTwice(store: ThreadCatalogEventStore, archivedFacts: PendingThreadListFacts): Promise { - for (let attempt = 0; attempt < 2; attempt += 1) { - try { - acknowledgeThreadListSnapshot(archivedFacts, await store.refreshArchivedThreads()); - } catch { - // The query observer retains the failure for diagnostics; continue so a joined stale request gets one fresh retry. + return [ + { kind: "upsert", list: "active", thread: { ...event.thread, archived: false } }, + { kind: "remove", list: "archived", threadId: event.thread.id }, + ]; + case "thread-unarchived": { + const thread = threadById(store.archivedThreadsSnapshot(), event.threadId); + return [ + { kind: "remove", list: "archived", threadId: event.threadId }, + ...(thread + ? [{ kind: "upsert", list: "active", thread: { ...thread, archived: false } } satisfies ThreadListMutation] + : [{ kind: "refresh", list: "active" } satisfies ThreadListMutation]), + ]; } } } -function refreshThreadListsAfterUnknownUnarchive( - store: ThreadCatalogEventStore, - activeFacts: PendingThreadListFacts, - archivedFacts: PendingThreadListFacts, -): void { - // Unknown unarchives need both lists: active gains the thread, archived loses it. - void Promise.allSettled([store.refreshActiveThreads(), store.refreshArchivedThreads()]).then(([active, archived]) => { - if (active.status === "fulfilled") acknowledgeThreadListSnapshot(activeFacts, active.value); - if (archived.status === "fulfilled") acknowledgeThreadListSnapshot(archivedFacts, archived.value); - }); +function threadById(threads: readonly Thread[] | null, threadId: string): Thread | null { + return threads?.find((thread) => thread.id === threadId) ?? null; } diff --git a/src/plugin-runtime.ts b/src/plugin-runtime.ts index 268734bb..76299176 100644 --- a/src/plugin-runtime.ts +++ b/src/plugin-runtime.ts @@ -97,7 +97,6 @@ export class CodexPanelRuntime implements AppServerClientAccess { this.selectionRewriteController = null; this.panels.reset(); this.appServerResourceStore.reset(); - this.threadCatalog.clear(); } activeWorkspaceLeafChanged(leaf: Parameters[0]): void { diff --git a/tests/app-server/query-cache.test.ts b/tests/app-server/query-cache.test.ts index 1782d7d3..7892dc87 100644 --- a/tests/app-server/query-cache.test.ts +++ b/tests/app-server/query-cache.test.ts @@ -4,6 +4,7 @@ import { AppServerQueryCache } from "../../src/app-server/query/cache"; import type { AppServerQueryContext, AppServerQueryContextIdentity } from "../../src/app-server/query/keys"; import type { RateLimitSnapshot } from "../../src/domain/runtime/metrics"; import type { RuntimePermissionProfileSummary } from "../../src/domain/runtime/permissions"; +import type { Thread } from "../../src/domain/threads/model"; describe("AppServerQueryCache", () => { it("garbage-collects inactive query records", async () => { @@ -124,6 +125,33 @@ describe("AppServerQueryCache", () => { expect(cache.hasMoreActiveThreads()).toBe(true); }); + it("rejects only a thread read started before an event and accepts the following server value", async () => { + const staleRead = deferred<{ data: ReturnType[]; nextCursor: null }>(); + const postEventRead = deferred<{ data: ReturnType[]; nextCursor: null }>(); + const listThreads = vi + .fn() + .mockResolvedValueOnce({ data: [{ ...thread("target"), name: "initial" }], nextCursor: null }) + .mockImplementationOnce(() => staleRead.promise) + .mockImplementationOnce(() => postEventRead.promise); + const cache = cacheWithRequestHandlers({ "thread/list": listThreads }); + await cache.refreshActiveThreads(); + + const refresh = cache.refreshActiveThreads(); + await flushMicrotasks(); + cache.applyThreadListMutations([{ kind: "update", list: "active", threadId: "target", changes: { name: "from-event" } }]); + + expect(cache.activeThreadsSnapshot()?.[0]?.name).toBe("from-event"); + staleRead.resolve({ data: [{ ...thread("target"), name: "stale" }], nextCursor: null }); + await vi.waitFor(() => expect(listThreads).toHaveBeenCalledTimes(3)); + expect(cache.activeThreadsSnapshot()?.[0]?.name).toBe("from-event"); + + postEventRead.resolve({ data: [{ ...thread("target"), name: "authoritative" }], nextCursor: null }); + await refresh; + + expect(cache.activeThreadsSnapshot()?.[0]?.name).toBe("authoritative"); + expect(listThreads).toHaveBeenCalledTimes(3); + }); + it("retries a full active-thread inventory when a first-page refresh wins the race", async () => { const oldInventory = deferred<{ data: ReturnType[]; nextCursor: string | null }>(); const listThreads = vi @@ -231,6 +259,48 @@ describe("AppServerQueryCache", () => { unsubscribe(); }); + it("publishes metadata resources only after a retry settles", async () => { + const retry = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); + const nextRetry = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); + const listSkills = vi + .fn() + .mockRejectedValueOnce(new Error("skills offline")) + .mockImplementationOnce(() => retry.promise) + .mockImplementationOnce(() => nextRetry.promise); + const cache = cacheWithRequestHandlers({ "skills/list": listSkills }); + await cache.refreshSkills(); + const listener = vi.fn(); + const unsubscribe = cache.observeAppServerMetadataResources(listener, { emitCurrent: false }); + + const refresh = cache.refreshSkills(); + await flushMicrotasks(); + + expect(listener).not.toHaveBeenCalled(); + retry.resolve({ data: [{ skills: [catalogSkill("writer")] }] }); + await refresh; + + expect(listener).toHaveBeenCalledOnce(); + expect(listener).toHaveBeenCalledWith({ + id: "skills", + value: [expect.objectContaining({ name: "writer" })], + probe: expect.objectContaining({ id: "skills", status: "ok" }), + }); + + const nextRefresh = cache.refreshSkills(); + await flushMicrotasks(); + expect(listener).toHaveBeenCalledOnce(); + nextRetry.resolve({ data: [{ skills: [catalogSkill("editor")] }] }); + await nextRefresh; + + expect(listener).toHaveBeenCalledTimes(2); + expect(listener).toHaveBeenLastCalledWith({ + id: "skills", + value: [expect.objectContaining({ name: "editor" })], + probe: expect.objectContaining({ id: "skills", status: "ok" }), + }); + unsubscribe(); + }); + it("coalesces a burst of skills notifications into one forced trailing refresh", async () => { const first = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); const second = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); @@ -601,7 +671,7 @@ function appServerRateLimit(usedPercent: number): RateLimitSnapshot { }; } -function thread(id: string, archived = false) { +function thread(id: string, archived = false): Thread { return { id, preview: "", diff --git a/tests/features/threads/catalog/thread-catalog.test.ts b/tests/features/threads/catalog/thread-catalog.test.ts index 0eab9182..bf52e344 100644 --- a/tests/features/threads/catalog/thread-catalog.test.ts +++ b/tests/features/threads/catalog/thread-catalog.test.ts @@ -1,585 +1,154 @@ -import { describe, expect, it, type Mock, vi } from "vitest"; +import { describe, expect, it, vi } from "vitest"; -import type { AppServerQueryClientRunner } from "../../../../src/app-server/query/cache"; -import type { AppServerQueryContext, AppServerQueryContextIdentity } from "../../../../src/app-server/query/keys"; -import type { ObservedResult, ObservedResultListener } from "../../../../src/app-server/query/observed-result"; -import { AppServerResourceStore } from "../../../../src/app-server/query/resource-store"; +import type { AppServerQueryContextIdentity } from "../../../../src/app-server/query/keys"; +import type { ObservedResultListener } from "../../../../src/app-server/query/observed-result"; +import { applyThreadListMutation, type ThreadListMutation } from "../../../../src/app-server/query/thread-list-mutation"; import type { Thread } from "../../../../src/domain/threads/model"; -import { createThreadCatalog, type ThreadCatalog } from "../../../../src/features/threads/catalog/thread-catalog"; +import { createThreadCatalog } from "../../../../src/features/threads/catalog/thread-catalog"; describe("ThreadCatalog", () => { - it("applies rename mutations after updating the catalog cache", () => { - const { catalog } = catalogFixture(); - receiveActive(catalog, [thread("thread"), thread("other")]); - - catalog.apply({ type: "thread-renamed", threadId: "thread", name: "Renamed" }); - - expect(catalog.activeSnapshot()).toEqual([{ ...thread("thread"), name: "Renamed" }, thread("other")]); - }); - - it("applies archive mutations after updating catalog membership", () => { - const { catalog } = catalogFixture(); - const listener = vi.fn(); - const archivedListener = vi.fn(); - catalog.observeActive(listener); - catalog.observeArchived(archivedListener); - receiveActive(catalog, [thread("thread"), thread("other")]); - receiveArchived(catalog, [thread("archived", true)]); - - catalog.apply({ type: "thread-archived", threadId: "thread" }); - - expect(catalog.activeSnapshot()).toEqual([thread("other")]); - expect(catalog.archivedSnapshot()).toEqual([{ ...thread("thread"), archived: true }, thread("archived", true)]); - expect(listener).toHaveBeenLastCalledWith(expect.objectContaining({ value: [thread("other")] })); - expect(archivedListener).toHaveBeenLastCalledWith( - expect.objectContaining({ value: [{ ...thread("thread"), archived: true }, thread("archived", true)] }), - ); - }); - - it("refreshes archived snapshots after unknown archive mutations even when an older archived refresh is in flight", async () => { - const staleArchivedRefresh = deferred(); - const secondArchivedRefreshStarted = deferred(); - let archivedRefreshCount = 0; - const fetchThreads = vi.fn((_context: { codexPath: string; vaultPath: string }, archived: boolean) => { - if (!archived) return Promise.resolve([]); - archivedRefreshCount += 1; - if (archivedRefreshCount === 1) return staleArchivedRefresh.promise; - secondArchivedRefreshStarted.resolve(undefined); - return Promise.resolve([thread("thread", true)]); + it("projects rename and archive events through the shared query owner", () => { + const store = catalogStore({ + active: [thread("active"), thread("other")], + archived: [thread("archived", true)], }); - const { catalog } = catalogFixture({ fetchThreads }); - - const staleRefresh = catalog.refreshArchived(); - catalog.apply({ type: "thread-archived", threadId: "thread" }); - - staleArchivedRefresh.resolve([thread("old", true)]); - await staleRefresh; - await secondArchivedRefreshStarted.promise; - - await vi.waitFor(() => { - expect(catalog.archivedSnapshot()).toEqual([thread("thread", true)]); - }); - expect(fetchThreads).toHaveBeenCalledTimes(2); - }); - - it("applies known delete mutations to cache", () => { - const { catalog } = catalogFixture(); - receiveActive(catalog, [thread("thread"), thread("other")]); - receiveArchived(catalog, [thread("thread", true), thread("archived", true)]); - - catalog.apply({ type: "thread-deleted", threadId: "thread" }); - - expect(catalog.activeSnapshot()).toEqual([thread("other")]); - expect(catalog.archivedSnapshot()).toEqual([thread("archived", true)]); - }); - - it("records started, forked, and restored thread membership", () => { - const { catalog } = catalogFixture(); - receiveActive(catalog, [thread("existing")]); - receiveArchived(catalog, [thread("restored", true), thread("archived", true)]); - - catalog.apply({ type: "thread-started", thread: thread("started") }); - catalog.apply({ type: "thread-forked", thread: thread("forked") }); - catalog.apply({ type: "thread-restored", thread: thread("restored") }); - - expect(catalog.activeSnapshot()?.map((item) => item.id)).toEqual(["restored", "forked", "started", "existing"]); - expect(catalog.archivedSnapshot()?.map((item) => item.id)).toEqual(["archived"]); - }); - - it("keeps app-server lifecycle threads visible until the server list acknowledges them", async () => { - const fetchThreads = vi - .fn() - .mockResolvedValueOnce([thread("other")]) - .mockResolvedValueOnce([thread("started"), thread("other")]) - .mockResolvedValueOnce([thread("other")]); - const { catalog } = catalogFixture({ fetchThreads }); - const listener = vi.fn(); - catalog.observeActive(listener); - - catalog.apply({ type: "thread-started", thread: thread("started") }); - - await expect(catalog.refreshActive()).resolves.toEqual([thread("started"), thread("other")]); - expect(catalog.activeSnapshot()).toEqual([thread("started"), thread("other")]); - expect(observedActiveThreadIds(listener)).not.toContainEqual(["other"]); - - await expect(catalog.refreshActive()).resolves.toEqual([thread("started"), thread("other")]); - await expect(catalog.refreshActive()).resolves.toEqual([thread("other")]); - expect(catalog.activeSnapshot()).toEqual([thread("other")]); - }); - - it("starts a fresh lifecycle overlay for every resource lease", () => { - const { catalog, replaceContext } = catalogFixture({ context: () => ({ codexPath: "codex-a", vaultPath: "/vault" }) }); - catalog.apply({ type: "thread-started", thread: thread("started-a1") }); - expect(catalog.activeSnapshot()).toEqual([thread("started-a1")]); - - replaceContext({ codexPath: "codex-b", vaultPath: "/vault" }); - expect(catalog.activeSnapshot()).toBeNull(); - replaceContext({ codexPath: "codex-a", vaultPath: "/vault" }); - - expect(catalog.activeSnapshot()).toBeNull(); - }); - - it("does not publish connection events captured by an earlier lease", () => { - const { catalog, contextIdentity, replaceContext } = catalogFixture({ - context: () => ({ codexPath: "codex-a", vaultPath: "/vault" }), - }); - const sourceA1 = contextIdentity(); - - replaceContext({ codexPath: "codex-b", vaultPath: "/vault" }); - catalog.applyConnectionEvent(sourceA1, { type: "thread-started", thread: thread("stale-a1") }); - replaceContext({ codexPath: "codex-a", vaultPath: "/vault" }); - - expect(catalog.activeSnapshot()).toBeNull(); - }); - - it("preserves raw query status when lifecycle overlays publish", async () => { - const refresh = deferred(); - const { catalog } = catalogFixture({ fetchThreads: () => refresh.promise }); - const listener = vi.fn(); - catalog.observeActive(listener); - const refreshing = catalog.refreshActive(); - await flushMicrotasks(); - - catalog.apply({ type: "thread-started", thread: thread("started") }); - - expect(listener).toHaveBeenLastCalledWith({ value: [thread("started")], error: null, isFetching: true }); - refresh.resolve([]); - await refreshing; - }); - - it("keeps rollback fork metadata until active snapshots catch up to the rollback version", () => { - const { catalog } = catalogFixture(); - const forkBeforeRollback = thread("forked", false, { name: "Before", preview: "Before rollback", updatedAt: 20 }); - const forkAfterRollback = thread("forked", false, { name: "After", preview: "After rollback", updatedAt: 20 }); - const forkAfterFutureUpdate = thread("forked", false, { name: "Future", preview: "Future update", updatedAt: 21 }); - receiveActive(catalog, [thread("existing")]); - - catalog.apply({ type: "thread-forked", thread: forkAfterRollback }); - receiveActive(catalog, [forkBeforeRollback, thread("existing")]); - - expect(catalog.activeSnapshot()).toEqual([forkAfterRollback, thread("existing")]); - - receiveActive(catalog, [forkAfterFutureUpdate, thread("existing")]); - - expect(catalog.activeSnapshot()).toEqual([forkAfterFutureUpdate, thread("existing")]); - }); - - it("keeps app-server rename facts when an older active list resolves later", async () => { - const staleRefresh = deferred(); - const fetchThreads = vi - .fn() - .mockReturnValueOnce(staleRefresh.promise) - .mockResolvedValueOnce([{ ...thread("thread"), name: "Renamed" }, thread("other")]); - const { catalog } = catalogFixture({ fetchThreads }); - receiveActive(catalog, [thread("thread"), thread("other")]); - - const refresh = catalog.refreshActive(); - await flushMicrotasks(); - catalog.apply({ type: "thread-renamed", threadId: "thread", name: "Renamed" }); - staleRefresh.resolve([thread("thread"), thread("other")]); - - await expect(refresh).resolves.toEqual([{ ...thread("thread"), name: "Renamed" }, thread("other")]); - expect(catalog.activeSnapshot()).toEqual([{ ...thread("thread"), name: "Renamed" }, thread("other")]); - - await expect(catalog.refreshActive()).resolves.toEqual([{ ...thread("thread"), name: "Renamed" }, thread("other")]); - expect(catalog.activeSnapshot()).toEqual([{ ...thread("thread"), name: "Renamed" }, thread("other")]); - }); - - it("keeps app-server removal facts when older active and archived lists resolve later", async () => { - const staleActiveRefresh = deferred(); - const staleArchivedRefresh = deferred(); - const fetchThreads = vi.fn((_context: { codexPath: string; vaultPath: string }, archived: boolean) => - archived ? staleArchivedRefresh.promise : staleActiveRefresh.promise, - ); - const { catalog } = catalogFixture({ fetchThreads }); - receiveActive(catalog, [thread("active"), thread("other")]); - receiveArchived(catalog, [thread("archived", true), thread("kept", true)]); - - const activeRefresh = catalog.refreshActive(); - const archivedRefresh = catalog.refreshArchived(); - await flushMicrotasks(); + const onEventApplied = vi.fn(); + const catalog = createThreadCatalog({ store, onEventApplied }); + catalog.apply({ type: "thread-renamed", threadId: "active", name: "Renamed" }); catalog.apply({ type: "thread-archived", threadId: "active" }); - catalog.apply({ type: "thread-deleted", threadId: "archived" }); - staleActiveRefresh.resolve([thread("active"), thread("other")]); - staleArchivedRefresh.resolve([thread("archived", true), thread("kept", true)]); - await expect(activeRefresh).resolves.toEqual([thread("other")]); - await expect(archivedRefresh).resolves.toEqual([thread("active", true), thread("kept", true)]); expect(catalog.activeSnapshot()).toEqual([thread("other")]); - expect(catalog.archivedSnapshot()).toEqual([thread("active", true), thread("kept", true)]); + expect(catalog.archivedSnapshot()).toEqual([{ ...thread("active"), name: "Renamed", archived: true }, thread("archived", true)]); + expect(onEventApplied).toHaveBeenCalledTimes(2); }); - it("archives unacknowledged active lifecycle threads without waiting for list acknowledgement", async () => { - const fetchThreads = vi.fn().mockResolvedValue([thread("other")]); - const { catalog } = catalogFixture({ fetchThreads }); - const listener = vi.fn(); - const archivedListener = vi.fn(); - catalog.observeActive(listener); - catalog.observeArchived(archivedListener); + it("requests an authoritative archived read when an archive event lacks a source snapshot", () => { + const store = catalogStore(); + const catalog = createThreadCatalog({ store }); - catalog.apply({ type: "thread-started", thread: thread("started") }); - catalog.apply({ type: "thread-archived", threadId: "started" }); + catalog.apply({ type: "thread-archived", threadId: "unknown" }); - expect(catalog.activeSnapshot()).toEqual([]); - expect(catalog.archivedSnapshot()).toEqual([thread("started", true)]); - expect(archivedListener).toHaveBeenLastCalledWith(expect.objectContaining({ value: [thread("started", true)] })); - - await expect(catalog.refreshActive()).resolves.toEqual([thread("other")]); - expect(catalog.activeSnapshot()).toEqual([thread("other")]); - expect(observedActiveThreadIds(listener)).not.toContainEqual(["started", "other"]); - }); - - it("deletes unacknowledged active lifecycle threads", async () => { - const fetchThreads = vi.fn().mockResolvedValue([thread("other")]); - const { catalog } = catalogFixture({ fetchThreads }); - catalog.apply({ type: "thread-started", thread: thread("started") }); - - catalog.apply({ type: "thread-deleted", threadId: "started" }); - - expect(catalog.activeSnapshot()).toEqual([]); - expect(catalog.archivedSnapshot()).toEqual([]); - await expect(catalog.refreshActive()).resolves.toEqual([thread("other")]); - expect(catalog.activeSnapshot()).toEqual([thread("other")]); - }); - - it("records active thread touches as catalog ordering facts", () => { - const { catalog } = catalogFixture(); - receiveActive(catalog, [ - thread("active", false, { updatedAt: 1, recencyAt: 1 }), - thread("other", false, { updatedAt: 10, recencyAt: 10 }), + expect(store.appliedMutations).toEqual([ + { kind: "remove", list: "active", threadId: "unknown" }, + { kind: "refresh", list: "archived" }, ]); + }); + + it("moves known restored and unarchived threads between lists", () => { + const store = catalogStore({ + active: [thread("active")], + archived: [thread("restored", true), thread("unarchived", true)], + }); + const catalog = createThreadCatalog({ store }); + + catalog.apply({ type: "thread-restored", thread: thread("restored", true) }); + catalog.apply({ type: "thread-unarchived", threadId: "unarchived" }); + + expect(catalog.activeSnapshot()).toEqual([thread("unarchived"), thread("restored"), thread("active")]); + expect(catalog.archivedSnapshot()).toEqual([]); + }); + + it("patches touch and delete facts without inventing unknown thread records", () => { + const store = catalogStore({ + active: [thread("active", false, { recencyAt: 1 }), thread("kept")], + archived: [thread("deleted", true)], + }); + const catalog = createThreadCatalog({ store }); catalog.apply({ type: "thread-touched", threadId: "active", recencyAt: 20 }); + catalog.apply({ type: "thread-renamed", threadId: "missing", name: "Not invented" }); + catalog.apply({ type: "thread-deleted", threadId: "deleted" }); - expect(catalog.activeSnapshot()).toEqual([ - thread("active", false, { updatedAt: 1, recencyAt: 20 }), - thread("other", false, { updatedAt: 10, recencyAt: 10 }), - ]); - }); - - it("model-checks stale snapshots around unacknowledged rename and archive facts", () => { - const maxEventDepth = 3; - - for (const sequence of eventSequences(renameOrderingEvents(), maxEventDepth)) { - const { catalog } = catalogFixture(); - receiveActive(catalog, [thread("target"), thread("other")]); - receiveArchived(catalog, []); - catalog.apply({ type: "thread-renamed", threadId: "target", name: "Renamed" }); - - let renameAcknowledged = false; - for (const event of sequence) { - event.apply(catalog); - renameAcknowledged = renameAcknowledged || event.acknowledgesRename; - } - - if (!renameAcknowledged) { - expectTargetName(catalog.activeSnapshot(), "Renamed", sequenceDescription(sequence)); - expectTargetName(catalog.archivedSnapshot(), "Renamed", sequenceDescription(sequence)); - } - } - - for (const sequence of eventSequences(archiveOrderingEvents(), maxEventDepth)) { - const { catalog } = catalogFixture(); - receiveActive(catalog, [thread("target"), thread("other")]); - receiveArchived(catalog, [thread("archived", true)]); - catalog.apply({ type: "thread-archived", threadId: "target" }); - - let activeRemovalAcknowledged = false; - let archivedUpsertAcknowledged = false; - for (const event of sequence) { - event.apply(catalog); - activeRemovalAcknowledged = activeRemovalAcknowledged || event.acknowledgesActiveRemoval; - archivedUpsertAcknowledged = archivedUpsertAcknowledged || event.acknowledgesArchivedUpsert; - } - - const sequenceName = sequenceDescription(sequence); - if (!activeRemovalAcknowledged) expectNoTarget(catalog.activeSnapshot(), sequenceName); - if (!archivedUpsertAcknowledged) expectHasArchivedTarget(catalog.archivedSnapshot(), sequenceName); - } - }); - - it("moves known unarchived threads through the catalog and refreshes unknown unarchives", async () => { - const unknownActiveRefreshStarted = deferred(); - const unknownArchivedRefreshStarted = deferred(); - const fetchThreads = vi.fn((_context: { codexPath: string; vaultPath: string }, archived: boolean) => { - if (archived) { - unknownArchivedRefreshStarted.resolve(undefined); - return Promise.resolve([]); - } - unknownActiveRefreshStarted.resolve(undefined); - return Promise.resolve([thread("unknown")]); - }); - const { catalog } = catalogFixture({ fetchThreads }); - receiveActive(catalog, [thread("active")]); - receiveArchived(catalog, [thread("known", true)]); - - catalog.apply({ type: "thread-unarchived", threadId: "known" }); - - expect(catalog.activeSnapshot()).toEqual([thread("known"), thread("active")]); + expect(catalog.activeSnapshot()).toEqual([thread("active", false, { recencyAt: 20 }), thread("kept")]); expect(catalog.archivedSnapshot()).toEqual([]); + }); - catalog.apply({ type: "thread-unarchived", threadId: "unknown" }); + it("ignores connection events from an earlier resource lease", () => { + const store = catalogStore(); + const catalog = createThreadCatalog({ store }); + const staleContext = { codexPath: "codex", vaultPath: "/vault", generation: 0 }; - await unknownActiveRefreshStarted.promise; - await unknownArchivedRefreshStarted.promise; - await vi.waitFor(() => { - expect(catalog.activeSnapshot()).toEqual([thread("known"), thread("unknown")]); - }); - expect(fetchThreads).toHaveBeenCalledTimes(2); + catalog.applyConnectionEvent(staleContext, { type: "thread-started", thread: thread("stale") }); + + expect(store.appliedMutations).toEqual([]); + expect(catalog.activeSnapshot()).toBeNull(); + }); + + it("delegates reads, pagination, and observation without a second projection store", async () => { + const store = catalogStore({ active: [thread("active")], archived: [thread("archived", true)] }); + const catalog = createThreadCatalog({ store }); + const activeObserver = vi.fn(); + const archivedObserver = vi.fn(); + + expect(await catalog.loadActive()).toEqual([thread("active")]); + expect(await catalog.refreshActive()).toEqual([thread("active")]); + expect(catalog.hasMoreActive()).toBe(false); + expect(await catalog.loadMoreActive()).toEqual([thread("active")]); + expect(await catalog.refreshArchived()).toEqual([thread("archived", true)]); + + const unsubscribeActive = catalog.observeActive(activeObserver); + const unsubscribeArchived = catalog.observeArchived(archivedObserver); + expect(activeObserver).toHaveBeenCalledWith({ value: [thread("active")], error: null, isFetching: false }); + expect(archivedObserver).toHaveBeenCalledWith({ value: [thread("archived", true)], error: null, isFetching: false }); + unsubscribeActive(); + unsubscribeArchived(); }); }); -interface ModeledCatalogEvent { - readonly name: string; - readonly acknowledgesRename: boolean; - readonly acknowledgesActiveRemoval: boolean; - readonly acknowledgesArchivedUpsert: boolean; - apply(catalog: ThreadCatalog): void; +interface CatalogStoreOptions { + readonly active?: readonly Thread[] | null; + readonly archived?: readonly Thread[] | null; } -function renameOrderingEvents(): readonly ModeledCatalogEvent[] { - return [ - modeledCatalogEvent("stale active snapshot", (catalog) => { - receiveActive(catalog, [thread("target"), thread("other")]); - }), - modeledCatalogEvent( - "rename-ack active snapshot", - (catalog) => { - receiveActive(catalog, [{ ...thread("target"), name: "Renamed" }, thread("other")]); - }, - { acknowledgesRename: true }, - ), - modeledCatalogEvent("empty archived snapshot", (catalog) => { - receiveArchived(catalog, []); - }), - ]; -} - -function archiveOrderingEvents(): readonly ModeledCatalogEvent[] { - return [ - modeledCatalogEvent("stale active snapshot", (catalog) => { - receiveActive(catalog, [thread("target"), thread("other")]); - }), - modeledCatalogEvent( - "archive-ack active snapshot", - (catalog) => { - receiveActive(catalog, [thread("other")]); - }, - { acknowledgesActiveRemoval: true }, - ), - modeledCatalogEvent("stale archived snapshot", (catalog) => { - receiveArchived(catalog, [thread("archived", true)]); - }), - modeledCatalogEvent( - "archive-ack archived snapshot", - (catalog) => { - receiveArchived(catalog, [thread("target", true), thread("archived", true)]); - }, - { acknowledgesArchivedUpsert: true }, - ), - ]; -} - -function modeledCatalogEvent( - name: string, - apply: (catalog: ThreadCatalog) => void, - acknowledgements: Partial< - Pick - > = {}, -): ModeledCatalogEvent { - return { - name, - apply, - acknowledgesRename: acknowledgements.acknowledgesRename ?? false, - acknowledgesActiveRemoval: acknowledgements.acknowledgesActiveRemoval ?? false, - acknowledgesArchivedUpsert: acknowledgements.acknowledgesArchivedUpsert ?? false, +function catalogStore(options: CatalogStoreOptions = {}) { + let active = options.active ?? null; + let archived = options.archived ?? null; + const activeObservers = new Set>(); + const archivedObservers = new Set>(); + const appliedMutations: ThreadListMutation[] = []; + const currentContext: AppServerQueryContextIdentity = { + codexPath: "codex", + vaultPath: "/vault", + generation: 1, }; -} - -function eventSequences(events: readonly T[], maxDepth: number): T[][] { - const sequences: T[][] = [[]]; - for (let depth = 1; depth <= maxDepth; depth += 1) { - for (const prefix of sequences.filter((sequence) => sequence.length === depth - 1)) { - for (const event of events) { - sequences.push([...prefix, event]); + const applyMutations = (mutations: readonly ThreadListMutation[]): void => { + appliedMutations.push(...mutations); + for (const mutation of mutations) { + if (mutation.list === "active") { + active = applyThreadListMutation(active, mutation); + } else { + archived = applyThreadListMutation(archived, mutation); } } - } - return sequences; -} - -function expectTargetName(threads: readonly Thread[] | null, expectedName: string, sequence: string): void { - const target = threads?.find((item) => item.id === "target") ?? null; - if (!target) return; - expect(target.name, sequence).toBe(expectedName); -} - -function expectNoTarget(threads: readonly Thread[] | null, sequence: string): void { - expect(threads?.some((item) => item.id === "target") ?? false, sequence).toBe(false); -} - -function expectHasArchivedTarget(threads: readonly Thread[] | null, sequence: string): void { - expect(threads?.some((item) => item.id === "target" && item.archived) ?? false, sequence).toBe(true); -} - -function sequenceDescription(sequence: readonly ModeledCatalogEvent[]): string { - return sequence.length === 0 ? "no additional snapshots" : sequence.map((event) => event.name).join(" -> "); -} - -const catalogStores = new WeakMap(); - -function receiveActive(catalog: ThreadCatalog, threads: readonly Thread[]): void { - catalogStores.get(catalog)?.receiveActive(threads); -} - -function receiveArchived(catalog: ThreadCatalog, threads: readonly Thread[]): void { - catalogStores.get(catalog)?.receiveArchived(threads); -} - -function catalogFixture( - options: { - fetchThreads?: (context: { codexPath: string; vaultPath: string }, archived: boolean) => Promise; - context?: () => { codexPath: string; vaultPath: string }; - } = {}, -) { - const context = options.context ?? (() => ({ codexPath: "codex", vaultPath: "/vault" })); - const queries = new AppServerResourceStore({ - clientRunner: clientRunnerWithThreads(options.fetchThreads ?? (() => Promise.resolve([]))), - }); - queries.initialize(context()); - const store = new TestThreadCatalogStore(queries); - const catalog = createThreadCatalog({ store }); - catalogStores.set(catalog, store); - catalog.observeActive(() => undefined); - catalog.observeArchived(() => undefined); - return { - catalog, - contextIdentity: () => queries.contextIdentity(), - replaceContext: (nextContext: AppServerQueryContext) => queries.replaceContext(nextContext), }; -} - -class TestThreadCatalogStore { - private readonly activeSnapshots = new Map(); - private readonly archivedSnapshots = new Map(); - private readonly activeObservers = new Set>(); - private readonly archivedObservers = new Set>(); - - constructor(private readonly queries: AppServerResourceStore) {} - - contextKey(): string { - return this.queries.contextKey(); - } - - contextKeyFor(context: AppServerQueryContextIdentity): string { - return this.queries.contextKeyFor(context); - } - - activeThreadsSnapshot(): readonly Thread[] | null { - return this.activeSnapshots.get(this.contextKey()) ?? this.queries.activeThreadsSnapshot(); - } - - archivedThreadsSnapshot(): readonly Thread[] | null { - return this.archivedSnapshots.get(this.contextKey()) ?? this.queries.archivedThreadsSnapshot(); - } - - async fetchAllActiveThreads(): Promise { - return this.receiveLoadedActive(await this.queries.fetchAllActiveThreads()); - } - - hasMoreActiveThreads(): boolean { - return this.queries.hasMoreActiveThreads(); - } - - async loadMoreActiveThreads(): Promise { - return this.receiveLoadedActive(await this.queries.loadMoreActiveThreads()); - } - - async refreshActiveThreads(): Promise { - return this.receiveLoadedActive(await this.queries.refreshActiveThreads()); - } - - async refreshArchivedThreads(): Promise { - return this.receiveLoadedArchived(await this.queries.refreshArchivedThreads()); - } - - observeActiveThreadsResult(observer: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void { - this.activeObservers.add(observer); - const unsubscribe = this.queries.observeActiveThreadsResult((result) => { - if (result.value && !result.isFetching) this.activeSnapshots.delete(this.contextKey()); - observer(result); - }, options); - return () => { - this.activeObservers.delete(observer); - unsubscribe(); - }; - } - - observeArchivedThreadsResult(observer: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void { - this.archivedObservers.add(observer); - const unsubscribe = this.queries.observeArchivedThreadsResult((result) => { - if (result.value && !result.isFetching) this.archivedSnapshots.delete(this.contextKey()); - observer(result); - }, options); - return () => { - this.archivedObservers.delete(observer); - unsubscribe(); - }; - } - - receiveActive(threads: readonly Thread[]): void { - this.activeSnapshots.set(this.contextKey(), threads); - const result = observedSnapshot(threads); - for (const observer of this.activeObservers) observer(result); - } - - receiveArchived(threads: readonly Thread[]): void { - this.archivedSnapshots.set(this.contextKey(), threads); - const result = observedSnapshot(threads); - for (const observer of this.archivedObservers) observer(result); - } - - private receiveLoadedActive(threads: readonly Thread[]): readonly Thread[] { - this.activeSnapshots.delete(this.contextKey()); - return threads; - } - - private receiveLoadedArchived(threads: readonly Thread[]): readonly Thread[] { - this.archivedSnapshots.delete(this.contextKey()); - return threads; - } -} - -function observedSnapshot(threads: readonly Thread[]): ObservedResult { - return { value: threads, error: null, isFetching: false }; -} - -function clientRunnerWithThreads( - fetchThreads: (context: { codexPath: string; vaultPath: string }, archived: boolean) => Promise, -): AppServerQueryClientRunner { return { - runWithClient: async (context, operation) => { - return operation({ - request: async (method: string, params: { archived?: boolean } = {}) => { - if (method !== "thread/list") throw new Error(`Unexpected app-server request: ${method}`); - return { - data: await fetchThreads(context, params.archived ?? false), - nextCursor: null, - }; - }, - } as never); + appliedMutations, + contextKey: () => contextKey(currentContext), + contextKeyFor: contextKey, + activeThreadsSnapshot: () => active, + archivedThreadsSnapshot: () => archived, + fetchAllActiveThreads: async () => active ?? [], + hasMoreActiveThreads: () => false, + loadMoreActiveThreads: async () => active ?? [], + refreshActiveThreads: async () => active ?? [], + refreshArchivedThreads: async () => archived ?? [], + applyThreadListMutations: applyMutations, + observeActiveThreadsResult: (observer: ObservedResultListener, observeOptions: { emitCurrent?: boolean } = {}) => { + activeObservers.add(observer); + if (observeOptions.emitCurrent ?? true) observer({ value: active, error: null, isFetching: false }); + return () => activeObservers.delete(observer); + }, + observeArchivedThreadsResult: (observer: ObservedResultListener, observeOptions: { emitCurrent?: boolean } = {}) => { + archivedObservers.add(observer); + if (observeOptions.emitCurrent ?? true) observer({ value: archived, error: null, isFetching: false }); + return () => archivedObservers.delete(observer); }, }; } -function observedActiveThreadIds(listener: Mock): string[][] { - return listener.mock.calls - .map((call) => { - const result = call[0] as { value: readonly Thread[] | null }; - return result.value?.map((item) => item.id) ?? null; - }) - .filter((ids): ids is string[] => ids !== null); +function contextKey(context: AppServerQueryContextIdentity): string { + return `${String(context.generation)}:${context.codexPath}:${context.vaultPath}`; } function thread(id: string, archived = false, overrides: Partial = {}): Thread { @@ -594,23 +163,3 @@ function thread(id: string, archived = false, overrides: Partial = {}): ...overrides, }; } - -function deferred(): { - promise: Promise; - resolve: (value: T) => void; - reject: (reason?: unknown) => void; -} { - let resolve!: (value: T) => void; - let reject!: (reason?: unknown) => void; - const promise = new Promise((promiseResolve, promiseReject) => { - resolve = promiseResolve; - reject = promiseReject; - }); - return { promise, resolve, reject }; -} - -async function flushMicrotasks(): Promise { - for (let index = 0; index < 10; index += 1) { - await Promise.resolve(); - } -}