diff --git a/src/plugin-runtime.ts b/src/plugin-runtime.ts index 3cf3e278..193af41d 100644 --- a/src/plugin-runtime.ts +++ b/src/plugin-runtime.ts @@ -335,13 +335,29 @@ export class CodexPanelRuntime implements AppServerClientAccess { if (!appServerQueryContextIsComplete(context)) { throw new Error("Codex app-server query context is incomplete."); } - const result = await this.runWithContextClient(context, operation, options); - if (!appServerQueryContextIdentityMatches(this.appServerResourceStore.contextIdentity(), context)) { - throw new StaleAppServerResourceContextError(); - } + this.assertCurrentAppServerContext(context); + const result = await this.runWithContextClient( + context, + (client) => { + this.assertCurrentAppServerContext(context); + return operation(client); + }, + options, + ); + this.assertCurrentAppServerContext(context); return result; } + private assertCurrentAppServerContext(context: AppServerQueryContextIdentity): void { + let current = false; + try { + current = appServerQueryContextIdentityMatches(this.appServerResourceStore.contextIdentity(), context); + } catch { + // A reset resource store has no current app-server context. + } + if (!current) throw new StaleAppServerResourceContextError(); + } + private runWithContextClient( context: AppServerQueryContextIdentity, operation: (client: AppServerClient) => Promise, diff --git a/src/settings/app-server-dynamic-data.ts b/src/settings/app-server-dynamic-data.ts index 0bceff72..f0adee87 100644 --- a/src/settings/app-server-dynamic-data.ts +++ b/src/settings/app-server-dynamic-data.ts @@ -9,6 +9,7 @@ import type { ThreadCatalogArchivedReader, ThreadCatalogEventSink } from "../fea import { type SettingsDynamicDataAccess, type SettingsHookCatalog, StaleSettingsDynamicDataContextError } from "./dynamic-data"; interface SettingsAppServerQueries { + contextKey(): string; modelsSnapshot(): readonly ModelMetadata[] | null; observeModelsResult(listener: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void; fetchModels(): Promise; @@ -25,10 +26,19 @@ export interface SettingsAppServerDynamicDataOptions { } export function createSettingsAppServerDynamicData(options: SettingsAppServerDynamicDataOptions): SettingsDynamicDataAccess { + const hookMutations = createSettingsMutationQueue<"hooks">(); + const archivedThreadMutations = createSettingsMutationQueue(); const withSettingsConnection = (operation: (client: AppServerClient) => Promise): Promise => options.clientAccess.withClient(operation, { serverRequests: { kind: "reject", message: "Codex Panel settings does not handle server requests." }, }); + const runMutation = (queue: SettingsMutationQueue, key: K, contextKey: string, operation: () => Promise): Promise => + queue.run(key, () => + mapStaleContextError(async () => { + if (options.appServerQueries.contextKey() !== contextKey) throw new StaleSettingsDynamicDataContextError(); + return operation(); + }), + ); return { modelsSnapshot: () => options.appServerQueries.modelsSnapshot(), @@ -39,20 +49,53 @@ export function createSettingsAppServerDynamicData(options: SettingsAppServerDyn observeArchivedThreadsResult: (listener, observeOptions) => options.threadCatalog.observeArchived(listener, observeOptions), refreshArchivedThreads: () => mapStaleContextError(() => options.threadCatalog.refreshArchived()), loadHooks: () => withSettingsConnection((client) => loadSettingsHookCatalog(client, options.vaultPath)), - trustHook: (hook) => withSettingsConnection((client) => trustHookItem(client, hook)), - setHookEnabled: (hook, enabled) => withSettingsConnection((client) => setHookItemEnabled(client, hook, enabled)), - restoreArchivedThread: async (threadId, mutationOptions) => { - const thread = await withSettingsConnection((client) => restoreArchivedThreadOnAppServer(client, threadId)); - if (mutationOptions?.shouldPublish?.() ?? true) { - options.threadCatalog.apply({ type: "thread-restored", thread }); - } - return thread; + trustHook: (hook) => { + const contextKey = options.appServerQueries.contextKey(); + return runMutation(hookMutations, "hooks", contextKey, () => withSettingsConnection((client) => trustHookItem(client, hook))); }, - deleteArchivedThread: async (threadId, mutationOptions) => { - await withSettingsConnection((client) => deleteThread(client, threadId)); - if (mutationOptions?.shouldPublish?.() ?? true) { + setHookEnabled: (hook, enabled) => { + const contextKey = options.appServerQueries.contextKey(); + return runMutation(hookMutations, "hooks", contextKey, () => + withSettingsConnection((client) => setHookItemEnabled(client, hook, enabled)), + ); + }, + restoreArchivedThread: (threadId) => { + const contextKey = options.appServerQueries.contextKey(); + return runMutation(archivedThreadMutations, threadId, contextKey, async () => { + const thread = await withSettingsConnection((client) => restoreArchivedThreadOnAppServer(client, threadId)); + options.threadCatalog.apply({ type: "thread-restored", thread }); + return thread; + }); + }, + deleteArchivedThread: (threadId) => { + const contextKey = options.appServerQueries.contextKey(); + return runMutation(archivedThreadMutations, threadId, contextKey, async () => { + await withSettingsConnection((client) => deleteThread(client, threadId)); options.threadCatalog.apply({ type: "thread-deleted", threadId }); - } + }); + }, + }; +} + +interface SettingsMutationQueue { + run(key: K, operation: () => Promise): Promise; +} + +function createSettingsMutationQueue(): SettingsMutationQueue { + const pendingByKey = new Map>(); + return { + run(key, operation) { + const previous = pendingByKey.get(key) ?? Promise.resolve(); + const result = previous.then(operation); + const pending = result.then( + () => undefined, + () => undefined, + ); + pendingByKey.set(key, pending); + void pending.then(() => { + if (pendingByKey.get(key) === pending) pendingByKey.delete(key); + }); + return result; }, }; } diff --git a/src/settings/dynamic-data.ts b/src/settings/dynamic-data.ts index c06c9eed..cd2ce99a 100644 --- a/src/settings/dynamic-data.ts +++ b/src/settings/dynamic-data.ts @@ -9,10 +9,6 @@ export interface SettingsHookCatalog { status: string; } -interface SettingsDynamicDataMutationOptions { - shouldPublish?: () => boolean; -} - export interface SettingsDynamicDataAccess { modelsSnapshot(): readonly ModelMetadata[] | null; observeModelsResult(listener: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void; @@ -24,8 +20,8 @@ export interface SettingsDynamicDataAccess { loadHooks(): Promise; trustHook(hook: HookItem): Promise; setHookEnabled(hook: HookItem, enabled: boolean): Promise; - restoreArchivedThread(threadId: string, options?: SettingsDynamicDataMutationOptions): Promise; - deleteArchivedThread(threadId: string, options?: SettingsDynamicDataMutationOptions): Promise; + restoreArchivedThread(threadId: string): Promise; + deleteArchivedThread(threadId: string): Promise; } export class StaleSettingsDynamicDataContextError extends Error { diff --git a/src/settings/dynamic-sections-controller.ts b/src/settings/dynamic-sections-controller.ts index ff34be46..cc75cf82 100644 --- a/src/settings/dynamic-sections-controller.ts +++ b/src/settings/dynamic-sections-controller.ts @@ -275,9 +275,7 @@ export class SettingsDynamicSectionsController { failureStatus: (error) => `Could not restore archived thread: ${errorMessage(error)}`, failureNotice: "Could not restore archived Codex thread.", operation: async (operationToken) => { - const restoredThread = await this.host.dynamicData.restoreArchivedThread(threadId, { - shouldPublish: () => !this.isStaleArchivedThreadsOperation(operationToken), - }); + const restoredThread = await this.host.dynamicData.restoreArchivedThread(threadId); if (this.isStaleArchivedThreadsOperation(operationToken)) return; this.archivedThreads = this.archivedThreads.filter((thread) => thread.id !== threadId); this.archivedThreadsLifecycle = settingsDynamicSectionLoaded(`Restored "${threadArchiveDisplayTitle(restoredThread)}".`); @@ -294,9 +292,7 @@ export class SettingsDynamicSectionsController { failureStatus: (error) => `Could not delete archived thread: ${errorMessage(error)}`, failureNotice: "Could not delete archived Codex thread.", operation: async (operationToken) => { - await this.host.dynamicData.deleteArchivedThread(threadId, { - shouldPublish: () => !this.isStaleArchivedThreadsOperation(operationToken), - }); + await this.host.dynamicData.deleteArchivedThread(threadId); if (this.isStaleArchivedThreadsOperation(operationToken)) return; this.archivedThreads = this.archivedThreads.filter((thread) => thread.id !== threadId); this.archivedThreadsLifecycle = settingsDynamicSectionLoaded(`Deleted "${title}".`); @@ -365,7 +361,7 @@ export class SettingsDynamicSectionsController { try { await options.operation(operationToken); } catch (error) { - if (stale()) return; + if (stale() || isStaleSettingsDynamicDataContextError(error)) return; setLifecycle(settingsDynamicSectionFailed(options.failureStatus(error))); this.callbacks.notify(options.failureNotice); } finally { diff --git a/tests/plugin-runtime.integration.test.ts b/tests/plugin-runtime.integration.test.ts index d5e6c225..2b471492 100644 --- a/tests/plugin-runtime.integration.test.ts +++ b/tests/plugin-runtime.integration.test.ts @@ -228,6 +228,28 @@ describe("CodexPanelPlugin runtime integration", () => { await expect(operation).rejects.toThrow("Codex app-server resource context changed while loading."); }); + it("does not start a short-lived operation when its app-server context changes while connecting", async () => { + const clientReady = deferred(); + const shortLivedClient = { request: vi.fn().mockResolvedValue({}) }; + withShortLivedAppServerClientMock.mockImplementation( + (_codexPath: string, _vaultPath: string, operation: (client: typeof shortLivedClient) => Promise) => + clientReady.promise.then(() => operation(shortLivedClient)), + ); + const plugin = await pluginWithLeaves([]); + await publishCodexPath(plugin, "codex-a"); + const callback = vi.fn(() => Promise.resolve("unused")); + + const operation = plugin.runtime.withClient(callback, { + serverRequests: { kind: "reject", message: "test" }, + }); + await publishCodexPath(plugin, "codex-b"); + clientReady.resolve(); + + await expect(operation).rejects.toThrow("Codex app-server resource context changed while loading."); + expect(callback).not.toHaveBeenCalled(); + expect(shortLivedClient.request).not.toHaveBeenCalled(); + }); + it("publishes a persisted context and its new resource lease atomically", async () => { const plugin = await pluginWithLeaves([]); const firstLease = plugin.runtime.appServerContextLease(); diff --git a/tests/settings/dynamic-sections-controller.test.ts b/tests/settings/dynamic-sections-controller.test.ts index f6dea350..47965b13 100644 --- a/tests/settings/dynamic-sections-controller.test.ts +++ b/tests/settings/dynamic-sections-controller.test.ts @@ -183,7 +183,7 @@ describe("SettingsDynamicSectionsController", () => { expect(snapshot.hooks.map((item) => item.currentHash)).toEqual(["newhash"]); }); - it("ignores stale archived restore results after a newer dynamic operation completes", async () => { + it("publishes stale-view archived restore facts without replacing a newer section result", async () => { const staleRestore = deferred<{ thread: ThreadRecord }>(); const applyThreadCatalogEvent = vi.fn(); const initialClient = settingsClient(); @@ -211,10 +211,109 @@ describe("SettingsDynamicSectionsController", () => { staleRestore.resolve({ thread: appServerThread({ id: "thread-old", preview: "Restored old" }) }); await restore; - expect(applyThreadCatalogEvent).not.toHaveBeenCalled(); + expect(applyThreadCatalogEvent).toHaveBeenCalledWith({ + type: "thread-restored", + thread: expect.objectContaining({ id: "thread-old", preview: "Restored old", archived: false }), + }); expect(controller.snapshot().archivedThreads.map((thread) => thread.preview)).toEqual(["New archived"]); }); + it("serializes conflicting archived mutations and publishes their facts in order", async () => { + const restoreResult = deferred<{ thread: ThreadRecord }>(); + const deleteResult = deferred(); + const restoreRequest = vi.fn(() => restoreResult.promise); + const deleteRequest = vi.fn(() => deleteResult.promise); + const restoreClient = settingsRequestClient({ + "thread/unarchive": restoreRequest, + }); + const deleteClient = settingsRequestClient({ + "thread/delete": deleteRequest, + }); + const applyThreadCatalogEvent = vi.fn(); + useShortLivedClients(restoreClient, deleteClient); + const controller = new SettingsDynamicSectionsController(settingsTabHost({ applyThreadCatalogEvent }), { + display: vi.fn(), + notify: vi.fn(), + }); + + const restore = controller.restoreArchivedThread("thread-old"); + const deletion = controller.deleteArchivedThread("thread-old"); + await flushPromises(); + + expect(restoreRequest).toHaveBeenCalledOnce(); + expect(deleteRequest).not.toHaveBeenCalled(); + expect(withShortLivedAppServerClientMock).toHaveBeenCalledOnce(); + + restoreResult.resolve({ thread: appServerThread({ id: "thread-old", preview: "Restored old" }) }); + await flushPromises(); + + expect(deleteRequest).toHaveBeenCalledOnce(); + expect(applyThreadCatalogEvent).toHaveBeenCalledTimes(1); + + deleteResult.resolve({}); + await Promise.all([restore, deletion]); + + expect(applyThreadCatalogEvent.mock.calls.map(([event]) => event.type)).toEqual(["thread-restored", "thread-deleted"]); + expect(withShortLivedAppServerClientMock).toHaveBeenCalledTimes(2); + }); + + it("publishes a completed archived mutation after the settings view is disposed", async () => { + const restoreResult = deferred<{ thread: ThreadRecord }>(); + const restoreClient = settingsRequestClient({ + "thread/unarchive": vi.fn(() => restoreResult.promise), + }); + const applyThreadCatalogEvent = vi.fn(); + const display = vi.fn(); + useShortLivedClients(restoreClient); + const controller = new SettingsDynamicSectionsController(settingsTabHost({ applyThreadCatalogEvent }), { + display, + notify: vi.fn(), + }); + + const restore = controller.restoreArchivedThread("thread-old"); + await flushPromises(); + controller.dispose(); + display.mockClear(); + + restoreResult.resolve({ thread: appServerThread({ id: "thread-old", preview: "Restored after close" }) }); + await restore; + + expect(applyThreadCatalogEvent).toHaveBeenCalledWith({ + type: "thread-restored", + thread: expect.objectContaining({ id: "thread-old", preview: "Restored after close", archived: false }), + }); + expect(display).not.toHaveBeenCalled(); + }); + + it("serializes hook mutations before starting the next short-lived client operation", async () => { + const firstWrite = deferred(); + const firstRequest = vi.fn(() => firstWrite.promise); + const secondRequest = vi.fn().mockResolvedValue({}); + const firstClient = settingsRequestClient({ + "config/batchWrite": firstRequest, + }); + const secondClient = settingsRequestClient({ + "config/batchWrite": secondRequest, + }); + useShortLivedClients(firstClient, secondClient); + const dynamicData = settingsTabHost().dynamicData; + + const trust = dynamicData.trustHook(hook({ key: "hook-first" })); + const toggle = dynamicData.setHookEnabled(hook({ key: "hook-second" }), false); + await flushPromises(); + + expect(firstRequest).toHaveBeenCalledOnce(); + expect(secondRequest).not.toHaveBeenCalled(); + expect(withShortLivedAppServerClientMock).toHaveBeenCalledOnce(); + + firstWrite.resolve({}); + await trust; + await toggle; + + expect(secondRequest).toHaveBeenCalledOnce(); + expect(withShortLivedAppServerClientMock).toHaveBeenCalledTimes(2); + }); + it("records restored archived threads in the active catalog", async () => { const notify = vi.fn(); const applyThreadCatalogEvent = vi.fn(); diff --git a/tests/settings/test-support.ts b/tests/settings/test-support.ts index 3bae17f7..22c428b4 100644 --- a/tests/settings/test-support.ts +++ b/tests/settings/test-support.ts @@ -214,6 +214,7 @@ export function settingsTabHost(options: SettingsTabHostOptions = {}): CodexPane sendShortcut: options.sendShortcut ?? "enter", }; const appServerQueries = { + contextKey: () => settings.codexPath, modelsSnapshot: vi.fn(() => options.modelsSnapshot ?? []), fetchModels: options.fetchModels ?? vi.fn().mockResolvedValue(options.modelsSnapshot ?? []), refreshModels: options.refreshModels ?? vi.fn().mockResolvedValue(options.modelsSnapshot ?? []),