diff --git a/src/app-server/query/cache.ts b/src/app-server/query/cache.ts index 33e81471..1fed6397 100644 --- a/src/app-server/query/cache.ts +++ b/src/app-server/query/cache.ts @@ -152,23 +152,13 @@ export class AppServerQueryCache { await this.client.invalidateQueries({ queryKey: key, refetchType: "none" }); this.assertUsable(); } - let data: ActiveThreadData; - try { - data = await this.client.fetchInfiniteQuery({ - ...this.activeThreadsQueryOptions(), - ...(options.force ? { staleTime: 0 } : {}), - }); - } catch (error) { - if (!(error instanceof CancelledError)) throw error; - const snapshot = this.activeThreadsSnapshot(); - if (snapshot) return snapshot; - data = await this.client.fetchInfiniteQuery({ - ...this.activeThreadsQueryOptions(), - ...(options.force ? { staleTime: 0 } : {}), - }); - } - this.assertUsable(); - return cloneThreads(activeThreadsFromData(data) ?? []); + return this.readThroughQueryCancellation( + async () => { + const data = await this.client.fetchInfiniteQuery(this.activeThreadsQueryOptions()); + return cloneThreads(activeThreadsFromData(data) ?? []); + }, + () => this.activeThreadsSnapshot(), + ); } async refreshActiveThreads(): Promise { @@ -178,21 +168,13 @@ export class AppServerQueryCache { async fetchActiveThreadSearchInventory(): Promise { this.assertUsable(); if (!appServerQueryContextIsComplete(this.context)) return []; + const key = activeThreadSearchInventoryQueryKey(this.context); const options = { - queryKey: activeThreadSearchInventoryQueryKey(this.context), + queryKey: key, queryFn: ({ signal }: { signal: AbortSignal }) => this.runWithClient((client) => listThreads(client, this.context.vaultPath, { signal })), - staleTime: Number.POSITIVE_INFINITY, }; - let threads: readonly Thread[]; - try { - threads = await this.client.fetchQuery(options); - } catch (error) { - if (!(error instanceof CancelledError)) throw error; - threads = await this.client.fetchQuery(options); - } - this.assertUsable(); - return cloneThreads(threads); + return this.readFreshThroughQueryCancellation(key, async () => cloneThreads(await this.client.fetchQuery(options))); } hasMoreActiveThreads(): boolean { @@ -234,9 +216,9 @@ export class AppServerQueryCache { await this.client.invalidateQueries({ queryKey: key, refetchType: "none" }); this.assertUsable(); } - const threads = await this.client.fetchQuery(this.archivedThreadsQueryOptions()); - this.assertUsable(); - return cloneThreads(threads); + return this.readFreshThroughQueryCancellation(key, async () => + cloneThreads(await this.client.fetchQuery(this.archivedThreadsQueryOptions())), + ); } applyThreadListMutations(mutations: readonly ThreadListMutation[]): void { @@ -640,6 +622,29 @@ export class AppServerQueryCache { return runner.runWithClient(this.context, operation, options); } + private async readThroughQueryCancellation(read: () => Promise, fallback?: () => T | null): Promise { + for (;;) { + try { + const value = await read(); + this.assertUsable(); + return value; + } catch (error) { + if (!(error instanceof CancelledError)) throw error; + this.assertUsable(); + const fallbackValue = fallback?.(); + if (fallbackValue != null) return fallbackValue; + } + } + } + + private async readFreshThroughQueryCancellation(queryKey: readonly unknown[], read: () => Promise): Promise { + for (;;) { + const value = await this.readThroughQueryCancellation(read); + this.assertUsable(); + if (!this.client.getQueryState(queryKey)?.isInvalidated) return value; + } + } + private assertUsable(): void { if (this.disposed) throw new Error("Codex app-server query cache was disposed."); } @@ -665,8 +670,8 @@ function createAppServerQueryClient(): QueryClient { return new QueryClient({ defaultOptions: { queries: { + networkMode: "always", retry: false, - refetchOnReconnect: false, refetchOnWindowFocus: false, }, mutations: { diff --git a/src/features/chat/application/connection/connection-actions.ts b/src/features/chat/application/connection/connection-actions.ts index accc4de3..7e89d3ae 100644 --- a/src/features/chat/application/connection/connection-actions.ts +++ b/src/features/chat/application/connection/connection-actions.ts @@ -27,7 +27,6 @@ export interface ChatConnectionActionsHost { metadata: ChatConnectionMetadataActions; diagnostics: ChatConnectionDiagnosticsActions; invalidateThreadWork: () => void; - loadSharedThreads: () => Promise; refreshSharedThreads: () => Promise; scheduleDeferredDiagnostics: () => void; clearDeferredDiagnostics: () => void; @@ -165,7 +164,7 @@ async function hydrateConnectedResources(host: ChatConnectionActionsHost, isStal if (isStale()) return; try { - await host.loadSharedThreads(); + await host.refreshSharedThreads(); } catch (error) { if (isStale() || host.isStaleResourceContextError(error)) return; host.addSystemMessage(`Could not refresh Codex threads: ${errorMessage(error)}`); diff --git a/src/features/chat/host/bundles/connection-bundle.ts b/src/features/chat/host/bundles/connection-bundle.ts index 48122a34..56c5a17b 100644 --- a/src/features/chat/host/bundles/connection-bundle.ts +++ b/src/features/chat/host/bundles/connection-bundle.ts @@ -134,9 +134,6 @@ export function createConnectionBundle( diagnosticsTransport, appServerMetadataSnapshot: () => environment.plugin.appServerQueries.appServerMetadataSnapshot(), }); - const loadSharedThreads = async (): Promise => { - await environment.plugin.threadCatalog.loadActive(); - }; const refreshSharedThreads = async (): Promise => { await environment.plugin.threadCatalog.refreshActive(); }; @@ -208,7 +205,6 @@ export function createConnectionBundle( diagnostics: { refreshServerDiagnostics: (options) => serverDiagnostics.refreshServerDiagnostics(options), }, - loadSharedThreads, refreshSharedThreads, scheduleDeferredDiagnostics: () => { scheduleDeferredDiagnosticsRefresh({ diff --git a/tests/app-server/query-cache.test.ts b/tests/app-server/query-cache.test.ts index 6dbd3b1e..c05f5d62 100644 --- a/tests/app-server/query-cache.test.ts +++ b/tests/app-server/query-cache.test.ts @@ -1,3 +1,4 @@ +import { onlineManager } from "@tanstack/query-core"; import { describe, expect, it, vi } from "vitest"; import type { CatalogModel, CatalogSkillMetadata } from "../../src/app-server/protocol/catalog"; import { AppServerQueryCache } from "../../src/app-server/query/cache"; @@ -303,6 +304,48 @@ describe("AppServerQueryCache", () => { expect(listThreads).toHaveBeenCalledTimes(2); }); + it("rejoins the active thread query through repeated event cancellations", async () => { + const firstRead = deferred<{ data: ReturnType[]; nextCursor: null }>(); + const secondRead = deferred<{ data: ReturnType[]; nextCursor: null }>(); + const listThreads = vi + .fn() + .mockImplementationOnce(() => firstRead.promise) + .mockImplementationOnce(() => secondRead.promise) + .mockResolvedValueOnce({ data: [thread("authoritative")], nextCursor: null }); + const cache = cacheWithRequestHandlers({ "thread/list": listThreads }); + + const initial = cache.refreshActiveThreads(); + await flushMicrotasks(); + cache.applyThreadListMutations([{ kind: "update", list: "active", threadId: "first-event", changes: { name: "first" } }]); + await vi.waitFor(() => expect(listThreads).toHaveBeenCalledTimes(2)); + cache.applyThreadListMutations([{ kind: "update", list: "active", threadId: "second-event", changes: { name: "second" } }]); + + await expect(initial).resolves.toEqual([thread("authoritative")]); + expect(listThreads).toHaveBeenCalledTimes(3); + firstRead.resolve({ data: [thread("obsolete-first")], nextCursor: null }); + secondRead.resolve({ data: [thread("obsolete-second")], nextCursor: null }); + }); + + it("rejoins an archived thread refresh cancelled by an exact event", async () => { + const staleRead = deferred<{ data: ReturnType[]; nextCursor: null }>(); + const listThreads = vi + .fn() + .mockResolvedValueOnce({ data: [thread("cached", true)], nextCursor: null }) + .mockImplementationOnce(() => staleRead.promise) + .mockResolvedValueOnce({ data: [thread("authoritative", true)], nextCursor: null }); + const cache = cacheWithRequestHandlers({ "thread/list": listThreads }); + await cache.refreshArchivedThreads(); + + const refresh = cache.refreshArchivedThreads(); + await vi.waitFor(() => expect(listThreads).toHaveBeenCalledTimes(2)); + cache.applyThreadListMutations([{ kind: "update", list: "archived", threadId: "cached", changes: { name: "from-event" } }]); + + await vi.waitFor(() => expect(listThreads).toHaveBeenCalledTimes(3)); + await expect(refresh).resolves.toEqual([thread("authoritative", true)]); + expect(cache.archivedThreadsSnapshot()).toEqual([thread("authoritative", true)]); + staleRead.resolve({ data: [thread("stale", true)], nextCursor: null }); + }); + it("keeps a thread-picker inventory read out of the shared recent list", async () => { const oldInventory = deferred<{ data: ReturnType[]; nextCursor: string | null }>(); const listThreads = vi @@ -321,6 +364,73 @@ describe("AppServerQueryCache", () => { expect(cache.hasMoreActiveThreads()).toBe(true); }); + it("refreshes the complete thread-picker inventory for each operation", async () => { + const listThreads = vi + .fn() + .mockResolvedValueOnce({ data: [thread("first")], nextCursor: null }) + .mockResolvedValueOnce({ data: [thread("second")], nextCursor: null }); + const cache = cacheWithRequestHandlers({ "thread/list": listThreads }); + + await expect(cache.fetchActiveThreadSearchInventory()).resolves.toEqual([thread("first")]); + await expect(cache.fetchActiveThreadSearchInventory()).resolves.toEqual([thread("second")]); + + expect(listThreads).toHaveBeenCalledTimes(2); + }); + + it("rejoins the complete thread-picker inventory through repeated event cancellations", async () => { + const firstRead = deferred<{ data: ReturnType[]; nextCursor: null }>(); + const secondRead = deferred<{ data: ReturnType[]; nextCursor: null }>(); + const listThreads = vi + .fn() + .mockImplementationOnce(() => firstRead.promise) + .mockImplementationOnce(() => secondRead.promise) + .mockResolvedValueOnce({ data: [thread("authoritative")], nextCursor: null }); + const cache = cacheWithRequestHandlers({ "thread/list": listThreads }); + + const inventory = cache.fetchActiveThreadSearchInventory(); + await flushMicrotasks(); + cache.applyThreadListMutations([{ kind: "update", list: "active", threadId: "first-event", changes: { name: "first" } }]); + await vi.waitFor(() => expect(listThreads).toHaveBeenCalledTimes(2)); + cache.applyThreadListMutations([{ kind: "update", list: "active", threadId: "second-event", changes: { name: "second" } }]); + + await expect(inventory).resolves.toEqual([thread("authoritative")]); + expect(listThreads).toHaveBeenCalledTimes(3); + firstRead.resolve({ data: [thread("obsolete-first")], nextCursor: null }); + secondRead.resolve({ data: [thread("obsolete-second")], nextCursor: null }); + }); + + it("does not reuse a cached complete inventory after an event cancels its refresh", async () => { + const staleRead = deferred<{ data: ReturnType[]; nextCursor: null }>(); + const listThreads = vi + .fn() + .mockResolvedValueOnce({ data: [thread("cached")], nextCursor: null }) + .mockImplementationOnce(() => staleRead.promise) + .mockResolvedValueOnce({ data: [thread("authoritative")], nextCursor: null }); + const cache = cacheWithRequestHandlers({ "thread/list": listThreads }); + await cache.fetchActiveThreadSearchInventory(); + + const inventory = cache.fetchActiveThreadSearchInventory(); + await vi.waitFor(() => expect(listThreads).toHaveBeenCalledTimes(2)); + cache.applyThreadListMutations([{ kind: "update", list: "active", threadId: "event", changes: { name: "changed" } }]); + + await expect(inventory).resolves.toEqual([thread("authoritative")]); + expect(listThreads).toHaveBeenCalledTimes(3); + staleRead.resolve({ data: [thread("stale")], nextCursor: null }); + }); + + it("runs local app-server queries independently of browser network state", async () => { + const listModels = vi.fn().mockResolvedValue({ data: [catalogModel("local")] }); + const cache = cacheWithRequestHandlers({ "model/list": listModels }); + onlineManager.setOnline(false); + + try { + await expect(cache.fetchModels()).resolves.toMatchObject([{ model: "local" }]); + expect(listModels).toHaveBeenCalledOnce(); + } finally { + onlineManager.setOnline(true); + } + }); + it("clears snapshots and rejects new reads after disposal", async () => { const listModels = vi.fn().mockResolvedValue({ data: [catalogModel("cached")] }); const cache = cacheWithRequestHandlers({ "model/list": listModels }); diff --git a/tests/features/chat/application/connection/connection-actions.test.ts b/tests/features/chat/application/connection/connection-actions.test.ts index 2bb7daf8..566f739d 100644 --- a/tests/features/chat/application/connection/connection-actions.test.ts +++ b/tests/features/chat/application/connection/connection-actions.test.ts @@ -38,7 +38,6 @@ function createActionsHarness({ connected = false, canConnect = true } = {}) { metadata, diagnostics, invalidateThreadWork: vi.fn(), - loadSharedThreads: vi.fn().mockResolvedValue(undefined), refreshSharedThreads: vi.fn().mockResolvedValue(undefined), scheduleDeferredDiagnostics: vi.fn(), clearDeferredDiagnostics: vi.fn(), @@ -119,8 +118,7 @@ describe("ChatConnectionActions", () => { userAgent: "test", }); expect(refreshAppServerMetadata).toHaveBeenCalledOnce(); - expect(host.loadSharedThreads).toHaveBeenCalledOnce(); - expect(host.refreshSharedThreads).not.toHaveBeenCalled(); + expect(host.refreshSharedThreads).toHaveBeenCalledOnce(); expect(host.scheduleDeferredDiagnostics).toHaveBeenCalledOnce(); expect(host.setStatus).toHaveBeenCalledWith("Connected.", { kind: "connected" }); }); @@ -132,7 +130,7 @@ describe("ChatConnectionActions", () => { await actions.ensureConnected(); expect(stateStore.getState().connection.initializeResponse).toMatchObject({ codexHome: "/codex" }); - expect(host.loadSharedThreads).toHaveBeenCalledOnce(); + expect(host.refreshSharedThreads).toHaveBeenCalledOnce(); expect(host.setStatus).toHaveBeenCalledWith("Connected.", { kind: "connected" }); expect(host.setStatus).not.toHaveBeenCalledWith("Connection failed.", expect.anything()); expect(host.addSystemMessage).toHaveBeenCalledWith("Could not refresh Codex metadata: config unavailable"); @@ -141,7 +139,7 @@ describe("ChatConnectionActions", () => { it("keeps the initialized connection usable when thread hydration fails", async () => { const { actions, host } = createActionsHarness(); - vi.mocked(host.loadSharedThreads).mockRejectedValueOnce(new Error("threads unavailable")); + vi.mocked(host.refreshSharedThreads).mockRejectedValueOnce(new Error("threads unavailable")); await actions.ensureConnected();