fix(query): align resource refresh semantics

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<readonly Thread[]> {
@@ -178,21 +168,13 @@ export class AppServerQueryCache {
   async fetchActiveThreadSearchInventory(): Promise<readonly Thread[]> {
     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<T>(read: () => Promise<T>, fallback?: () => T | null): Promise<T> {
+    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<T>(queryKey: readonly unknown[], read: () => Promise<T>): Promise<T> {
+    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: {

src/features/chat/application/connection/connection-actions.ts

@@ -27,7 +27,6 @@ export interface ChatConnectionActionsHost {
   metadata: ChatConnectionMetadataActions;
   diagnostics: ChatConnectionDiagnosticsActions;
   invalidateThreadWork: () => void;
-  loadSharedThreads: () => Promise<void>;
   refreshSharedThreads: () => Promise<void>;
   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)}`);

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<void> => {
-    await environment.plugin.threadCatalog.loadActive();
-  };
   const refreshSharedThreads = async (): Promise<void> => {
     await environment.plugin.threadCatalog.refreshActive();
   };
@@ -208,7 +205,6 @@ export function createConnectionBundle(
     diagnostics: {
       refreshServerDiagnostics: (options) => serverDiagnostics.refreshServerDiagnostics(options),
     },
-    loadSharedThreads,
     refreshSharedThreads,
     scheduleDeferredDiagnostics: () => {
       scheduleDeferredDiagnosticsRefresh({

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<typeof thread>[]; nextCursor: null }>();
+    const secondRead = deferred<{ data: ReturnType<typeof thread>[]; 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<typeof thread>[]; 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<typeof thread>[]; 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<typeof thread>[]; nextCursor: null }>();
+    const secondRead = deferred<{ data: ReturnType<typeof thread>[]; 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<typeof thread>[]; 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 });

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();