From cb172edfb7325b6a5ee09bd5d515e0cdcd6edc19 Mon Sep 17 00:00:00 2001 From: murashit Date: Thu, 16 Jul 2026 21:44:28 +0900 Subject: [PATCH] refactor(app-server): own context leases in resource store --- .../connection/connection-manager.ts | 16 +- src/app-server/query/cache.ts | 58 ++-- src/app-server/query/keys.ts | 61 +++- src/app-server/query/resource-store.ts | 260 ++++++++++++++++++ src/app-server/query/shared-queries.ts | 205 -------------- .../chat/app-server/inbound/handler.ts | 10 +- .../connection/connection-actions.ts | 10 +- .../connection/server-metadata-actions.ts | 6 +- .../threads/auto-title-coordinator.ts | 21 +- .../threads/rename-editor-actions.ts | 5 + .../chat/host/bundles/connection-bundle.ts | 9 +- .../chat/host/bundles/thread-bundle.ts | 6 +- src/features/chat/host/contracts.ts | 5 +- src/features/chat/host/session-runtime.ts | 18 +- src/features/chat/host/session.ts | 22 +- src/features/threads-view/session.ts | 8 +- .../threads/app-server/workflow-transports.ts | 16 +- .../threads/catalog/thread-catalog.ts | 38 +-- src/features/threads/workflows/ports.ts | 2 +- .../threads/workflows/thread-title-service.ts | 35 ++- src/main.ts | 5 +- src/plugin-runtime.ts | 78 ++++-- src/settings/app-server-dynamic-data.ts | 8 +- src/settings/dynamic-data.ts | 1 - src/settings/host.ts | 4 +- src/settings/tab.obsidian.tsx | 30 +- tests/app-server/connection-manager.test.ts | 30 +- tests/app-server/query-cache.test.ts | 27 +- tests/app-server/resource-store.test.ts | 139 ++++++++++ tests/app-server/shared-queries.test.ts | 204 -------------- .../chat/app-server/inbound/handler.test.ts | 2 +- .../connection/connection-actions.test.ts | 8 +- .../connection/server-actions.test.ts | 18 +- .../threads/auto-title-coordinator.test.ts | 52 +++- .../threads/rename-editor-actions.test.ts | 20 ++ .../chat/host/session-runtime.test.ts | 15 +- .../chat/host/view-connection.test.ts | 1 + tests/features/threads-view/view.test.ts | 2 + .../threads/catalog/thread-catalog.test.ts | 157 ++--------- .../workflows/thread-title-service.test.ts | 32 ++- tests/main.test.ts | 62 +++-- tests/settings/settings-tab.test.ts | 26 +- 42 files changed, 937 insertions(+), 795 deletions(-) create mode 100644 src/app-server/query/resource-store.ts delete mode 100644 src/app-server/query/shared-queries.ts create mode 100644 tests/app-server/resource-store.test.ts delete mode 100644 tests/app-server/shared-queries.test.ts diff --git a/src/app-server/connection/connection-manager.ts b/src/app-server/connection/connection-manager.ts index 9fd66549..d1ef7594 100644 --- a/src/app-server/connection/connection-manager.ts +++ b/src/app-server/connection/connection-manager.ts @@ -16,6 +16,7 @@ export type AppServerClientFactory = (codexPath: string, cwd: string, handlers: export interface AppServerConnectionContext { readonly codexPath: string; readonly cwd: string; + readonly generation: number; } type ConnectionLifecycleState = @@ -23,12 +24,13 @@ type ConnectionLifecycleState = | { kind: "connecting"; generation: number; + contextGeneration: number; codexPath: string; cwd: string; client: AppServerClient; promise: Promise; } - | { kind: "connected"; generation: number; codexPath: string; cwd: string; client: AppServerClient } + | { kind: "connected"; generation: number; contextGeneration: number; codexPath: string; cwd: string; client: AppServerClient } | { kind: "disconnected"; generation: number }; export class StaleConnectionError extends Error { @@ -47,6 +49,7 @@ export class ConnectionManager { initializeParams: InitializeParams, private readonly clientFactory: AppServerClientFactory = (codexPath, cwd, handlers) => new AppServerClient({ codexPath, cwd, handlers, initializeParams }), + private readonly contextGeneration: () => number = () => 0, ) {} currentClient(): AppServerClient | null { @@ -59,7 +62,7 @@ export class ConnectionManager { if (this.state.kind !== "connected" || !this.state.client.isConnected() || !this.stateMatchesCurrentContext(this.state)) { return null; } - return { codexPath: this.state.codexPath, cwd: this.state.cwd }; + return { codexPath: this.state.codexPath, cwd: this.state.cwd, generation: this.state.contextGeneration }; } isConnected(): boolean { @@ -75,9 +78,10 @@ export class ConnectionManager { if (this.state.kind === "connecting" || this.state.kind === "connected") this.disconnect(); const generation = this.state.generation + 1; + const contextGeneration = this.contextGeneration(); const codexPath = this.codexPath(); const cwd = this.cwd; - const context = { codexPath, cwd }; + const context = { codexPath, cwd, generation: contextGeneration }; const client = this.clientFactory(codexPath, cwd, { onNotification: (notification) => { if (this.isStale(generation)) return; @@ -104,7 +108,7 @@ export class ConnectionManager { client.disconnect(); throw new StaleConnectionError(); } - this.state = { kind: "connected", generation, codexPath, cwd, client }; + this.state = { kind: "connected", generation, contextGeneration, codexPath, cwd, client }; return serverInitializationFromResponse(response); }) .catch((error: unknown) => { @@ -118,7 +122,7 @@ export class ConnectionManager { throw error; }); - this.state = { kind: "connecting", generation, codexPath, cwd, client, promise }; + this.state = { kind: "connecting", generation, contextGeneration, codexPath, cwd, client, promise }; return promise; } @@ -139,7 +143,7 @@ export class ConnectionManager { } private stateMatchesCurrentContext(state: Extract): boolean { - return state.codexPath === this.codexPath() && state.cwd === this.cwd; + return state.codexPath === this.codexPath() && state.cwd === this.cwd && state.contextGeneration === this.contextGeneration(); } } diff --git a/src/app-server/query/cache.ts b/src/app-server/query/cache.ts index 98d775c5..f0811e0f 100644 --- a/src/app-server/query/cache.ts +++ b/src/app-server/query/cache.ts @@ -19,17 +19,17 @@ import { listModelMetadata } from "../services/catalog"; import { readEffectiveConfig } from "../services/runtime-metadata"; import { listThreads, readThreadPage } from "../services/threads"; import { - type AppServerQueryContext, + type AppServerQueryContextIdentity as AppServerQueryContext, activeThreadsQueryKey, appServerModelsQueryKey, appServerPermissionProfilesQueryKey, + appServerQueryContextIdentityKey, appServerQueryContextIsComplete, - appServerQueryContextKey, appServerRateLimitsQueryKey, appServerRuntimeConfigQueryKey, appServerSkillsQueryKey, archivedThreadsQueryKey, - cloneAppServerQueryContext, + cloneAppServerQueryContextIdentity, } from "./keys"; import { readPermissionProfileMetadataProbe, readRateLimitMetadataProbe, readSkillMetadataProbe } from "./metadata-probes"; import type { ObservedResult, ObservedResultListener } from "./observed-result"; @@ -98,6 +98,22 @@ export class AppServerQueryCache { this.client.clear(); } + release(context: AppServerQueryContext): void { + this.generation += 1; + const identityKey = appServerQueryContextIdentityKey(context); + const activeThreadsKey = JSON.stringify(activeThreadsQueryKey(context)); + this.activeThreadCursors.delete(activeThreadsKey); + this.activeThreadRevisions.delete(activeThreadsKey); + this.metadataRefreshes.delete(identityKey); + this.metadataProjectionListeners.delete(identityKey); + for (const resource of ["skills", "permissionProfiles", "rateLimits"] as const) { + const resourceKey = this.metadataResourceKey(context, resource); + this.metadataResourceFetches.delete(resourceKey); + this.metadataNotificationRefreshes.delete(resourceKey); + } + this.client.removeQueries({ queryKey: ["app-server", context.generation, context.codexPath, context.vaultPath] }); + } + activeThreadsSnapshot(context: AppServerQueryContext): readonly Thread[] | null { return this.threadListSnapshot(context, "active"); } @@ -135,7 +151,7 @@ export class AppServerQueryCache { } async fetchAllActiveThreads(context: AppServerQueryContext): Promise { - const refreshContext = cloneAppServerQueryContext(context); + const refreshContext = cloneAppServerQueryContextIdentity(context); if (!appServerQueryContextIsComplete(refreshContext)) return []; const cursorKey = this.activeThreadCursorKey(refreshContext); const snapshot = this.activeThreadsSnapshot(refreshContext); @@ -157,7 +173,7 @@ export class AppServerQueryCache { } async loadMoreActiveThreads(context: AppServerQueryContext): Promise { - const refreshContext = cloneAppServerQueryContext(context); + const refreshContext = cloneAppServerQueryContextIdentity(context); if (!appServerQueryContextIsComplete(refreshContext)) return []; const current = this.activeThreadsSnapshot(refreshContext) ?? (await this.fetchActiveThreads(refreshContext)); const cursor = this.activeThreadCursors.get(this.activeThreadCursorKey(refreshContext)) ?? null; @@ -191,7 +207,7 @@ export class AppServerQueryCache { kind: ThreadListKind, options: { force?: boolean } = {}, ): Promise { - const refreshContext = cloneAppServerQueryContext(context); + const refreshContext = cloneAppServerQueryContextIdentity(context); if (!appServerQueryContextIsComplete(refreshContext)) { return []; } @@ -232,7 +248,7 @@ export class AppServerQueryCache { listener: ObservedResultListener, options: { emitCurrent?: boolean } = {}, ): () => void { - const refreshContext = cloneAppServerQueryContext(context); + const refreshContext = cloneAppServerQueryContextIdentity(context); const generation = this.generation; const observers = [ new QueryObserver(this.client, { ...this.runtimeConfigQueryOptions(refreshContext), enabled: false }), @@ -256,11 +272,11 @@ export class AppServerQueryCache { }; const schedule = (): void => { if (disposed || generation !== this.generation) return; - if ((this.metadataRefreshes.get(appServerQueryContextKey(refreshContext)) ?? 0) > 0 || queued) return; + if ((this.metadataRefreshes.get(appServerQueryContextIdentityKey(refreshContext)) ?? 0) > 0 || queued) return; queued = true; queueMicrotask(emit); }; - const contextKey = appServerQueryContextKey(refreshContext); + const contextKey = appServerQueryContextIdentityKey(refreshContext); const projectionListeners = this.metadataProjectionListeners.get(contextKey) ?? new Set<() => void>(); projectionListeners.add(schedule); this.metadataProjectionListeners.set(contextKey, projectionListeners); @@ -275,7 +291,7 @@ export class AppServerQueryCache { } async refreshAppServerMetadata(context: AppServerQueryContext): Promise { - const refreshContext = cloneAppServerQueryContext(context); + const refreshContext = cloneAppServerQueryContextIdentity(context); if (!appServerQueryContextIsComplete(refreshContext)) return null; return this.runMetadataRefresh(refreshContext, async () => { const runtimeResult = this.fetchRuntimeConfig(refreshContext, true).then( @@ -297,7 +313,7 @@ export class AppServerQueryCache { } async refreshSkills(context: AppServerQueryContext): Promise { - const refreshContext = cloneAppServerQueryContext(context); + const refreshContext = cloneAppServerQueryContextIdentity(context); if (!appServerQueryContextIsComplete(refreshContext)) return null; return this.runMetadataRefresh(refreshContext, async () => { const current = await this.refreshMetadataResourceNotification(refreshContext, "skills"); @@ -306,7 +322,7 @@ export class AppServerQueryCache { } async refreshRateLimits(context: AppServerQueryContext): Promise { - const refreshContext = cloneAppServerQueryContext(context); + const refreshContext = cloneAppServerQueryContextIdentity(context); if (!appServerQueryContextIsComplete(refreshContext)) return null; return this.runMetadataRefresh(refreshContext, async () => { const current = await this.refreshMetadataResourceNotification(refreshContext, "rateLimits"); @@ -329,7 +345,7 @@ export class AppServerQueryCache { } async fetchModels(context: AppServerQueryContext, options: { force?: boolean } = {}): Promise { - const refreshContext = cloneAppServerQueryContext(context); + const refreshContext = cloneAppServerQueryContextIdentity(context); if (!appServerQueryContextIsComplete(refreshContext)) { return []; } @@ -344,7 +360,7 @@ export class AppServerQueryCache { } private threadListQueryOptions(context: AppServerQueryContext, kind: ThreadListKind): AppServerQueryOptions { - const refreshContext = cloneAppServerQueryContext(context); + const refreshContext = cloneAppServerQueryContextIdentity(context); return { queryKey: this.threadListQueryKey(refreshContext, kind), queryFn: async (): Promise => { @@ -373,7 +389,7 @@ export class AppServerQueryCache { } private runtimeConfigQueryOptions(context: AppServerQueryContext): AppServerQueryOptions { - const refreshContext = cloneAppServerQueryContext(context); + const refreshContext = cloneAppServerQueryContextIdentity(context); return { queryKey: appServerRuntimeConfigQueryKey(refreshContext), queryFn: async (): Promise => @@ -388,7 +404,7 @@ export class AppServerQueryCache { context: AppServerQueryContext, forceReload = false, ): AppServerQueryOptions> { - const refreshContext = cloneAppServerQueryContext(context); + const refreshContext = cloneAppServerQueryContextIdentity(context); return { queryKey: appServerSkillsQueryKey(refreshContext), queryFn: async () => @@ -402,7 +418,7 @@ export class AppServerQueryCache { private permissionProfilesQueryOptions( context: AppServerQueryContext, ): AppServerQueryOptions> { - const refreshContext = cloneAppServerQueryContext(context); + const refreshContext = cloneAppServerQueryContextIdentity(context); return { queryKey: appServerPermissionProfilesQueryKey(refreshContext), queryFn: async () => @@ -416,7 +432,7 @@ export class AppServerQueryCache { private rateLimitsQueryOptions( context: AppServerQueryContext, ): AppServerQueryOptions> { - const refreshContext = cloneAppServerQueryContext(context); + const refreshContext = cloneAppServerQueryContextIdentity(context); return { queryKey: appServerRateLimitsQueryKey(refreshContext), queryFn: async () => @@ -503,7 +519,7 @@ export class AppServerQueryCache { } private metadataResourceKey(context: AppServerQueryContext, resource: MetadataResourceKind): string { - return `${appServerQueryContextKey(context)}\u0000${resource}`; + return `${appServerQueryContextIdentityKey(context)}\u0000${resource}`; } private metadataResourceState( @@ -547,7 +563,7 @@ export class AppServerQueryCache { } private async runMetadataRefresh(context: AppServerQueryContext, operation: () => Promise): Promise { - const key = appServerQueryContextKey(context); + const key = appServerQueryContextIdentityKey(context); this.metadataRefreshes.set(key, (this.metadataRefreshes.get(key) ?? 0) + 1); try { return await operation(); @@ -563,7 +579,7 @@ export class AppServerQueryCache { } private modelsQueryOptions(context: AppServerQueryContext): AppServerQueryOptions { - const refreshContext = cloneAppServerQueryContext(context); + const refreshContext = cloneAppServerQueryContextIdentity(context); return { queryKey: appServerModelsQueryKey(refreshContext), queryFn: async (): Promise => { diff --git a/src/app-server/query/keys.ts b/src/app-server/query/keys.ts index a2254c89..b05c86bc 100644 --- a/src/app-server/query/keys.ts +++ b/src/app-server/query/keys.ts @@ -3,7 +3,16 @@ export interface AppServerQueryContext { vaultPath: string; } -type AppServerQueryScope = readonly ["app-server", string, string]; +export interface AppServerContextLease { + readonly context: Readonly; + readonly generation: number; +} + +export interface AppServerQueryContextIdentity extends AppServerQueryContext { + readonly generation: number; +} + +type AppServerQueryScope = readonly ["app-server", number, string, string]; export type AppServerActiveThreadsQueryKey = readonly [...AppServerQueryScope, "threads", "active"]; export type AppServerArchivedThreadsQueryKey = readonly [...AppServerQueryScope, "threads", "archived"]; export type AppServerModelsQueryKey = readonly [...AppServerQueryScope, "models"]; @@ -16,51 +25,77 @@ export function appServerQueryContextIsComplete(context: AppServerQueryContext): return nonEmptyString(context.codexPath) && nonEmptyString(context.vaultPath); } -export function cloneAppServerQueryContext(context: AppServerQueryContext): AppServerQueryContext { +function cloneAppServerQueryContext(context: AppServerQueryContext): AppServerQueryContext { return { ...context }; } +export function createAppServerContextLease(context: AppServerQueryContext, generation: number): AppServerContextLease { + return Object.freeze({ + context: Object.freeze(cloneAppServerQueryContext(context)), + generation, + }); +} + +export function appServerQueryContextIdentity(lease: AppServerContextLease): AppServerQueryContextIdentity { + return { + ...lease.context, + generation: lease.generation, + }; +} + +export function cloneAppServerQueryContextIdentity(identity: AppServerQueryContextIdentity): AppServerQueryContextIdentity { + return { ...identity }; +} + export function appServerQueryContextRawEquals(left: AppServerQueryContext, right: AppServerQueryContext): boolean { return left.codexPath === right.codexPath && left.vaultPath === right.vaultPath; } -export function appServerQueryContextMatches(left: AppServerQueryContext, right: AppServerQueryContext): boolean { +function appServerQueryContextMatches(left: AppServerQueryContext, right: AppServerQueryContext): boolean { return appServerQueryContextIsComplete(left) && appServerQueryContextIsComplete(right) && appServerQueryContextRawEquals(left, right); } -export function appServerQueryContextKey(context: AppServerQueryContext): string { +function appServerQueryContextKey(context: AppServerQueryContext): string { return `${context.codexPath}\u0000${context.vaultPath}`; } -function appServerQueryScope(context: AppServerQueryContext): AppServerQueryScope { - return ["app-server", context.codexPath, context.vaultPath]; +export function appServerQueryContextIdentityMatches(left: AppServerQueryContextIdentity, right: AppServerQueryContextIdentity): boolean { + return left.generation === right.generation && appServerQueryContextMatches(left, right); } -export function activeThreadsQueryKey(context: AppServerQueryContext): AppServerActiveThreadsQueryKey { +export function appServerQueryContextIdentityKey(identity: AppServerQueryContextIdentity): string { + return `${String(identity.generation)}\u0000${appServerQueryContextKey(identity)}`; +} + +function appServerQueryScope(context: AppServerQueryContextIdentity): AppServerQueryScope { + return ["app-server", context.generation, context.codexPath, context.vaultPath]; +} + +export function activeThreadsQueryKey(context: AppServerQueryContextIdentity): AppServerActiveThreadsQueryKey { return [...appServerQueryScope(context), "threads", "active"]; } -export function archivedThreadsQueryKey(context: AppServerQueryContext): AppServerArchivedThreadsQueryKey { +export function archivedThreadsQueryKey(context: AppServerQueryContextIdentity): AppServerArchivedThreadsQueryKey { return [...appServerQueryScope(context), "threads", "archived"]; } -export function appServerModelsQueryKey(context: AppServerQueryContext): AppServerModelsQueryKey { +export function appServerModelsQueryKey(context: AppServerQueryContextIdentity): AppServerModelsQueryKey { return [...appServerQueryScope(context), "models"]; } -export function appServerRuntimeConfigQueryKey(context: AppServerQueryContext): AppServerRuntimeConfigQueryKey { +export function appServerRuntimeConfigQueryKey(context: AppServerQueryContextIdentity): AppServerRuntimeConfigQueryKey { return [...appServerQueryScope(context), "runtime-config"]; } -export function appServerSkillsQueryKey(context: AppServerQueryContext): AppServerSkillsQueryKey { +export function appServerSkillsQueryKey(context: AppServerQueryContextIdentity): AppServerSkillsQueryKey { return [...appServerQueryScope(context), "skills"]; } -export function appServerPermissionProfilesQueryKey(context: AppServerQueryContext): AppServerPermissionProfilesQueryKey { +export function appServerPermissionProfilesQueryKey(context: AppServerQueryContextIdentity): AppServerPermissionProfilesQueryKey { return [...appServerQueryScope(context), "permission-profiles"]; } -export function appServerRateLimitsQueryKey(context: AppServerQueryContext): AppServerRateLimitsQueryKey { +export function appServerRateLimitsQueryKey(context: AppServerQueryContextIdentity): AppServerRateLimitsQueryKey { return [...appServerQueryScope(context), "rate-limits"]; } diff --git a/src/app-server/query/resource-store.ts b/src/app-server/query/resource-store.ts new file mode 100644 index 00000000..5ae65d88 --- /dev/null +++ b/src/app-server/query/resource-store.ts @@ -0,0 +1,260 @@ +import type { ModelMetadata } from "../../domain/catalog/metadata"; +import type { SharedServerMetadata } from "../../domain/server/metadata"; +import type { Thread } from "../../domain/threads/model"; +import { AppServerQueryCache, type AppServerQueryClientRunner } from "./cache"; +import { + type AppServerContextLease, + type AppServerQueryContext, + type AppServerQueryContextIdentity, + appServerQueryContextIdentity, + appServerQueryContextIdentityKey, + appServerQueryContextIdentityMatches, + appServerQueryContextRawEquals, + createAppServerContextLease, +} from "./keys"; +import type { ObservedResultListener } from "./observed-result"; + +export interface AppServerResourceStoreOptions { + cache?: AppServerQueryCache; + clientRunner?: AppServerQueryClientRunner; +} + +export class StaleAppServerResourceContextError extends Error { + constructor() { + super("Codex app-server resource context changed while loading."); + this.name = "StaleAppServerResourceContextError"; + } +} + +export function isStaleAppServerResourceContextError(error: unknown): error is StaleAppServerResourceContextError { + return error instanceof StaleAppServerResourceContextError; +} + +interface ResourceObserver { + readonly observe: ( + context: AppServerQueryContextIdentity, + listener: (value: T) => void, + options: { emitCurrent?: boolean }, + ) => () => void; + readonly listener: (value: T) => void; + readonly options: { emitCurrent?: boolean }; + firstSubscribe: boolean; + unsubscribeQuery: (() => void) | null; +} + +export class AppServerResourceStore { + private readonly observers = new Set>(); + private readonly cache: AppServerQueryCache; + private lease: AppServerContextLease | null = null; + private generation = 0; + + constructor(options: AppServerResourceStoreOptions = {}) { + this.cache = options.cache ?? new AppServerQueryCache(options.clientRunner ? { clientRunner: options.clientRunner } : {}); + } + + initialize(context: AppServerQueryContext): AppServerContextLease { + if (this.lease) throw new Error("Codex app-server resource store is already initialized."); + this.lease = this.nextLease(context); + this.rebindObservers(); + return this.lease; + } + + replaceContext(context: AppServerQueryContext): AppServerContextLease { + if (!this.lease) throw new Error("Codex app-server resource store is not initialized."); + if (appServerQueryContextRawEquals(this.lease.context, context)) return this.lease; + this.unsubscribeObservers(); + this.cache.release(appServerQueryContextIdentity(this.lease)); + this.lease = this.nextLease(context); + this.rebindObservers(); + return this.lease; + } + + reset(): void { + this.unsubscribeObservers(); + if (this.lease) this.cache.release(appServerQueryContextIdentity(this.lease)); + this.lease = null; + this.cache.clear(); + } + + contextLease(): AppServerContextLease { + if (!this.lease) throw new Error("Codex app-server resource store is not initialized."); + return this.lease; + } + + contextIdentity(): AppServerQueryContextIdentity { + return appServerQueryContextIdentity(this.contextLease()); + } + + contextKey(): string { + return appServerQueryContextIdentityKey(this.contextIdentity()); + } + + contextKeyFor(context: AppServerQueryContextIdentity): string { + return appServerQueryContextIdentityKey(context); + } + + activeThreadsSnapshot(): readonly Thread[] | null { + return this.lease ? this.cache.activeThreadsSnapshot(this.contextIdentity()) : null; + } + + archivedThreadsSnapshot(): readonly Thread[] | null { + return this.lease ? this.cache.archivedThreadsSnapshot(this.contextIdentity()) : null; + } + + fetchAllActiveThreads(): Promise { + return this.runForCurrentContext((context) => this.cache.fetchAllActiveThreads(context)); + } + + hasMoreActiveThreads(): boolean { + return this.lease ? this.cache.hasMoreActiveThreads(this.contextIdentity()) : false; + } + + loadMoreActiveThreads(): Promise { + return this.runForCurrentContext((context) => this.cache.loadMoreActiveThreads(context)); + } + + refreshActiveThreads(): Promise { + return this.runForCurrentContext((context) => this.cache.refreshActiveThreads(context)); + } + + refreshArchivedThreads(): Promise { + return this.runForCurrentContext((context) => this.cache.refreshArchivedThreads(context)); + } + + observeActiveThreadsResult(listener: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void { + return this.observeCurrentContext( + (context, contextListener, observeOptions) => this.cache.observeActiveThreadsResult(context, contextListener, observeOptions), + listener, + options, + ); + } + + observeArchivedThreadsResult(listener: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void { + return this.observeCurrentContext( + (context, contextListener, observeOptions) => this.cache.observeArchivedThreadsResult(context, contextListener, observeOptions), + listener, + options, + ); + } + + appServerMetadataSnapshot(): SharedServerMetadata | null { + return this.lease ? this.cache.appServerMetadataSnapshot(this.contextIdentity()) : null; + } + + refreshAppServerMetadata(): Promise { + return this.runForCurrentContext((context) => this.cache.refreshAppServerMetadata(context)); + } + + refreshSkills(): Promise { + return this.runForCurrentContext((context) => this.cache.refreshSkills(context)); + } + + refreshRateLimits(): Promise { + return this.runForCurrentContext((context) => this.cache.refreshRateLimits(context)); + } + + observeAppServerMetadataResult(listener: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void { + return this.observeCurrentContext( + (context, contextListener, observeOptions) => this.cache.observeAppServerMetadataResult(context, contextListener, observeOptions), + listener, + options, + ); + } + + modelsSnapshot(): readonly ModelMetadata[] | null { + return this.lease ? this.cache.modelsSnapshot(this.contextIdentity()) : null; + } + + fetchModels(): Promise { + return this.runForCurrentContext((context) => this.cache.fetchModels(context)); + } + + refreshModels(): Promise { + return this.runForCurrentContext((context) => this.cache.refreshModels(context)); + } + + observeModelsResult(listener: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void { + return this.observeCurrentContext( + (context, contextListener, observeOptions) => this.cache.observeModelsResult(context, contextListener, observeOptions), + listener, + options, + ); + } + + private nextLease(context: AppServerQueryContext): AppServerContextLease { + return createAppServerContextLease(context, ++this.generation); + } + + private isCurrent(context: AppServerQueryContextIdentity): boolean { + return Boolean(this.lease && appServerQueryContextIdentityMatches(this.contextIdentity(), context)); + } + + private runForCurrentContext(operation: (context: AppServerQueryContextIdentity) => Promise): Promise { + const context = this.contextIdentity(); + return this.runForIdentity(context, () => operation(context)); + } + + private async runForIdentity(context: AppServerQueryContextIdentity, operation: () => Promise): Promise { + if (!this.isCurrent(context)) throw new StaleAppServerResourceContextError(); + let result: T; + try { + result = await operation(); + } catch (error) { + if (!this.isCurrent(context)) throw new StaleAppServerResourceContextError(); + throw error; + } + if (!this.isCurrent(context)) throw new StaleAppServerResourceContextError(); + return result; + } + + private observeCurrentContext( + observe: ResourceObserver["observe"], + listener: ResourceObserver["listener"], + options: { emitCurrent?: boolean } = {}, + ): () => void { + const observer: ResourceObserver = { + observe, + listener, + options, + firstSubscribe: true, + unsubscribeQuery: null, + }; + this.observers.add(observer as ResourceObserver); + this.bindObserver(observer); + return () => { + this.observers.delete(observer as ResourceObserver); + observer.unsubscribeQuery?.(); + observer.unsubscribeQuery = null; + }; + } + + private bindObserver(observer: ResourceObserver): void { + if (!this.lease) return; + const context = this.contextIdentity(); + const observeOptions: { emitCurrent?: boolean } = {}; + if (observer.firstSubscribe) { + if (observer.options.emitCurrent !== undefined) observeOptions.emitCurrent = observer.options.emitCurrent; + } else { + observeOptions.emitCurrent = true; + } + observer.unsubscribeQuery = observer.observe( + context, + (value) => { + if (this.isCurrent(context)) observer.listener(value); + }, + observeOptions, + ); + observer.firstSubscribe = false; + } + + private unsubscribeObservers(): void { + for (const observer of this.observers) { + observer.unsubscribeQuery?.(); + observer.unsubscribeQuery = null; + } + } + + private rebindObservers(): void { + for (const observer of this.observers) this.bindObserver(observer); + } +} diff --git a/src/app-server/query/shared-queries.ts b/src/app-server/query/shared-queries.ts deleted file mode 100644 index 98906961..00000000 --- a/src/app-server/query/shared-queries.ts +++ /dev/null @@ -1,205 +0,0 @@ -import type { ModelMetadata } from "../../domain/catalog/metadata"; -import type { SharedServerMetadata } from "../../domain/server/metadata"; -import type { Thread } from "../../domain/threads/model"; -import type { AppServerQueryCache } from "./cache"; -import { - type AppServerQueryContext, - appServerQueryContextKey, - appServerQueryContextMatches, - appServerQueryContextRawEquals, - cloneAppServerQueryContext, -} from "./keys"; -import type { ObservedResultListener } from "./observed-result"; - -export interface AppServerSharedQueriesOptions { - cache: AppServerQueryCache; - context: () => AppServerQueryContext; -} - -export class StaleAppServerSharedQueryContextError extends Error { - constructor() { - super("Codex app-server query context changed while loading shared queries."); - this.name = "StaleAppServerSharedQueryContextError"; - } -} - -export function isStaleAppServerSharedQueryContextError(error: unknown): error is StaleAppServerSharedQueryContextError { - return error instanceof StaleAppServerSharedQueryContextError; -} - -export class AppServerSharedQueries { - private readonly contextChangeListeners = new Set<() => void>(); - - constructor(private readonly options: AppServerSharedQueriesOptions) {} - - contextKey(): string { - return appServerQueryContextKey(this.context()); - } - - contextKeyFor(context: AppServerQueryContext): string { - return appServerQueryContextKey(context); - } - - activeThreadsSnapshot(): readonly Thread[] | null { - return this.options.cache.activeThreadsSnapshot(this.context()); - } - - activeThreadsSnapshotFor(context: AppServerQueryContext): readonly Thread[] | null { - return this.options.cache.activeThreadsSnapshot(context); - } - - archivedThreadsSnapshot(): readonly Thread[] | null { - return this.options.cache.archivedThreadsSnapshot(this.context()); - } - - archivedThreadsSnapshotFor(context: AppServerQueryContext): readonly Thread[] | null { - return this.options.cache.archivedThreadsSnapshot(context); - } - - fetchAllActiveThreads(): Promise { - return this.runForCurrentContext((context) => this.options.cache.fetchAllActiveThreads(context)); - } - - hasMoreActiveThreads(): boolean { - return this.options.cache.hasMoreActiveThreads(this.context()); - } - - loadMoreActiveThreads(): Promise { - return this.runForCurrentContext((context) => this.options.cache.loadMoreActiveThreads(context)); - } - - refreshActiveThreads(): Promise { - return this.runForCurrentContext((context) => this.options.cache.refreshActiveThreads(context)); - } - - refreshActiveThreadsFor(context: AppServerQueryContext): Promise { - return this.options.cache.refreshActiveThreads(context); - } - - refreshArchivedThreads(): Promise { - return this.runForCurrentContext((context) => this.options.cache.refreshArchivedThreads(context)); - } - - refreshArchivedThreadsFor(context: AppServerQueryContext): Promise { - return this.options.cache.refreshArchivedThreads(context); - } - - observeActiveThreadsResult(listener: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void { - return this.observeCurrentContext( - (context, contextListener, observeOptions) => this.options.cache.observeActiveThreadsResult(context, contextListener, observeOptions), - listener, - options, - ); - } - - observeArchivedThreadsResult(listener: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void { - return this.observeCurrentContext( - (context, contextListener, observeOptions) => - this.options.cache.observeArchivedThreadsResult(context, contextListener, observeOptions), - listener, - options, - ); - } - - appServerMetadataSnapshot(): SharedServerMetadata | null { - return this.options.cache.appServerMetadataSnapshot(this.context()); - } - - refreshAppServerMetadata(): Promise { - return this.runForCurrentContext((context) => this.options.cache.refreshAppServerMetadata(context)); - } - - refreshSkills(): Promise { - return this.runForCurrentContext((context) => this.options.cache.refreshSkills(context)); - } - - refreshRateLimits(): Promise { - return this.runForCurrentContext((context) => this.options.cache.refreshRateLimits(context)); - } - - observeAppServerMetadataResult(listener: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void { - return this.observeCurrentContext( - (context, contextListener, observeOptions) => - this.options.cache.observeAppServerMetadataResult(context, contextListener, observeOptions), - listener, - options, - ); - } - - modelsSnapshot(): readonly ModelMetadata[] | null { - return this.options.cache.modelsSnapshot(this.context()); - } - - fetchModels(): Promise { - return this.runForCurrentContext((context) => this.options.cache.fetchModels(context)); - } - - refreshModels(): Promise { - return this.runForCurrentContext((context) => this.options.cache.refreshModels(context)); - } - - observeModelsResult(listener: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void { - return this.observeCurrentContext( - (context, contextListener, observeOptions) => this.options.cache.observeModelsResult(context, contextListener, observeOptions), - listener, - options, - ); - } - - notifyContextChanged(): void { - for (const listener of [...this.contextChangeListeners]) { - listener(); - } - } - - private context(): AppServerQueryContext { - return this.options.context(); - } - - private async runForCurrentContext(operation: (context: AppServerQueryContext) => Promise): Promise { - const context = cloneAppServerQueryContext(this.context()); - const result = await operation(context); - if (!appServerQueryContextRawEquals(this.context(), context)) { - throw new StaleAppServerSharedQueryContextError(); - } - return result; - } - - private observeCurrentContext( - observe: (context: AppServerQueryContext, listener: (value: T) => void, options: { emitCurrent?: boolean }) => () => void, - listener: (value: T) => void, - options: { emitCurrent?: boolean } = {}, - ): () => void { - let observedContext: AppServerQueryContext | null = null; - let unsubscribeQuery: (() => void) | null = null; - let firstSubscribe = true; - - const subscribe = (): void => { - const context = cloneAppServerQueryContext(this.context()); - if (observedContext && appServerQueryContextMatches(observedContext, context)) return; - unsubscribeQuery?.(); - observedContext = context; - const observeOptions: { emitCurrent?: boolean } = {}; - if (firstSubscribe) { - if (options.emitCurrent !== undefined) observeOptions.emitCurrent = options.emitCurrent; - } else { - observeOptions.emitCurrent = true; - } - unsubscribeQuery = observe( - context, - (value) => { - if (appServerQueryContextMatches(this.context(), context)) listener(value); - }, - observeOptions, - ); - firstSubscribe = false; - }; - - subscribe(); - this.contextChangeListeners.add(subscribe); - return () => { - this.contextChangeListeners.delete(subscribe); - unsubscribeQuery?.(); - }; - } -} diff --git a/src/features/chat/app-server/inbound/handler.ts b/src/features/chat/app-server/inbound/handler.ts index b3866867..d7a859aa 100644 --- a/src/features/chat/app-server/inbound/handler.ts +++ b/src/features/chat/app-server/inbound/handler.ts @@ -1,5 +1,5 @@ import type { RequestId, ServerNotification, ServerRequest } from "../../../../app-server/connection/rpc-messages"; -import type { AppServerQueryContext } from "../../../../app-server/query/keys"; +import type { AppServerQueryContextIdentity } from "../../../../app-server/query/keys"; import { routeServerRequest, serverRequestApprovalResponse, @@ -36,13 +36,13 @@ export interface ChatInboundHandlerActions { refreshServerDiagnostics: (options?: { forceResourceProbes?: boolean }) => void; applyAppServerResourceEvent: (event: AppServerResourceEvent) => void; maybeNameThread: (threadId: string, turnId: string, completedTurnTranscriptSummary: TurnTranscriptSummary | null) => void; - applyThreadCatalogEvent: (event: ThreadCatalogEvent, sourceContext: AppServerQueryContext) => void; + applyThreadCatalogEvent: (event: ThreadCatalogEvent, sourceContext: AppServerQueryContextIdentity) => void; respondToServerRequest: (requestId: RequestId, result: unknown) => boolean; rejectServerRequest: (requestId: RequestId, code: number, message: string) => boolean; } export interface ChatInboundHandler { - handleNotification(notification: ServerNotification, sourceContext: AppServerQueryContext): void; + handleNotification(notification: ServerNotification, sourceContext: AppServerQueryContextIdentity): void; handleServerRequest(request: ServerRequest): void; handleAppServerLog(message: string): void; resolveApproval(requestId: PendingRequestId, action: ApprovalAction): void; @@ -111,7 +111,7 @@ function dispatch(context: ChatInboundHandlerContext, action: ChatAction): void function handleNotification( context: ChatInboundHandlerContext, notification: ServerNotification, - sourceContext: AppServerQueryContext, + sourceContext: AppServerQueryContextIdentity, ): void { const plan = planChatNotification(state(context), notification, (prefix) => localItemId(context, prefix)); for (const action of plan.actions) dispatch(context, action); @@ -246,7 +246,7 @@ function localItemId(context: ChatInboundHandlerContext, prefix: string): string function runNotificationEffect( context: ChatInboundHandlerContext, effect: ChatNotificationEffect, - sourceContext: AppServerQueryContext, + sourceContext: AppServerQueryContextIdentity, ): void { switch (effect.type) { case "refresh-threads": diff --git a/src/features/chat/application/connection/connection-actions.ts b/src/features/chat/application/connection/connection-actions.ts index c31148c3..3da22626 100644 --- a/src/features/chat/application/connection/connection-actions.ts +++ b/src/features/chat/application/connection/connection-actions.ts @@ -36,7 +36,7 @@ export interface ChatConnectionActionsHost { configuredCommand: () => string; refreshLiveState: () => void; isStaleConnectionError: (error: unknown) => boolean; - isStaleSharedQueryError: (error: unknown) => boolean; + isStaleResourceContextError: (error: unknown) => boolean; notifyConnectionFailed: () => void; } @@ -105,7 +105,7 @@ async function refreshActiveThreads(host: ChatConnectionActionsHost): Promise Promise; refreshSkills: () => Promise; refreshRateLimits: () => Promise; - isStaleSharedQueryError: (error: unknown) => boolean; + isStaleResourceContextError: (error: unknown) => boolean; } export interface ServerMetadataActions { @@ -81,7 +81,7 @@ async function refreshAppServerMetadata(host: ServerMetadataActionsHost): Promis try { metadata = await host.refreshAppServerMetadata(); } catch (error) { - if (host.isStaleSharedQueryError(error)) return null; + if (host.isStaleResourceContextError(error)) return null; throw error; } if (!metadata) return null; @@ -102,7 +102,7 @@ async function refreshMetadataResource( const metadata = await refresh(); if (metadata) applyAppServerMetadata(host, metadata); } catch (error) { - if (!host.isStaleSharedQueryError(error)) throw error; + if (!host.isStaleResourceContextError(error)) throw error; } } diff --git a/src/features/chat/application/threads/auto-title-coordinator.ts b/src/features/chat/application/threads/auto-title-coordinator.ts index 37127909..aef98131 100644 --- a/src/features/chat/application/threads/auto-title-coordinator.ts +++ b/src/features/chat/application/threads/auto-title-coordinator.ts @@ -15,12 +15,14 @@ export interface AutoTitleCoordinatorHost { } export interface AutoTitleCoordinator { + invalidate(): void; resetThreadTurnPresence(hadTurns: boolean): void; maybeAutoTitleThread(threadId: string, turnId: string, completedTurnTranscriptSummary: TurnTranscriptSummary | null): void; } export function createAutoTitleCoordinator(host: AutoTitleCoordinatorHost): AutoTitleCoordinator { let activeThreadHadTurns = false; + let generation = 0; const attemptedThreadIds = new Set(); const inFlightThreadIds = new Set(); @@ -31,23 +33,30 @@ export function createAutoTitleCoordinator(host: AutoTitleCoordinatorHost): Auto const candidate = thread(threadId); return Boolean(candidate && !candidate.name?.trim()); }; - const generateAndSetTitle = async (threadId: string, context: ThreadTitleContext): Promise => { + const generateAndSetTitle = async (threadId: string, context: ThreadTitleContext, operationGeneration: number): Promise => { try { const title = await host.generateTitleFromContext(context); - if (!title || !threadCanReceiveGeneratedTitle(threadId)) return; + if (operationGeneration !== generation || !title || !threadCanReceiveGeneratedTitle(threadId)) return; await host.renameGeneratedTitle(threadId, title, { - shouldStart: () => threadCanReceiveGeneratedTitle(threadId), - shouldPublish: () => threadCanReceiveGeneratedTitle(threadId), + shouldStart: () => operationGeneration === generation && threadCanReceiveGeneratedTitle(threadId), + shouldPublish: () => operationGeneration === generation && threadCanReceiveGeneratedTitle(threadId), }); } catch { // Auto-title is best-effort metadata. Leave the thread preview untouched on failure. } finally { - inFlightThreadIds.delete(threadId); + if (operationGeneration === generation) inFlightThreadIds.delete(threadId); } }; return { + invalidate() { + generation += 1; + activeThreadHadTurns = false; + attemptedThreadIds.clear(); + inFlightThreadIds.clear(); + }, + resetThreadTurnPresence(hadTurns) { activeThreadHadTurns = hadTurns; }, @@ -64,7 +73,7 @@ export function createAutoTitleCoordinator(host: AutoTitleCoordinatorHost): Auto attemptedThreadIds.add(threadId); inFlightThreadIds.add(threadId); - void generateAndSetTitle(threadId, context); + void generateAndSetTitle(threadId, context, generation); }, }; } diff --git a/src/features/chat/application/threads/rename-editor-actions.ts b/src/features/chat/application/threads/rename-editor-actions.ts index 8ac4f30e..c04eaa29 100644 --- a/src/features/chat/application/threads/rename-editor-actions.ts +++ b/src/features/chat/application/threads/rename-editor-actions.ts @@ -20,6 +20,7 @@ export interface ThreadRenameEditorActionsHost { } export interface ThreadRenameEditorActions { + invalidate(): void; editState(threadId: string): RenameEditState | null; isEditing(): boolean; start(threadId: string): void; @@ -33,6 +34,10 @@ export function createThreadRenameEditorActions(host: ThreadRenameEditorActionsH let nextRenameGenerationToken = 1; const action = { + invalidate(): void { + dispatch(host, { type: "ui/rename-cleared" }); + }, + editState(threadId: string): RenameEditState | null { const current = renameState(host); if (current.kind === "idle" || current.threadId !== threadId) return null; diff --git a/src/features/chat/host/bundles/connection-bundle.ts b/src/features/chat/host/bundles/connection-bundle.ts index c44d9fca..1e340da8 100644 --- a/src/features/chat/host/bundles/connection-bundle.ts +++ b/src/features/chat/host/bundles/connection-bundle.ts @@ -2,7 +2,7 @@ import { Notice } from "obsidian"; import type { AppServerClient, AppServerServerRequestResponder } from "../../../../app-server/connection/client"; import { type ConnectionManager, StaleConnectionError } from "../../../../app-server/connection/connection-manager"; -import { isStaleAppServerSharedQueryContextError } from "../../../../app-server/query/shared-queries"; +import { isStaleAppServerResourceContextError } from "../../../../app-server/query/resource-store"; import type { SharedServerMetadata } from "../../../../domain/server/metadata"; import { type ChatInboundHandler, createChatInboundHandler } from "../../app-server/inbound/handler"; import { type ChatConnectionActions, createChatConnectionActions } from "../../application/connection/connection-actions"; @@ -128,7 +128,7 @@ export function createConnectionBundle( refreshAppServerMetadata: () => environment.plugin.appServerQueries.refreshAppServerMetadata(), refreshSkills: () => environment.plugin.appServerQueries.refreshSkills(), refreshRateLimits: () => environment.plugin.appServerQueries.refreshRateLimits(), - isStaleSharedQueryError: isStaleAppServerSharedQueryContextError, + isStaleResourceContextError: isStaleAppServerResourceContextError, }); const serverDiagnostics = createServerDiagnosticsActions({ stateStore, @@ -141,7 +141,7 @@ export function createConnectionBundle( }; const refreshSharedThreadsQuietly = (): void => { void refreshSharedThreads().catch((error: unknown) => { - if (isStaleAppServerSharedQueryContextError(error)) return; + if (isStaleAppServerResourceContextError(error)) return; status.addSystemMessage(error instanceof Error ? error.message : String(error)); }); }; @@ -194,6 +194,7 @@ export function createConnectionBundle( inboundHandler.handleNotification(notification, { codexPath: sourceContext.codexPath, vaultPath: sourceContext.cwd, + generation: sourceContext.generation, }); host.deferLiveStateRefresh(); }, @@ -245,7 +246,7 @@ export function createConnectionBundle( host.refreshLiveState(); }, isStaleConnectionError: (error) => error instanceof StaleConnectionError, - isStaleSharedQueryError: isStaleAppServerSharedQueryContextError, + isStaleResourceContextError: isStaleAppServerResourceContextError, notifyConnectionFailed: () => { new Notice("Codex app-server connection failed."); }, diff --git a/src/features/chat/host/bundles/thread-bundle.ts b/src/features/chat/host/bundles/thread-bundle.ts index 8dd5fac3..027b2b47 100644 --- a/src/features/chat/host/bundles/thread-bundle.ts +++ b/src/features/chat/host/bundles/thread-bundle.ts @@ -114,8 +114,8 @@ export function createThreadFoundation(host: ChatPanelThreadHost, input: ChatPan const threadOperationsTransport = createThreadOperationsTransport(appServer.clientAccess); const threadTitleTransport = createThreadTitleTransport({ clientAccess: appServer.clientAccess, - codexPath: () => environment.plugin.settingsRef.settings.codexPath(), - vaultPath: environment.plugin.settingsRef.vaultPath, + codexPath: () => environment.plugin.appServerQueries.contextLease().context.codexPath, + vaultPath: environment.plugin.appServerQueries.contextLease().context.vaultPath, threadNamingModel: () => environment.plugin.settingsRef.settings.threadNamingModel(), threadNamingEffort: () => environment.plugin.settingsRef.settings.threadNamingEffort(), }); @@ -165,6 +165,8 @@ export function createThreadFoundation(host: ChatPanelThreadHost, input: ChatPan const invalidateThreadWork = () => { host.resumeWork.invalidate(); history.invalidate(); + titleService.invalidate(); + autoTitleCoordinator.invalidate(); }; const goalSync = createThreadGoalSyncActions({ stateStore, diff --git a/src/features/chat/host/contracts.ts b/src/features/chat/host/contracts.ts index 53b1a2f8..a8877475 100644 --- a/src/features/chat/host/contracts.ts +++ b/src/features/chat/host/contracts.ts @@ -1,7 +1,7 @@ import type { App, Component, EventRef } from "obsidian"; import type { AppServerClient } from "../../../app-server/connection/client"; -import type { AppServerQueryContext } from "../../../app-server/query/keys"; +import type { AppServerContextLease, AppServerQueryContextIdentity } from "../../../app-server/query/keys"; import type { ObservedResultListener } from "../../../app-server/query/observed-result"; import type { ModelMetadata, ReasoningEffort } from "../../../domain/catalog/metadata"; import type { SendShortcut } from "../../../domain/input/send-shortcut"; @@ -54,6 +54,7 @@ interface WorkspacePanels { type ChatThreadCatalog = ThreadCatalogActiveReader & ThreadCatalogEventSink & ThreadCatalogConnectionEventSink; interface ChatAppServerQueries { + contextLease(): AppServerContextLease; appServerMetadataSnapshot(): SharedServerMetadata | null; refreshAppServerMetadata(): Promise; refreshSkills(): Promise; @@ -122,7 +123,7 @@ export interface ChatSharedThreadSurface { } export interface ChatPanelClientSurface { - canServeAppServerContext(context: AppServerQueryContext): boolean; + canServeAppServerContext(context: AppServerQueryContextIdentity): boolean; runWithAppServerClient(operation: (client: AppServerClient) => Promise): Promise; } diff --git a/src/features/chat/host/session-runtime.ts b/src/features/chat/host/session-runtime.ts index aae9bc54..69d35469 100644 --- a/src/features/chat/host/session-runtime.ts +++ b/src/features/chat/host/session-runtime.ts @@ -1,6 +1,6 @@ import { codexPanelAppServerInitializeParams } from "../../../app-server/connection/client-profile"; import { ConnectionManager } from "../../../app-server/connection/connection-manager"; -import { isStaleAppServerSharedQueryContextError } from "../../../app-server/query/shared-queries"; +import { isStaleAppServerResourceContextError } from "../../../app-server/query/resource-store"; import { createChatAppServerGateway, createChatCurrentAppServerGateway } from "../app-server/session-gateway"; import { reconnectPanel } from "../application/connection/reconnect-actions"; import { createLocalIdSource, type LocalIdSource } from "../application/local-id-source"; @@ -114,9 +114,10 @@ function composeChatPanelSessionRuntime(host: ChatPanelSessionRuntimeHost): Chat const localItemIds = createLocalIdSource(); const connection = createConnectionManager(environment); const currentClient = () => connection.currentClient(); + const resourceContext = () => environment.plugin.appServerQueries.contextLease().context; const currentAppServer = createChatCurrentAppServerGateway({ - codexPath: () => environment.plugin.settingsRef.settings.codexPath(), - vaultPath: environment.plugin.settingsRef.vaultPath, + codexPath: () => resourceContext().codexPath, + vaultPath: resourceContext().vaultPath, currentClient, }); const status = createSessionStatus(stateStore, localItemIds); @@ -167,7 +168,7 @@ function composeChatPanelSessionRuntime(host: ChatPanelSessionRuntimeHost): Chat } = connectionBundle; const ensureConnected = () => connectionActions.ensureConnected(); const appServer = createChatAppServerGateway(currentAppServer, { - vaultPath: environment.plugin.settingsRef.vaultPath, + vaultPath: resourceContext().vaultPath, currentClient, connectedClient: async () => { await connectionActions.ensureConnected(); @@ -259,6 +260,7 @@ function composeChatPanelSessionRuntime(host: ChatPanelSessionRuntimeHost): Chat const prepareAppServerContextChange = (): void => { connectionActions.invalidate(); threadFoundation.invalidateThreadWork(); + threadLifecycle.rename.invalidate(); threadLifecycle.restoration.invalidate(); host.deferredTasks.clearDiagnostics(); connectionBundle.invalidateConnectionScope(); @@ -302,7 +304,7 @@ function composeChatPanelSessionRuntime(host: ChatPanelSessionRuntimeHost): Chat try { await connectionBundle.refreshSharedThreads(); } catch (error) { - if (isStaleAppServerSharedQueryContextError(error)) return; + if (isStaleAppServerResourceContextError(error)) return; throw error; } }; @@ -355,9 +357,11 @@ function composeChatPanelSessionRuntime(host: ChatPanelSessionRuntimeHost): Chat function createConnectionManager(environment: ChatPanelEnvironment): ConnectionManager { return new ConnectionManager( - () => environment.plugin.settingsRef.settings.codexPath(), - environment.plugin.settingsRef.vaultPath, + () => environment.plugin.appServerQueries.contextLease().context.codexPath, + environment.plugin.appServerQueries.contextLease().context.vaultPath, codexPanelAppServerInitializeParams(), + undefined, + () => environment.plugin.appServerQueries.contextLease().generation, ); } diff --git a/src/features/chat/host/session.ts b/src/features/chat/host/session.ts index a73093f4..c6d81d73 100644 --- a/src/features/chat/host/session.ts +++ b/src/features/chat/host/session.ts @@ -1,5 +1,9 @@ import type { AppServerClient } from "../../../app-server/connection/client"; -import { type AppServerQueryContext, appServerQueryContextMatches, appServerQueryContextRawEquals } from "../../../app-server/query/keys"; +import { + type AppServerQueryContextIdentity, + appServerQueryContextIdentity, + appServerQueryContextIdentityMatches, +} from "../../../app-server/query/keys"; import { pendingRequestCountsFromQueues } from "../../../domain/pending-requests/aggregate"; import { threadMeaningfulTitle, threadWindowTitle } from "../../../domain/threads/title"; import { @@ -25,7 +29,7 @@ export class ChatPanelSession implements ChatPanelHandle { private readonly deferredTasks: ChatViewDeferredTasks; private readonly resumeWork = new ChatResumeWorkTracker(); private readonly threadStreamScrollBinding: ChatThreadStreamScrollBinding = createChatThreadStreamScrollBinding(); - private observedAppServerContext: AppServerQueryContext; + private observedAppServerContext: AppServerQueryContextIdentity; private appServerContextReconnectAttemptGeneration = 0; private appServerContextReplacementGeneration = 0; private pendingAppServerContextReplacement: { activeThreadId: string | null; generation: number } | null = null; @@ -84,7 +88,7 @@ export class ChatPanelSession implements ChatPanelHandle { refreshSettings(): void { const nextContext = this.currentAppServerContext(); - if (!appServerQueryContextRawEquals(this.observedAppServerContext, nextContext)) { + if (!appServerQueryContextIdentityMatches(this.observedAppServerContext, nextContext)) { this.observedAppServerContext = nextContext; const replacement = this.pendingAppServerContextReplacement ?? this.captureAppServerContextReplacement(activeThreadId(this.state)); void this.reconnectAfterAppServerContextChange(replacement); @@ -142,14 +146,15 @@ export class ChatPanelSession implements ChatPanelHandle { return this.runtime.actions.refreshSharedThreads(); } - canServeAppServerContext(context: AppServerQueryContext): boolean { + canServeAppServerContext(context: AppServerQueryContextIdentity): boolean { const connectionContext = this.runtime.connection.manager.currentConnectionContext(); return Boolean( connectionContext && - appServerQueryContextMatches( + appServerQueryContextIdentityMatches( { codexPath: connectionContext.codexPath, vaultPath: connectionContext.cwd, + generation: connectionContext.generation, }, context, ), @@ -276,11 +281,8 @@ export class ChatPanelSession implements ChatPanelHandle { return this.environment.view.viewWindow() ?? window; } - private currentAppServerContext(): AppServerQueryContext { - return { - codexPath: this.environment.plugin.settingsRef.settings.codexPath(), - vaultPath: this.environment.plugin.settingsRef.vaultPath, - }; + private currentAppServerContext(): AppServerQueryContextIdentity { + return appServerQueryContextIdentity(this.environment.plugin.appServerQueries.contextLease()); } private closeToolbarPanelOnOutsidePointer(event: PointerEvent): void { diff --git a/src/features/threads-view/session.ts b/src/features/threads-view/session.ts index 43a9e87b..3f4e7aad 100644 --- a/src/features/threads-view/session.ts +++ b/src/features/threads-view/session.ts @@ -3,7 +3,7 @@ import { Notice } from "obsidian"; import { type AppServerQueryContext, appServerQueryContextRawEquals } from "../../app-server/query/keys"; import type { ObservedResult } from "../../app-server/query/observed-result"; import { observedInitialError, observedInitialLoading } from "../../app-server/query/observed-result"; -import { isStaleAppServerSharedQueryContextError } from "../../app-server/query/shared-queries"; +import { isStaleAppServerResourceContextError } from "../../app-server/query/resource-store"; import type { ReasoningEffort } from "../../domain/catalog/metadata"; import type { ArchiveExportSettings } from "../../domain/threads/archive-markdown"; import type { Thread } from "../../domain/threads/model"; @@ -115,6 +115,7 @@ export class ThreadsViewSession { close(): void { this.lifetime.dispose(); + this.titleService.invalidate(); this.activeRefresh = null; this.renderTask.clear(); this.unsubscribeThreads?.(); @@ -138,7 +139,7 @@ export class ThreadsViewSession { this.status = threads.length === 0 ? { kind: "empty", message: "No threads" } : { kind: "idle" }; } catch (error) { if (!this.lifetime.isCurrent(lifetime)) return; - if (isStaleAppServerSharedQueryContextError(error)) return; + if (isStaleAppServerResourceContextError(error)) return; if (!this.currentThreadsSnapshot()) { this.status = { kind: "error", message: error instanceof Error ? error.message : String(error) }; } @@ -159,7 +160,7 @@ export class ThreadsViewSession { this.threadsLoaded = true; this.status = threads.length === 0 ? { kind: "empty", message: "No threads" } : { kind: "idle" }; } catch (error) { - if (!this.lifetime.isCurrent(lifetime) || isStaleAppServerSharedQueryContextError(error)) return; + if (!this.lifetime.isCurrent(lifetime) || isStaleAppServerResourceContextError(error)) return; this.status = { kind: "error", message: error instanceof Error ? error.message : String(error) }; } finally { this.finishRefresh(refresh); @@ -172,6 +173,7 @@ export class ThreadsViewSession { prepareAppServerContextChange(): void { this.operationGeneration += 1; + this.titleService.invalidate(); this.activeRefresh = null; this.renderTask.clear(); this.threads = []; diff --git a/src/features/threads/app-server/workflow-transports.ts b/src/features/threads/app-server/workflow-transports.ts index c1ab152f..77ce9013 100644 --- a/src/features/threads/app-server/workflow-transports.ts +++ b/src/features/threads/app-server/workflow-transports.ts @@ -38,10 +38,16 @@ export function createThreadTitleTransport(options: { readCompletedTurnTranscriptSummariesPage(client, id, cursor, limit, sortDirection), }), ), - generateTitle: (context) => - generateThreadTitleWithCodex(options.codexPath(), options.vaultPath, context, { - threadNamingModel: options.threadNamingModel(), - threadNamingEffort: options.threadNamingEffort(), - }), + generateTitle: (context, signal) => + generateThreadTitleWithCodex( + options.codexPath(), + options.vaultPath, + context, + { + threadNamingModel: options.threadNamingModel(), + threadNamingEffort: options.threadNamingEffort(), + }, + { signal }, + ), }; } diff --git a/src/features/threads/catalog/thread-catalog.ts b/src/features/threads/catalog/thread-catalog.ts index 9cc5423f..45737614 100644 --- a/src/features/threads/catalog/thread-catalog.ts +++ b/src/features/threads/catalog/thread-catalog.ts @@ -1,4 +1,4 @@ -import type { AppServerQueryContext } from "../../../app-server/query/keys"; +import type { AppServerQueryContextIdentity } from "../../../app-server/query/keys"; import type { ObservedResult, ObservedResultListener } from "../../../app-server/query/observed-result"; import type { Thread } from "../../../domain/threads/model"; @@ -6,18 +6,14 @@ type ThreadListObserver = ObservedResultListener; interface ThreadCatalogStore { contextKey(): string; - contextKeyFor(context: AppServerQueryContext): string; + contextKeyFor(context: AppServerQueryContextIdentity): string; activeThreadsSnapshot(): readonly Thread[] | null; - activeThreadsSnapshotFor(context: AppServerQueryContext): readonly Thread[] | null; archivedThreadsSnapshot(): readonly Thread[] | null; - archivedThreadsSnapshotFor(context: AppServerQueryContext): readonly Thread[] | null; fetchAllActiveThreads(): Promise; hasMoreActiveThreads(): boolean; loadMoreActiveThreads(): Promise; refreshActiveThreads(): Promise; - refreshActiveThreadsFor(context: AppServerQueryContext): Promise; refreshArchivedThreads(): Promise; - refreshArchivedThreadsFor(context: AppServerQueryContext): Promise; observeActiveThreadsResult(observer: ThreadListObserver, options?: { emitCurrent?: boolean }): () => void; observeArchivedThreadsResult(observer: ThreadListObserver, options?: { emitCurrent?: boolean }): () => void; } @@ -87,7 +83,7 @@ export interface ThreadCatalogEventSink { } export interface ThreadCatalogConnectionEventSink { - applyConnectionEvent(context: AppServerQueryContext, event: ThreadCatalogEvent): void; + applyConnectionEvent(context: AppServerQueryContextIdentity, event: ThreadCatalogEvent): void; } export interface ThreadCatalog @@ -138,12 +134,8 @@ export function createThreadCatalog(options: ThreadCatalogOptions): ThreadCatalo apply, applyConnectionEvent: (context, event) => { const capturedContextKey = store.contextKeyFor(context); - applyToFacts( - threadCatalogEventStoreForContext(store, context), - threadCatalogFactsForContext(factsByContext, capturedContextKey), - event, - capturedContextKey === store.contextKey(), - ); + if (capturedContextKey !== store.contextKey()) return; + applyToFacts(store, currentFacts(), event, true); }, clear: () => { factsByContext.clear(); @@ -190,19 +182,10 @@ export function createThreadCatalog(options: ThreadCatalogOptions): ThreadCatalo }; } -function threadCatalogEventStoreForContext(store: ThreadCatalogStore, context: AppServerQueryContext): ThreadCatalogEventStore { - return { - activeThreadsSnapshot: () => store.activeThreadsSnapshotFor(context), - archivedThreadsSnapshot: () => store.archivedThreadsSnapshotFor(context), - refreshActiveThreads: () => store.refreshActiveThreadsFor(context), - refreshArchivedThreads: () => store.refreshArchivedThreadsFor(context), - }; -} - type ThreadListKind = "active" | "archived"; function threadCatalogFactsForContext(factsByContext: Map, contextKey: string): ThreadCatalogFacts { - pruneInactiveSettledContexts(factsByContext, contextKey); + pruneInactiveContexts(factsByContext, contextKey); const existing = factsByContext.get(contextKey); if (existing) { factsByContext.delete(contextKey); @@ -219,18 +202,13 @@ function threadCatalogFactsForContext(factsByContext: Map, activeContextKey: string): void { - for (const [contextKey, facts] of factsByContext) { +function pruneInactiveContexts(factsByContext: Map, activeContextKey: string): void { + for (const contextKey of factsByContext.keys()) { if (contextKey === activeContextKey) continue; - if (threadListFactsPending(facts.active) || threadListFactsPending(facts.archived)) continue; factsByContext.delete(contextKey); } } -function threadListFactsPending(facts: PendingThreadListFacts): boolean { - return facts.upserts.size > 0 || facts.removals.size > 0; -} - function applyThreadCatalogEvent( store: ThreadCatalogEventStore, facts: ThreadCatalogFacts, diff --git a/src/features/threads/workflows/ports.ts b/src/features/threads/workflows/ports.ts index 01dac8c3..cdfd0e9f 100644 --- a/src/features/threads/workflows/ports.ts +++ b/src/features/threads/workflows/ports.ts @@ -8,5 +8,5 @@ export interface ThreadOperationsTransport { export interface ThreadTitleTransport { persistedContext(threadId: string): Promise; - generateTitle(context: ThreadTitleContext): Promise; + generateTitle(context: ThreadTitleContext, signal: AbortSignal): Promise; } diff --git a/src/features/threads/workflows/thread-title-service.ts b/src/features/threads/workflows/thread-title-service.ts index 6f0fd962..7712292f 100644 --- a/src/features/threads/workflows/thread-title-service.ts +++ b/src/features/threads/workflows/thread-title-service.ts @@ -10,10 +10,11 @@ export interface ThreadTitleServiceHost { transport: ThreadTitleTransport; visibleContext?(threadId: string): ThreadTitleContext | null; visibleCompletedTurnContext?(turnId: string): ThreadTitleContext | null; - generateThreadTitle?(context: ThreadTitleContext): Promise; + generateThreadTitle?(context: ThreadTitleContext, signal: AbortSignal): Promise; } export interface ThreadTitleService { + invalidate(): void; generateTitle(threadId: string): Promise; resolveContext(threadId: string): Promise; completedTurnContext(turnId: string, completedTurnTranscriptSummary: TurnTranscriptSummary | null): ThreadTitleContext | null; @@ -21,19 +22,26 @@ export interface ThreadTitleService { } export function createThreadTitleService(host: ThreadTitleServiceHost): ThreadTitleService { + let controller = new AbortController(); + return { - generateTitle: (threadId) => generateTitle(host, threadId), + invalidate() { + controller.abort(); + controller = new AbortController(); + }, + generateTitle: (threadId) => generateTitle(host, threadId, controller.signal), resolveContext: (threadId) => resolveThreadTitleContext(host, threadId), completedTurnContext: (turnId, completedTurnTranscriptSummary) => completedTurnContext(host, turnId, completedTurnTranscriptSummary), - generate: (context) => generateTitleFromContext(host, context), + generate: (context) => generateTitleFromContext(host, context, controller.signal), }; } -async function generateTitle(host: ThreadTitleServiceHost, threadId: string): Promise { +async function generateTitle(host: ThreadTitleServiceHost, threadId: string, signal: AbortSignal): Promise { const context = await resolveThreadTitleContext(host, threadId); + throwIfTitleGenerationCancelled(signal); if (!context) throw new Error(THREAD_TITLE_CONTEXT_UNAVAILABLE_MESSAGE); - const title = await generateTitleFromContext(host, context); + const title = await generateTitleFromContext(host, context, signal); if (!title) throw new Error("Codex did not return a usable thread title."); return title; } @@ -59,6 +67,19 @@ function completedTurnContext( ); } -async function generateTitleFromContext(host: ThreadTitleServiceHost, context: ThreadTitleContext): Promise { - return host.generateThreadTitle ? host.generateThreadTitle(context) : host.transport.generateTitle(context); +async function generateTitleFromContext( + host: ThreadTitleServiceHost, + context: ThreadTitleContext, + signal: AbortSignal, +): Promise { + throwIfTitleGenerationCancelled(signal); + const title = await (host.generateThreadTitle + ? host.generateThreadTitle(context, signal) + : host.transport.generateTitle(context, signal)); + throwIfTitleGenerationCancelled(signal); + return title; +} + +function throwIfTitleGenerationCancelled(signal: AbortSignal): void { + if (signal.aborted) throw new Error("Thread title generation cancelled."); } diff --git a/src/main.ts b/src/main.ts index 5c949385..730fa9a3 100644 --- a/src/main.ts +++ b/src/main.ts @@ -25,6 +25,7 @@ export default class CodexPanelPlugin extends Plugin { this.runtime.reset(); this.vaultPath = getVaultPath(this.app); await this.loadSettings(); + this.runtime.initialize(); this.registerView(VIEW_TYPE_CODEX_PANEL, (leaf) => new CodexChatView(leaf, this.runtime.chatHost())); this.registerView(VIEW_TYPE_CODEX_TURN_DIFF, (leaf) => new CodexTurnDiffView(leaf)); @@ -75,8 +76,8 @@ export default class CodexPanelPlugin extends Plugin { registerSelectionRewriteCommand( this, createAppServerSelectionRewriteTransport({ - codexPath: () => this.settings.codexPath, - cwd: this.vaultPath, + codexPath: () => this.runtime.appServerContextLease().context.codexPath, + cwd: this.runtime.appServerContextLease().context.vaultPath, }), ), ); diff --git a/src/plugin-runtime.ts b/src/plugin-runtime.ts index ab5587da..9cdef7c3 100644 --- a/src/plugin-runtime.ts +++ b/src/plugin-runtime.ts @@ -2,9 +2,14 @@ import type { App } from "obsidian"; import type { AppServerClient } from "./app-server/connection/client"; import type { AppServerClientAccess, AppServerClientAccessOptions } from "./app-server/connection/client-access"; import { withShortLivedAppServerClient } from "./app-server/connection/short-lived-client"; -import { AppServerQueryCache } from "./app-server/query/cache"; -import { type AppServerQueryContext, appServerQueryContextIsComplete, appServerQueryContextRawEquals } from "./app-server/query/keys"; -import { AppServerSharedQueries, StaleAppServerSharedQueryContextError } from "./app-server/query/shared-queries"; +import { + type AppServerContextLease, + type AppServerQueryContext, + type AppServerQueryContextIdentity, + appServerQueryContextIdentityMatches, + appServerQueryContextIsComplete, +} from "./app-server/query/keys"; +import { AppServerResourceStore, StaleAppServerResourceContextError } from "./app-server/query/resource-store"; import { VIEW_TYPE_CODEX_THREADS, VIEW_TYPE_CODEX_TURN_DIFF } from "./constants"; import { hasPendingRequests } from "./domain/pending-requests/aggregate"; import type { @@ -42,15 +47,11 @@ export interface CodexPanelRuntimeOptions { } export class CodexPanelRuntime implements AppServerClientAccess { - private readonly appServerQueries = new AppServerQueryCache({ + private readonly appServerResourceStore = new AppServerResourceStore({ clientRunner: { runWithClient: (context, operation, options) => this.runWithAppServerClient(context, operation, options), }, }); - private readonly appServerSharedQueries = new AppServerSharedQueries({ - cache: this.appServerQueries, - context: () => this.appServerQueryContext(), - }); private readonly panels: WorkspacePanelCoordinator; private readonly threadCatalog: ThreadCatalog; private readonly threadNameMutations = createThreadNameMutationCoordinator(); @@ -64,18 +65,26 @@ export class CodexPanelRuntime implements AppServerClientAccess { }, }); this.threadCatalog = createThreadCatalog({ - store: this.appServerSharedQueries, + store: this.appServerResourceStore, onEventApplied: (event) => { this.applyThreadCatalogSurfaceEvent(event); }, }); } + initialize(): void { + this.appServerResourceStore.initialize(this.configuredAppServerContext()); + } + + appServerContextLease(): AppServerContextLease { + return this.appServerResourceStore.contextLease(); + } + reset(): void { this.selectionRewriteController?.closeAll(); this.selectionRewriteController = null; this.panels.reset(); - this.appServerQueries.clear(); + this.appServerResourceStore.reset(); this.threadCatalog.clear(); } @@ -138,7 +147,7 @@ export class CodexPanelRuntime implements AppServerClientAccess { this.refreshThreadsViewLiveState(); }, }, - appServerQueries: this.appServerSharedQueries, + appServerQueries: this.appServerResourceStore, threadCatalog: this.threadCatalog, threadNameMutations: this.threadNameMutations, }; @@ -172,8 +181,8 @@ export class CodexPanelRuntime implements AppServerClientAccess { threadOperationsTransport: createThreadOperationsTransport(this), threadTitleTransport: createThreadTitleTransport({ clientAccess: this, - codexPath: () => this.options.settingsRef.settings.codexPath, - vaultPath: this.options.settingsRef.vaultPath, + codexPath: () => this.appServerResourceStore.contextLease().context.codexPath, + vaultPath: this.appServerResourceStore.contextLease().context.vaultPath, threadNamingModel: () => this.options.settingsRef.settings.threadNamingModel, threadNamingEffort: () => this.options.settingsRef.settings.threadNamingEffort, }), @@ -206,18 +215,10 @@ export class CodexPanelRuntime implements AppServerClientAccess { dynamicData: createSettingsAppServerDynamicData({ vaultPath: this.options.settingsRef.vaultPath, clientAccess: this, - appServerQueries: this.appServerSharedQueries, + appServerQueries: this.appServerResourceStore, threadCatalog: this.threadCatalog, }), - saveSettings: (settings) => this.options.saveSettings(settings), - prepareAppServerContextChange: () => { - this.selectionRewriteController?.closeAll(); - for (const view of this.panels.panelViews()) view.surface.prepareAppServerContextChange(); - for (const view of this.threadsViews()) view.prepareAppServerContextChange(); - }, - refreshOpenViews: () => { - this.refreshOpenViews(); - }, + publishSettings: (settings) => this.publishSettings(settings), }; } @@ -231,7 +232,24 @@ export class CodexPanelRuntime implements AppServerClientAccess { } withClient(operation: (client: AppServerClient) => Promise, options: AppServerClientAccessOptions = {}): Promise { - return this.runWithAppServerClient(this.appServerQueryContext(), operation, options); + return this.runWithAppServerClient(this.appServerResourceStore.contextIdentity(), operation, options); + } + + private async publishSettings(settings: CodexPanelSettings): Promise<{ appServerContextReplaced: boolean }> { + const previousSettings = { ...this.options.settingsRef.settings }; + await this.options.saveSettings(settings); + const appServerContextReplaced = previousSettings.codexPath !== settings.codexPath; + if (appServerContextReplaced) this.prepareAppServerContextChange(); + Object.assign(this.options.settingsRef.settings, settings); + if (appServerContextReplaced) this.appServerResourceStore.replaceContext(this.configuredAppServerContext()); + if (appServerContextReplaced || previousSettings.showToolbar !== settings.showToolbar) this.refreshOpenViews(); + return { appServerContextReplaced }; + } + + private prepareAppServerContextChange(): void { + this.selectionRewriteController?.closeAll(); + for (const view of this.panels.panelViews()) view.surface.prepareAppServerContextChange(); + for (const view of this.threadsViews()) view.prepareAppServerContextChange(); } private async openTurnDiff(state: TurnDiffViewState): Promise { @@ -309,7 +327,7 @@ export class CodexPanelRuntime implements AppServerClientAccess { } private async runWithAppServerClient( - context: AppServerQueryContext, + context: AppServerQueryContextIdentity, operation: (client: AppServerClient) => Promise, options: AppServerClientAccessOptions = {}, ): Promise { @@ -317,14 +335,14 @@ export class CodexPanelRuntime implements AppServerClientAccess { throw new Error("Codex app-server query context is incomplete."); } const result = await this.runWithContextClient(context, operation, options); - if (!appServerQueryContextRawEquals(this.appServerQueryContext(), context)) { - throw new StaleAppServerSharedQueryContextError(); + if (!appServerQueryContextIdentityMatches(this.appServerResourceStore.contextIdentity(), context)) { + throw new StaleAppServerResourceContextError(); } return result; } private runWithContextClient( - context: AppServerQueryContext, + context: AppServerQueryContextIdentity, operation: (client: AppServerClient) => Promise, options: AppServerClientAccessOptions, ): Promise { @@ -337,7 +355,7 @@ export class CodexPanelRuntime implements AppServerClientAccess { : withShortLivedAppServerClient(context.codexPath, context.vaultPath, operation, options); } - private connectedClientSurface(context: AppServerQueryContext): ChatPanelClientSurface | null { + private connectedClientSurface(context: AppServerQueryContextIdentity): ChatPanelClientSurface | null { for (const view of this.panels.panelViews()) { const workspaceSurface: ChatWorkspacePanelSurface = view.surface; if (!workspaceSurface.openPanelSnapshot().connected) continue; @@ -348,7 +366,7 @@ export class CodexPanelRuntime implements AppServerClientAccess { return null; } - private appServerQueryContext(): AppServerQueryContext { + private configuredAppServerContext(): AppServerQueryContext { return { codexPath: this.options.settingsRef.settings.codexPath, vaultPath: this.options.settingsRef.vaultPath, diff --git a/src/settings/app-server-dynamic-data.ts b/src/settings/app-server-dynamic-data.ts index 1e32a8a4..0bceff72 100644 --- a/src/settings/app-server-dynamic-data.ts +++ b/src/settings/app-server-dynamic-data.ts @@ -1,7 +1,7 @@ import type { AppServerClient } from "../app-server/connection/client"; import type { AppServerClientAccess } from "../app-server/connection/client-access"; import type { ObservedResultListener } from "../app-server/query/observed-result"; -import { isStaleAppServerSharedQueryContextError } from "../app-server/query/shared-queries"; +import { isStaleAppServerResourceContextError } from "../app-server/query/resource-store"; import { listHookCatalog, setHookItemEnabled, trustHookItem } from "../app-server/services/catalog"; import { deleteThread, restoreArchivedThread as restoreArchivedThreadOnAppServer } from "../app-server/services/threads"; import type { ModelMetadata } from "../domain/catalog/metadata"; @@ -13,7 +13,6 @@ interface SettingsAppServerQueries { observeModelsResult(listener: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void; fetchModels(): Promise; refreshModels(): Promise; - notifyContextChanged(): void; } type SettingsArchivedThreadCatalog = ThreadCatalogArchivedReader & ThreadCatalogEventSink; @@ -55,9 +54,6 @@ export function createSettingsAppServerDynamicData(options: SettingsAppServerDyn options.threadCatalog.apply({ type: "thread-deleted", threadId }); } }, - notifyContextChanged: () => { - options.appServerQueries.notifyContextChanged(); - }, }; } @@ -74,7 +70,7 @@ async function mapStaleContextError(operation: () => Promise): Promise try { return await operation(); } catch (error) { - if (isStaleAppServerSharedQueryContextError(error)) throw new StaleSettingsDynamicDataContextError(); + if (isStaleAppServerResourceContextError(error)) throw new StaleSettingsDynamicDataContextError(); throw error; } } diff --git a/src/settings/dynamic-data.ts b/src/settings/dynamic-data.ts index aae0a4dc..c06c9eed 100644 --- a/src/settings/dynamic-data.ts +++ b/src/settings/dynamic-data.ts @@ -26,7 +26,6 @@ export interface SettingsDynamicDataAccess { setHookEnabled(hook: HookItem, enabled: boolean): Promise; restoreArchivedThread(threadId: string, options?: SettingsDynamicDataMutationOptions): Promise; deleteArchivedThread(threadId: string, options?: SettingsDynamicDataMutationOptions): Promise; - notifyContextChanged(): void; } export class StaleSettingsDynamicDataContextError extends Error { diff --git a/src/settings/host.ts b/src/settings/host.ts index 62ada3dd..ea20a5c6 100644 --- a/src/settings/host.ts +++ b/src/settings/host.ts @@ -7,7 +7,5 @@ export interface SettingsDynamicSectionsHost { } export interface CodexPanelSettingTabHost extends SettingsDynamicSectionsHost { - saveSettings(settings: CodexPanelSettings): Promise; - prepareAppServerContextChange(): void; - refreshOpenViews(): void; + publishSettings(settings: CodexPanelSettings): Promise<{ appServerContextReplaced: boolean }>; } diff --git a/src/settings/tab.obsidian.tsx b/src/settings/tab.obsidian.tsx index 5e795ee3..5ae3f0ee 100644 --- a/src/settings/tab.obsidian.tsx +++ b/src/settings/tab.obsidian.tsx @@ -195,29 +195,17 @@ export class CodexPanelSettingTab extends PluginSettingTab { return true; }, { - beforePublish: () => { - this.plugin.prepareAppServerContextChange(); - }, - onPublished: () => { - this.dynamicSections.resetDynamicSectionContext(); - this.plugin.dynamicData.notifyContextChanged(); - this.plugin.refreshOpenViews(); + onPublished: ({ appServerContextReplaced }) => { + if (appServerContextReplaced) this.dynamicSections.resetDynamicSectionContext(); }, }, ); } private setShowToolbar(value: boolean): Promise { - return this.queueSettingsMutation( - (settings) => { - settings.showToolbar = value; - }, - { - onPublished: () => { - this.plugin.refreshOpenViews(); - }, - }, - ); + return this.queueSettingsMutation((settings) => { + settings.showToolbar = value; + }); } private setSendShortcut(value: "enter" | "mod-enter"): Promise { @@ -300,21 +288,19 @@ export class CodexPanelSettingTab extends PluginSettingTab { private queueSettingsMutation( mutate: (settings: CodexPanelSettings) => boolean | undefined, - publication: { beforePublish?: () => void; onPublished?: () => void } = {}, + publication: { onPublished?: (result: { appServerContextReplaced: boolean }) => void } = {}, ): Promise { const operation = this.settingsMutationQueue.then(async () => { const candidateSettings: CodexPanelSettings = { ...this.plugin.settings }; if (mutate(candidateSettings) === false) return; try { - await this.plugin.saveSettings(candidateSettings); + const result = await this.plugin.publishSettings(candidateSettings); + publication.onPublished?.(result); } catch (error) { this.settingsShellRevision += 1; new Notice(`Failed to save Codex Panel settings: ${error instanceof Error ? error.message : String(error)}`); return; } - publication.beforePublish?.(); - Object.assign(this.plugin.settings, candidateSettings); - publication.onPublished?.(); }); const settledOperation = operation.catch(() => undefined); this.settingsMutationQueue = settledOperation; diff --git a/tests/app-server/connection-manager.test.ts b/tests/app-server/connection-manager.test.ts index 77fea8e8..bb27925e 100644 --- a/tests/app-server/connection-manager.test.ts +++ b/tests/app-server/connection-manager.test.ts @@ -111,7 +111,7 @@ describe("ConnectionManager", () => { transports[0]?.emitLine({ id: 1, result: { codexHome: "/tmp/codex-a" } }); await expect(first).resolves.toMatchObject({ codexHome: "/tmp/codex-a" }); expect(manager.currentClient()).toBeInstanceOf(AppServerClient); - expect(manager.currentConnectionContext()).toEqual({ codexPath: "/bin/codex-a", cwd: "/vault" }); + expect(manager.currentConnectionContext()).toEqual({ codexPath: "/bin/codex-a", cwd: "/vault", generation: 0 }); codexPath = "/bin/codex-b"; @@ -124,7 +124,7 @@ describe("ConnectionManager", () => { expect(transports).toHaveLength(2); expect(transports[0]?.running).toBe(false); expect(manager.currentClient()).toBeInstanceOf(AppServerClient); - expect(manager.currentConnectionContext()).toEqual({ codexPath: "/bin/codex-b", cwd: "/vault" }); + expect(manager.currentConnectionContext()).toEqual({ codexPath: "/bin/codex-b", cwd: "/vault", generation: 0 }); }); it("labels inbound events with the context captured when the connection was created", async () => { @@ -149,7 +149,31 @@ describe("ConnectionManager", () => { transport.emitLine(notification); - expect(onNotification).toHaveBeenCalledWith(notification, { codexPath: "/bin/codex-a", cwd: "/vault" }); + expect(onNotification).toHaveBeenCalledWith(notification, { codexPath: "/bin/codex-a", cwd: "/vault", generation: 0 }); + }); + + it("does not reuse a connected client across generations of the same raw context", async () => { + let contextGeneration = 1; + const transports: SilentTransport[] = []; + const manager = new ConnectionManager( + () => "/bin/codex", + "/vault", + TEST_INITIALIZE_PARAMS, + testClientFactory({ onTransport: (transport) => transports.push(transport) }), + () => contextGeneration, + ); + const first = manager.connect(silentConnectionHandlers()); + transports[0]?.emitLine({ id: 1, result: { codexHome: "/tmp/codex" } }); + await first; + + contextGeneration = 3; + + expect(manager.currentClient()).toBeNull(); + const second = manager.connect(silentConnectionHandlers()); + transports[1]?.emitLine({ id: 1, result: { codexHome: "/tmp/codex" } }); + await second; + expect(manager.currentConnectionContext()).toEqual({ codexPath: "/bin/codex", cwd: "/vault", generation: 3 }); + expect(transports[0]?.running).toBe(false); }); it("rejects app-server exit during initialization without reporting a connected-client exit", async () => { diff --git a/tests/app-server/query-cache.test.ts b/tests/app-server/query-cache.test.ts index dfd380f4..ad51abf6 100644 --- a/tests/app-server/query-cache.test.ts +++ b/tests/app-server/query-cache.test.ts @@ -1,7 +1,7 @@ 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"; -import type { AppServerQueryContext } from "../../src/app-server/query/keys"; +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"; @@ -144,6 +144,26 @@ describe("AppServerQueryCache", () => { expect(cache.activeThreadsSnapshot(oldContext)?.map((item) => item.id)).toEqual(["old-thread"]); }); + it("does not join or publish thread refreshes across generations of the same raw context", async () => { + const firstRefresh = deferred[]>(); + const fetchThreads = vi.fn((context: AppServerQueryContextIdentity) => + context.generation === 1 ? firstRefresh.promise : Promise.resolve([thread("current-a")]), + ); + const cache = cacheWithThreads(fetchThreads); + const firstA = cacheContext({ generation: 1 }); + const secondA = cacheContext({ generation: 3 }); + + const stale = cache.refreshActiveThreads(firstA); + await flushMicrotasks(); + await expect(cache.refreshActiveThreads(secondA)).resolves.toEqual([thread("current-a")]); + firstRefresh.resolve([thread("stale-a")]); + await stale; + + expect(fetchThreads).toHaveBeenCalledTimes(2); + expect(cache.activeThreadsSnapshot(secondA)).toEqual([thread("current-a")]); + expect(cache.activeThreadsSnapshot(firstA)).toEqual([thread("stale-a")]); + }); + it("stores in-flight thread list refreshes under the captured app-server cache context", async () => { const context = cacheContext({ codexPath: "codex-captured" }); const capturedContext = { ...context }; @@ -537,16 +557,17 @@ describe("AppServerQueryCache", () => { }); }); -function cacheContext(overrides: Partial = {}): AppServerQueryContext { +function cacheContext(overrides: Partial = {}): AppServerQueryContextIdentity { return { codexPath: "codex", vaultPath: "/vault", + generation: 1, ...overrides, }; } function cacheWithThreads( - fetchThreads: (context: AppServerQueryContext, archived: boolean) => Promise[]>, + fetchThreads: (context: AppServerQueryContextIdentity, archived: boolean) => Promise[]>, ): AppServerQueryCache { return new AppServerQueryCache({ clientRunner: { diff --git a/tests/app-server/resource-store.test.ts b/tests/app-server/resource-store.test.ts new file mode 100644 index 00000000..aabbc816 --- /dev/null +++ b/tests/app-server/resource-store.test.ts @@ -0,0 +1,139 @@ +import { describe, expect, it, vi } from "vitest"; + +import type { AppServerQueryCache } from "../../src/app-server/query/cache"; +import type { AppServerQueryContextIdentity } from "../../src/app-server/query/keys"; +import type { ObservedResult } from "../../src/app-server/query/observed-result"; +import { AppServerResourceStore, StaleAppServerResourceContextError } from "../../src/app-server/query/resource-store"; +import type { Thread } from "../../src/domain/threads/model"; +import { deferred } from "../support/async"; + +describe("AppServerResourceStore", () => { + it("requires explicit initialization and freezes the active context lease", () => { + const store = new AppServerResourceStore({ cache: cacheWith() }); + + expect(() => store.contextLease()).toThrow("not initialized"); + const lease = store.initialize({ codexPath: "codex-a", vaultPath: "/vault" }); + + expect(lease).toEqual({ context: { codexPath: "codex-a", vaultPath: "/vault" }, generation: 1 }); + expect(Object.isFrozen(lease)).toBe(true); + expect(Object.isFrozen(lease.context)).toBe(true); + expect(() => store.initialize({ codexPath: "codex-b", vaultPath: "/vault" })).toThrow("already initialized"); + }); + + it("keeps the same lease and cache coordination when replacement receives the same raw context", async () => { + const pending = deferred(); + const refreshActiveThreads = vi.fn(() => pending.promise); + const cache = cacheWith({ refreshActiveThreads }); + const store = new AppServerResourceStore({ cache }); + const firstLease = store.initialize({ codexPath: "codex-a", vaultPath: "/vault" }); + + const refresh = store.refreshActiveThreads(); + const secondLease = store.replaceContext({ codexPath: "codex-a", vaultPath: "/vault" }); + pending.resolve([thread("same-lease")]); + + expect(secondLease).toBe(firstLease); + await expect(refresh).resolves.toEqual([thread("same-lease")]); + expect(cache.release).not.toHaveBeenCalled(); + }); + + it("rejects an A1 result after A to B to A replacement and never reuses a generation", async () => { + const pending = deferred(); + const identities: AppServerQueryContextIdentity[] = []; + const cache = cacheWith({ + refreshActiveThreads: vi.fn((identity: AppServerQueryContextIdentity) => { + identities.push(identity); + return pending.promise; + }), + }); + const store = new AppServerResourceStore({ cache }); + const firstA = store.initialize({ codexPath: "codex-a", vaultPath: "/vault" }); + + const refresh = store.refreshActiveThreads(); + const b = store.replaceContext({ codexPath: "codex-b", vaultPath: "/vault" }); + const secondA = store.replaceContext({ codexPath: "codex-a", vaultPath: "/vault" }); + pending.resolve([thread("stale-a")]); + + await expect(refresh).rejects.toBeInstanceOf(StaleAppServerResourceContextError); + expect([firstA.generation, b.generation, secondA.generation]).toEqual([1, 2, 3]); + expect(identities).toEqual([{ codexPath: "codex-a", vaultPath: "/vault", generation: 1 }]); + }); + + it("rebinds observers by generation and ignores events from an earlier identical raw context", () => { + const listeners = new Map) => void>(); + const unsubscribers = new Map>(); + const store = new AppServerResourceStore({ + cache: cacheWith({ + observeActiveThreadsResult: (identity, listener) => { + listeners.set(identity.generation, listener); + const unsubscribe = vi.fn(); + unsubscribers.set(identity.generation, unsubscribe); + return unsubscribe; + }, + }), + }); + store.initialize({ codexPath: "codex-a", vaultPath: "/vault" }); + const listener = vi.fn(); + store.observeActiveThreadsResult(listener); + + store.replaceContext({ codexPath: "codex-b", vaultPath: "/vault" }); + store.replaceContext({ codexPath: "codex-a", vaultPath: "/vault" }); + listeners.get(1)?.(observedResult([thread("stale-a")])); + listeners.get(3)?.(observedResult([thread("current-a")])); + + expect(unsubscribers.get(1)).toHaveBeenCalledOnce(); + expect(unsubscribers.get(2)).toHaveBeenCalledOnce(); + expect(listener).toHaveBeenCalledOnce(); + expect(listener).toHaveBeenCalledWith(expect.objectContaining({ value: [thread("current-a")] })); + }); + + it("does not reuse a lease generation after reset", () => { + const store = new AppServerResourceStore({ cache: cacheWith() }); + expect(store.initialize({ codexPath: "codex", vaultPath: "/vault" }).generation).toBe(1); + + store.reset(); + + expect(store.initialize({ codexPath: "codex", vaultPath: "/vault" }).generation).toBe(2); + }); +}); + +function cacheWith(overrides: Partial = {}): AppServerQueryCache { + return { + clear: vi.fn(), + release: vi.fn(), + activeThreadsSnapshot: vi.fn(() => null), + archivedThreadsSnapshot: vi.fn(() => null), + fetchAllActiveThreads: vi.fn(() => Promise.resolve([])), + hasMoreActiveThreads: vi.fn(() => false), + loadMoreActiveThreads: vi.fn(() => Promise.resolve([])), + refreshActiveThreads: vi.fn(() => Promise.resolve([])), + refreshArchivedThreads: vi.fn(() => Promise.resolve([])), + observeActiveThreadsResult: vi.fn(() => () => undefined), + observeArchivedThreadsResult: vi.fn(() => () => undefined), + appServerMetadataSnapshot: vi.fn(() => null), + refreshAppServerMetadata: vi.fn(() => Promise.resolve(null)), + refreshSkills: vi.fn(() => Promise.resolve(null)), + refreshRateLimits: vi.fn(() => Promise.resolve(null)), + observeAppServerMetadataResult: vi.fn(() => () => undefined), + modelsSnapshot: vi.fn(() => null), + fetchModels: vi.fn(() => Promise.resolve([])), + refreshModels: vi.fn(() => Promise.resolve([])), + observeModelsResult: vi.fn(() => () => undefined), + ...overrides, + } as unknown as AppServerQueryCache; +} + +function observedResult(value: T): ObservedResult { + return { value, error: null, isFetching: false }; +} + +function thread(id: string): Thread { + return { + id, + preview: id, + createdAt: 1, + updatedAt: 1, + name: null, + archived: false, + provenance: { kind: "interactive" }, + }; +} diff --git a/tests/app-server/shared-queries.test.ts b/tests/app-server/shared-queries.test.ts deleted file mode 100644 index 76812544..00000000 --- a/tests/app-server/shared-queries.test.ts +++ /dev/null @@ -1,204 +0,0 @@ -import { describe, expect, it, vi } from "vitest"; - -import type { AppServerQueryCache } from "../../src/app-server/query/cache"; -import type { ObservedResult } from "../../src/app-server/query/observed-result"; -import { AppServerSharedQueries, StaleAppServerSharedQueryContextError } from "../../src/app-server/query/shared-queries"; -import type { ModelMetadata } from "../../src/domain/catalog/metadata"; -import { createServerDiagnostics, diagnosticProbeOk, diagnosticsWithProbe } from "../../src/domain/server/diagnostics"; -import type { SharedServerMetadata } from "../../src/domain/server/metadata"; -import type { Thread } from "../../src/domain/threads/model"; -import { deferred } from "../support/async"; - -describe("AppServerSharedQueries", () => { - it("rejects active thread refreshes when the app-server query context changes while loading", async () => { - const context = { codexPath: "codex-a", vaultPath: "/vault" }; - const pending = deferred(); - const queries = new AppServerSharedQueries({ - cache: cacheWith({ - refreshActiveThreads: vi.fn(() => pending.promise), - }), - context: () => context, - }); - - const refresh = queries.refreshActiveThreads(); - context.codexPath = "codex-b"; - pending.resolve([thread("stale")]); - - await expect(refresh).rejects.toBeInstanceOf(StaleAppServerSharedQueryContextError); - }); - - it("rejects metadata refreshes when the app-server query context changes while loading", async () => { - const context = { codexPath: "codex-a", vaultPath: "/vault" }; - const pending = deferred(); - const queries = new AppServerSharedQueries({ - cache: cacheWith({ - refreshAppServerMetadata: vi.fn(() => pending.promise), - }), - context: () => context, - }); - - const refresh = queries.refreshAppServerMetadata(); - context.vaultPath = "/other-vault"; - pending.resolve(serverMetadata()); - - await expect(refresh).rejects.toBeInstanceOf(StaleAppServerSharedQueryContextError); - }); - - it("rejects model fetches when the app-server query context changes while loading", async () => { - const context = { codexPath: "codex-a", vaultPath: "/vault" }; - const pending = deferred(); - const queries = new AppServerSharedQueries({ - cache: cacheWith({ - fetchModels: vi.fn(() => pending.promise), - }), - context: () => context, - }); - - const fetch = queries.fetchModels(); - context.codexPath = "codex-b"; - pending.resolve([model("stale-model")]); - - await expect(fetch).rejects.toBeInstanceOf(StaleAppServerSharedQueryContextError); - }); - - it("does not notify stale active thread observers after the app-server query context changes", async () => { - const context = { codexPath: "codex-a", vaultPath: "/vault" }; - const listener = vi.fn(); - const queries = new AppServerSharedQueries({ - cache: cacheWith({ - observeActiveThreadsResult: (_queryContext, queryListener) => { - observedThreadListener = queryListener; - return () => undefined; - }, - }), - context: () => context, - }); - let observedThreadListener!: (result: ObservedResult) => void; - - queries.observeActiveThreadsResult(listener); - context.codexPath = "codex-b"; - observedThreadListener(observedResult([thread("stale")])); - - expect(listener).not.toHaveBeenCalled(); - }); - - it("resubscribes observers when the app-server query context changes", () => { - const context = { codexPath: "codex-a", vaultPath: "/vault" }; - const listeners = new Map) => void>(); - const queries = new AppServerSharedQueries({ - cache: cacheWith({ - observeActiveThreadsResult: (queryContext, listener) => { - listeners.set(queryContext.codexPath, listener); - return () => undefined; - }, - }), - context: () => context, - }); - const listener = vi.fn(); - queries.observeActiveThreadsResult(listener); - - listeners.get("codex-a")?.(observedResult([thread("a")])); - context.codexPath = "codex-b"; - queries.notifyContextChanged(); - listeners.get("codex-b")?.(observedResult([thread("b")])); - - expect(listener).toHaveBeenLastCalledWith(expect.objectContaining({ value: [thread("b")] })); - }); - - it("publishes metadata and model snapshots to shared query observers", () => { - const metadata = serverMetadata(); - const models = [model("gpt-test")]; - const metadataListener = vi.fn(); - const modelListener = vi.fn(); - const queries = new AppServerSharedQueries({ - cache: cacheWith({ - appServerMetadataSnapshot: () => metadata, - modelsSnapshot: () => models, - observeAppServerMetadataResult: (_context, listener) => { - metadataObserver = listener; - return () => undefined; - }, - observeModelsResult: (_context, listener) => { - modelObserver = listener; - return () => undefined; - }, - }), - context: () => ({ codexPath: "codex", vaultPath: "/vault" }), - }); - let metadataObserver!: (result: ObservedResult) => void; - let modelObserver!: (result: ObservedResult) => void; - - queries.observeAppServerMetadataResult(metadataListener); - queries.observeModelsResult(modelListener); - metadataObserver(observedResult(metadata)); - modelObserver(observedResult(models)); - - expect(queries.appServerMetadataSnapshot()).toEqual(metadata); - expect(queries.modelsSnapshot()).toEqual(models); - expect(metadataListener).toHaveBeenLastCalledWith(expect.objectContaining({ value: metadata })); - expect(modelListener).toHaveBeenCalledWith(expect.objectContaining({ value: models })); - }); -}); - -function cacheWith(overrides: Partial): AppServerQueryCache { - return { - activeThreadsSnapshot: vi.fn(() => null), - fetchActiveThreads: vi.fn(() => Promise.resolve([])), - refreshActiveThreads: vi.fn(() => Promise.resolve([])), - observeActiveThreadsResult: vi.fn(() => () => undefined), - appServerMetadataSnapshot: vi.fn(() => null), - refreshAppServerMetadata: vi.fn(() => Promise.resolve(null)), - refreshSkills: vi.fn(() => Promise.resolve(null)), - refreshRateLimits: vi.fn(() => Promise.resolve(null)), - observeAppServerMetadataResult: vi.fn(() => () => undefined), - modelsSnapshot: vi.fn(() => null), - fetchModels: vi.fn(() => Promise.resolve([])), - refreshModels: vi.fn(() => Promise.resolve([])), - observeModelsResult: vi.fn(() => () => undefined), - ...overrides, - } as unknown as AppServerQueryCache; -} - -function observedResult(value: T): ObservedResult { - return { value, error: null, isFetching: false }; -} - -function thread(id: string): Thread { - return { - id, - preview: id, - createdAt: 1, - updatedAt: 1, - name: null, - archived: false, - provenance: { kind: "interactive" }, - }; -} - -function model(modelId: string): ModelMetadata { - return { - id: modelId, - model: modelId, - displayName: modelId, - description: "", - hidden: false, - supportedReasoningEfforts: [], - defaultReasoningEffort: null, - inputModalities: [], - serviceTiers: [], - defaultServiceTier: null, - isDefault: false, - }; -} - -function serverMetadata(overrides: Partial = {}): SharedServerMetadata { - const diagnostics = diagnosticsWithProbe(createServerDiagnostics(), diagnosticProbeOk("models", "0 models", 1)); - return { - runtimeConfig: null, - availableSkills: [], - availablePermissionProfiles: [], - rateLimit: null, - serverDiagnostics: diagnostics, - ...overrides, - }; -} diff --git a/tests/features/chat/app-server/inbound/handler.test.ts b/tests/features/chat/app-server/inbound/handler.test.ts index 797afa7b..f275c150 100644 --- a/tests/features/chat/app-server/inbound/handler.test.ts +++ b/tests/features/chat/app-server/inbound/handler.test.ts @@ -23,7 +23,7 @@ import { chatStateFixture, chatStateWith } from "../../support/state"; import { chatStateThreadStreamItems, withChatStateThreadStreamItems } from "../../support/thread-stream"; type ThreadStartedNotification = Extract; -const TEST_APP_SERVER_CONTEXT = { codexPath: "codex", vaultPath: "/vault" } as const; +const TEST_APP_SERVER_CONTEXT = { codexPath: "codex", vaultPath: "/vault", generation: 1 } as const; type TestChatInboundHandler = Omit & { handleNotification(notification: ServerNotification): void; diff --git a/tests/features/chat/application/connection/connection-actions.test.ts b/tests/features/chat/application/connection/connection-actions.test.ts index df26677f..41c13296 100644 --- a/tests/features/chat/application/connection/connection-actions.test.ts +++ b/tests/features/chat/application/connection/connection-actions.test.ts @@ -47,7 +47,7 @@ function createActionsHarness({ connected = false } = {}) { configuredCommand: () => "codex", refreshLiveState: vi.fn(), isStaleConnectionError: () => false, - isStaleSharedQueryError: () => false, + isStaleResourceContextError: () => false, notifyConnectionFailed: vi.fn(), }; return { @@ -160,15 +160,15 @@ describe("ChatConnectionActions", () => { expect(refreshAppServerMetadata).not.toHaveBeenCalled(); }); - it("ignores stale shared query failures while refreshing active threads", async () => { + it("ignores stale resource context failures while refreshing active threads", async () => { const { actions, host } = createActionsHarness({ connected: true }); const error = new Error("stale"); vi.mocked(host.refreshSharedThreads).mockRejectedValueOnce(error); - host.isStaleSharedQueryError = vi.fn((candidate) => candidate === error); + host.isStaleResourceContextError = vi.fn((candidate) => candidate === error); await actions.refreshActiveThreads(); - expect(host.isStaleSharedQueryError).toHaveBeenCalledWith(error); + expect(host.isStaleResourceContextError).toHaveBeenCalledWith(error); expect(host.addSystemMessage).not.toHaveBeenCalled(); }); diff --git a/tests/features/chat/application/connection/server-actions.test.ts b/tests/features/chat/application/connection/server-actions.test.ts index a4a44725..6bb62d04 100644 --- a/tests/features/chat/application/connection/server-actions.test.ts +++ b/tests/features/chat/application/connection/server-actions.test.ts @@ -29,7 +29,7 @@ describe("server metadata actions", () => { stateStore, ...metadataCacheHost(), refreshAppServerMetadata: vi.fn().mockResolvedValue(metadata), - isStaleSharedQueryError: () => false, + isStaleResourceContextError: () => false, }); await actions.refreshAppServerMetadata(); @@ -56,7 +56,7 @@ describe("server metadata actions", () => { stateStore, ...metadataCacheHost(), refreshAppServerMetadata: vi.fn().mockResolvedValue(metadata), - isStaleSharedQueryError: () => false, + isStaleResourceContextError: () => false, }); await actions.refreshAppServerMetadata(); @@ -75,7 +75,7 @@ describe("server metadata actions", () => { stateStore, ...metadataCacheHost(cache), refreshAppServerMetadata: vi.fn().mockResolvedValue(null), - isStaleSharedQueryError: () => false, + isStaleResourceContextError: () => false, }); await actions.applyAppServerResourceEvent({ @@ -96,7 +96,7 @@ describe("server metadata actions", () => { stateStore, ...metadataCacheHost(), refreshAppServerMetadata: vi.fn().mockRejectedValue(stale), - isStaleSharedQueryError: (error) => error === stale, + isStaleResourceContextError: (error) => error === stale, }); await expect(actions.refreshAppServerMetadata()).resolves.toBeNull(); @@ -112,7 +112,7 @@ describe("server metadata actions", () => { stateStore, ...metadataCacheHost(), refreshSkills: vi.fn().mockRejectedValue(stale), - isStaleSharedQueryError: (error) => error === stale, + isStaleResourceContextError: (error) => error === stale, }); await actions.applyAppServerResourceEvent({ type: "skills-changed" }); @@ -176,7 +176,7 @@ describe("server diagnostics actions", () => { cache.current = refreshedMetadata; return refreshedMetadata; }), - isStaleSharedQueryError: () => false, + isStaleResourceContextError: () => false, }); const readServerDiagnostics = vi.fn().mockResolvedValue(serverDiagnosticsSnapshot()); const diagnostics = createServerDiagnosticsActions({ @@ -273,7 +273,7 @@ describe("server diagnostics actions", () => { stateStore, ...metadataCache, refreshAppServerMetadata: vi.fn().mockResolvedValue(null), - isStaleSharedQueryError: () => false, + isStaleResourceContextError: () => false, }); const diagnostics = createServerDiagnosticsActions({ stateStore, @@ -438,14 +438,14 @@ function metadataCacheHost(cache: { current: SharedServerMetadata | null } = { c refreshAppServerMetadata: () => Promise; refreshSkills: () => Promise; refreshRateLimits: () => Promise; - isStaleSharedQueryError: (error: unknown) => boolean; + isStaleResourceContextError: (error: unknown) => boolean; } { return { appServerMetadataSnapshot: () => cache.current, refreshAppServerMetadata: vi.fn().mockResolvedValue(null), refreshSkills: vi.fn().mockResolvedValue(null), refreshRateLimits: vi.fn().mockResolvedValue(null), - isStaleSharedQueryError: () => false, + isStaleResourceContextError: () => false, }; } diff --git a/tests/features/chat/application/threads/auto-title-coordinator.test.ts b/tests/features/chat/application/threads/auto-title-coordinator.test.ts index e55b869c..1a1db722 100644 --- a/tests/features/chat/application/threads/auto-title-coordinator.test.ts +++ b/tests/features/chat/application/threads/auto-title-coordinator.test.ts @@ -48,10 +48,13 @@ describe("AutoTitleCoordinator", () => { }); await flushPromises(); - expect(generateThreadTitle).toHaveBeenCalledWith({ - userRequest: "Visible streamed request.", - assistantResponse: "Visible streamed response.", - }); + expect(generateThreadTitle).toHaveBeenCalledWith( + { + userRequest: "Visible streamed request.", + assistantResponse: "Visible streamed response.", + }, + expect.any(AbortSignal), + ); expect(renameThreadRequest).toHaveBeenCalledWith({ threadId: "thread", name: "Visible context title" }); }); @@ -81,10 +84,13 @@ describe("AutoTitleCoordinator", () => { coordinator.maybeAutoTitleThread("thread", "turn", null); await flushPromises(); - expect(generateThreadTitle).toHaveBeenCalledWith({ - userRequest: "Please diagnose auto naming.", - assistantResponse: "I found the regression.", - }); + expect(generateThreadTitle).toHaveBeenCalledWith( + { + userRequest: "Please diagnose auto naming.", + assistantResponse: "I found the regression.", + }, + expect.any(AbortSignal), + ); expect(renameThreadRequest).toHaveBeenCalledWith({ threadId: "thread", name: "Visible context title" }); }); @@ -142,10 +148,38 @@ describe("AutoTitleCoordinator", () => { expect(persistedName).toBe("Manual title"); expect(stateStore.getState().threadList.listedThreads[0]?.name).toBe("Manual title"); }); + + it("retries after A→B→A invalidation without publishing the old title", async () => { + const oldTitle = deferred(); + const renameThreadRequest = vi.fn().mockResolvedValue({}); + const generateThreadTitle = vi.fn().mockReturnValueOnce(oldTitle.promise).mockResolvedValueOnce("Fresh title"); + const { coordinator } = coordinatorFixture({ + currentClient: () => fakeClient({ renameThreadRequest }), + generateThreadTitle, + }); + + coordinator.maybeAutoTitleThread("thread", "turn-a", { userText: "Old request", assistantText: "Old response" }); + await flushPromises(); + coordinator.invalidate(); + coordinator.invalidate(); + coordinator.maybeAutoTitleThread("thread", "turn-a2", { userText: "Fresh request", assistantText: "Fresh response" }); + await flushPromises(); + + expect(renameThreadRequest).toHaveBeenCalledOnce(); + expect(renameThreadRequest).toHaveBeenCalledWith({ threadId: "thread", name: "Fresh title" }); + + oldTitle.resolve("Stale title"); + await flushPromises(); + + expect(renameThreadRequest).toHaveBeenCalledOnce(); + }); }); function coordinatorFixture( - overrides: { currentClient?: () => AppServerClient; generateThreadTitle?: (context: ThreadTitleContext) => Promise } = {}, + overrides: { + currentClient?: () => AppServerClient; + generateThreadTitle?: (context: ThreadTitleContext, signal: AbortSignal) => Promise; + } = {}, ): AutoTitleCoordinatorHost & { coordinator: AutoTitleCoordinator; notifyThreadRenamed: ReturnType; diff --git a/tests/features/chat/application/threads/rename-editor-actions.test.ts b/tests/features/chat/application/threads/rename-editor-actions.test.ts index a1fb1311..77ac0663 100644 --- a/tests/features/chat/application/threads/rename-editor-actions.test.ts +++ b/tests/features/chat/application/threads/rename-editor-actions.test.ts @@ -153,6 +153,26 @@ describe("ThreadRenameEditorActions", () => { expect(actions.editState("thread")).toEqual({ draft: "New generated title", generating: false }); }); + + it("does not publish or report title work invalidated by a context replacement", async () => { + const oldTitle = deferred(); + const generateThreadTitle = vi.fn().mockReturnValueOnce(oldTitle.promise).mockResolvedValueOnce("Fresh title"); + const addSystemMessage = vi.fn(); + const { actions } = actionsFixture({ generateThreadTitle, addSystemMessage }); + + actions.start("thread"); + const staleAutoName = actions.autoNameDraft("thread"); + await flushPromises(); + actions.invalidate(); + + actions.start("thread"); + await actions.autoNameDraft("thread"); + oldTitle.resolve("Stale title"); + await staleAutoName; + + expect(actions.editState("thread")).toEqual({ draft: "Fresh title", generating: false }); + expect(addSystemMessage).not.toHaveBeenCalled(); + }); }); function actionsFixture( diff --git a/tests/features/chat/host/session-runtime.test.ts b/tests/features/chat/host/session-runtime.test.ts index ce6b3c41..29f6100f 100644 --- a/tests/features/chat/host/session-runtime.test.ts +++ b/tests/features/chat/host/session-runtime.test.ts @@ -2,7 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { StaleAppServerSharedQueryContextError } from "../../../../src/app-server/query/shared-queries"; +import { StaleAppServerResourceContextError } from "../../../../src/app-server/query/resource-store"; import type { ModelMetadata } from "../../../../src/domain/catalog/metadata"; import type { Thread } from "../../../../src/domain/threads/model"; import { activeThreadId } from "../../../../src/features/chat/application/state/root-reducer"; @@ -46,6 +46,16 @@ describe("ChatPanelSessionRuntime actions", () => { expect(invalidateHistory).toHaveBeenCalledOnce(); }); + it("clears rename work before replacing the app-server context", () => { + const { runtime, stateStore } = sessionRuntimeFixture(); + stateStore.dispatch({ type: "thread-list/applied", threads: [threadFixture({ id: "thread-1" })] }); + stateStore.dispatch({ type: "ui/rename-started", threadId: "thread-1", draft: "Draft" }); + + runtime.actions.prepareAppServerContextChange(); + + expect(stateStore.getState().ui.rename).toEqual({ kind: "idle" }); + }); + it("refreshes shared threads inside the runtime connection bundle", async () => { const thread = threadFixture({ id: "thread-1", preview: "From catalog" }); const refresh = vi.fn().mockResolvedValue([thread]); @@ -66,7 +76,7 @@ describe("ChatPanelSessionRuntime actions", () => { }); it("treats stale shared thread refreshes as runtime-local no-ops", async () => { - const refresh = vi.fn().mockRejectedValue(new StaleAppServerSharedQueryContextError()); + const refresh = vi.fn().mockRejectedValue(new StaleAppServerResourceContextError()); const { runtime, stateStore } = sessionRuntimeFixture({ environment: { plugin: { @@ -377,6 +387,7 @@ describe("ChatPanelSessionRuntime actions", () => { overrides: Partial = {}, ): ChatPanelEnvironment["plugin"]["appServerQueries"] { return { + contextLease: vi.fn(() => ({ context: { codexPath: "codex", vaultPath: "/vault" }, generation: 1 })), appServerMetadataSnapshot: vi.fn(() => null), refreshAppServerMetadata: vi.fn().mockResolvedValue(null), refreshSkills: vi.fn().mockResolvedValue(null), diff --git a/tests/features/chat/host/view-connection.test.ts b/tests/features/chat/host/view-connection.test.ts index 60620ba9..2c3e2c55 100644 --- a/tests/features/chat/host/view-connection.test.ts +++ b/tests/features/chat/host/view-connection.test.ts @@ -1406,6 +1406,7 @@ function chatHost(overrides: ChatHostFixtureOverrides = {}): TestCodexChatHost { openSideChat: overrides.openSideChat ?? vi.fn().mockResolvedValue(undefined), }, appServerQueries: { + contextLease: () => ({ context: { codexPath: settings.codexPath, vaultPath }, generation: 1 }), appServerMetadataSnapshot: overrides.appServerMetadataSnapshot ?? vi.fn(() => metadata), refreshAppServerMetadata: overrides.refreshAppServerMetadata ?? diff --git a/tests/features/threads-view/view.test.ts b/tests/features/threads-view/view.test.ts index f421b726..411f13b9 100644 --- a/tests/features/threads-view/view.test.ts +++ b/tests/features/threads-view/view.test.ts @@ -569,7 +569,9 @@ describe("CodexThreadsView", () => { await waitForAsyncWork(() => { expect(namingMock.generateThreadTitleWithCodex).toHaveBeenCalledOnce(); }); + const generationSignal = namingMock.generateThreadTitleWithCodex.mock.calls[0]?.[4]?.signal as AbortSignal | undefined; await view.onClose(); + expect(generationSignal?.aborted).toBe(true); generatedTitle.resolve("Late title"); for (let index = 0; index < 10; index += 1) await Promise.resolve(); diff --git a/tests/features/threads/catalog/thread-catalog.test.ts b/tests/features/threads/catalog/thread-catalog.test.ts index 1aaf9956..32a452fd 100644 --- a/tests/features/threads/catalog/thread-catalog.test.ts +++ b/tests/features/threads/catalog/thread-catalog.test.ts @@ -1,9 +1,9 @@ import { describe, expect, it, type Mock, vi } from "vitest"; import { AppServerQueryCache } from "../../../../src/app-server/query/cache"; -import { type AppServerQueryContext, appServerQueryContextKey } from "../../../../src/app-server/query/keys"; +import type { AppServerQueryContext, AppServerQueryContextIdentity } from "../../../../src/app-server/query/keys"; import type { ObservedResult, ObservedResultListener } from "../../../../src/app-server/query/observed-result"; -import { AppServerSharedQueries } from "../../../../src/app-server/query/shared-queries"; +import { AppServerResourceStore } from "../../../../src/app-server/query/resource-store"; import type { Thread } from "../../../../src/domain/threads/model"; import { createThreadCatalog, @@ -180,109 +180,29 @@ describe("ThreadCatalog", () => { expect(catalog.activeSnapshot()).toEqual([thread("other")]); }); - it("scopes pending lifecycle facts by app-server query context", () => { - const context = { codexPath: "codex-a", vaultPath: "/vault" }; - const { catalog } = catalogFixture({ context: () => context }); + 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")]); - catalog.apply({ type: "thread-started", thread: thread("started-a") }); - expect(catalog.activeSnapshot()).toEqual([thread("started-a")]); - - context.codexPath = "codex-b"; + replaceContext({ codexPath: "codex-b", vaultPath: "/vault" }); expect(catalog.activeSnapshot()).toBeNull(); - receiveActive(catalog, [thread("thread-b")]); - expect(catalog.activeSnapshot()).toEqual([thread("thread-b")]); - - context.codexPath = "codex-a"; - expect(catalog.activeSnapshot()).toEqual([thread("started-a")]); - }); - - it("applies a connection event to its captured context after the current context changes", () => { - const context = { codexPath: "codex-a", vaultPath: "/vault" }; - const { catalog } = catalogFixture({ context: () => context }); - const connectionContext = { ...context }; - - context.codexPath = "codex-b"; - catalog.applyConnectionEvent(connectionContext, { type: "thread-started", thread: thread("started-a") }); + replaceContext({ codexPath: "codex-a", vaultPath: "/vault" }); expect(catalog.activeSnapshot()).toBeNull(); - context.codexPath = "codex-a"; - expect(catalog.activeSnapshot()).toEqual([thread("started-a")]); }); - it("reads captured-context snapshots when applying connection events", async () => { - const context = { codexPath: "codex-a", vaultPath: "/vault" }; - const connectionContext = { ...context }; - const { catalog } = catalogFixture({ - context: () => context, - fetchThreads: (source) => Promise.resolve(source.codexPath === "codex-a" ? [thread("thread-a")] : []), + it("does not publish connection events captured by an earlier lease", () => { + const { catalog, contextIdentity, replaceContext } = catalogFixture({ + context: () => ({ codexPath: "codex-a", vaultPath: "/vault" }), }); - await catalog.refreshActive(); + const sourceA1 = contextIdentity(); - context.codexPath = "codex-b"; - catalog.applyConnectionEvent(connectionContext, { type: "thread-renamed", threadId: "thread-a", name: "Renamed A" }); + 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(); - context.codexPath = "codex-a"; - expect(catalog.activeSnapshot()).toEqual([thread("thread-a", false, { name: "Renamed A" })]); - }); - - it("clears pending lifecycle facts across every app-server query context", () => { - const context = { codexPath: "codex-a", vaultPath: "/vault" }; - const { catalog, cache } = catalogFixture({ context: () => context }); - catalog.apply({ type: "thread-started", thread: thread("started-a") }); - context.codexPath = "codex-b"; - catalog.apply({ type: "thread-started", thread: thread("started-b") }); - - cache.clear(); - catalog.clear(); - - expect(catalog.activeSnapshot()).toBeNull(); - context.codexPath = "codex-a"; - expect(catalog.activeSnapshot()).toBeNull(); - }); - - it("retains unacknowledged lifecycle facts across connection contexts", () => { - const context = { codexPath: "codex-a", vaultPath: "/vault" }; - const { catalog } = catalogFixture({ context: () => context }); - catalog.apply({ type: "thread-started", thread: thread("started-a") }); - - for (const suffix of ["b", "c", "d", "e"]) { - context.codexPath = `codex-${suffix}`; - catalog.activeSnapshot(); - } - - context.codexPath = "codex-a"; - receiveActive(catalog, [thread("server-a")]); - - expect(catalog.activeSnapshot()).toEqual([thread("started-a"), thread("server-a")]); - }); - - it("prunes inactive lifecycle facts without discarding store snapshots", () => { - const context = { codexPath: "codex-a", vaultPath: "/vault" }; - const { catalog } = catalogFixture({ context: () => context }); - receiveActive(catalog, [thread("server-a")]); - - context.codexPath = "codex-b"; - expect(catalog.activeSnapshot()).toBeNull(); - context.codexPath = "codex-a"; - - expect(catalog.activeSnapshot()).toEqual([thread("server-a")]); - }); - - it("prunes revisited contexts after pending lifecycle facts settle", () => { - const context = { codexPath: "codex-a", vaultPath: "/vault" }; - const { catalog } = catalogFixture({ context: () => context }); - catalog.apply({ type: "thread-started", thread: thread("started-a") }); - context.codexPath = "codex-b"; - catalog.apply({ type: "thread-started", thread: thread("started-b") }); - - context.codexPath = "codex-a"; - receiveActive(catalog, [thread("started-a")]); - context.codexPath = "codex-b"; - receiveActive(catalog, [thread("started-b")]); - context.codexPath = "codex-a"; - - expect(catalog.activeSnapshot()).toEqual([thread("started-a")]); }); it("preserves raw query status when lifecycle overlays publish", async () => { @@ -634,11 +554,9 @@ function catalogFixture( ) { const cache = cacheWithThreads(options.fetchThreads ?? (() => Promise.resolve([]))); const context = options.context ?? (() => ({ codexPath: "codex", vaultPath: "/vault" })); - const queries = new AppServerSharedQueries({ - cache, - context, - }); - const store = new TestThreadCatalogStore(cache, queries, context); + const queries = new AppServerResourceStore({ cache }); + queries.initialize(context()); + const store = new TestThreadCatalogStore(queries); const catalog = createThreadCatalog( options.onEventApplied ? { @@ -650,7 +568,12 @@ function catalogFixture( catalogStores.set(catalog, store); catalog.observeActive(() => undefined); catalog.observeArchived(() => undefined); - return { cache, catalog }; + return { + cache, + catalog, + contextIdentity: () => queries.contextIdentity(), + replaceContext: (nextContext: AppServerQueryContext) => queries.replaceContext(nextContext), + }; } class TestThreadCatalogStore { @@ -659,36 +582,24 @@ class TestThreadCatalogStore { private readonly activeObservers = new Set>(); private readonly archivedObservers = new Set>(); - constructor( - private readonly cache: AppServerQueryCache, - private readonly queries: AppServerSharedQueries, - private readonly currentContext: () => AppServerQueryContext, - ) {} + constructor(private readonly queries: AppServerResourceStore) {} contextKey(): string { - return appServerQueryContextKey(this.currentContext()); + return this.queries.contextKey(); } - contextKeyFor(context: AppServerQueryContext): string { - return appServerQueryContextKey(context); + contextKeyFor(context: AppServerQueryContextIdentity): string { + return this.queries.contextKeyFor(context); } activeThreadsSnapshot(): readonly Thread[] | null { return this.activeSnapshots.get(this.contextKey()) ?? this.queries.activeThreadsSnapshot(); } - activeThreadsSnapshotFor(context: AppServerQueryContext): readonly Thread[] | null { - return this.activeSnapshots.get(this.contextKeyFor(context)) ?? this.cache.activeThreadsSnapshot(context); - } - archivedThreadsSnapshot(): readonly Thread[] | null { return this.archivedSnapshots.get(this.contextKey()) ?? this.queries.archivedThreadsSnapshot(); } - archivedThreadsSnapshotFor(context: AppServerQueryContext): readonly Thread[] | null { - return this.archivedSnapshots.get(this.contextKeyFor(context)) ?? this.cache.archivedThreadsSnapshot(context); - } - async fetchAllActiveThreads(): Promise { return this.receiveLoadedActive(await this.queries.fetchAllActiveThreads()); } @@ -705,22 +616,10 @@ class TestThreadCatalogStore { return this.receiveLoadedActive(await this.queries.refreshActiveThreads()); } - async refreshActiveThreadsFor(context: AppServerQueryContext): Promise { - const threads = await this.cache.refreshActiveThreads(context); - this.activeSnapshots.delete(this.contextKeyFor(context)); - return threads; - } - async refreshArchivedThreads(): Promise { return this.receiveLoadedArchived(await this.queries.refreshArchivedThreads()); } - async refreshArchivedThreadsFor(context: AppServerQueryContext): Promise { - const threads = await this.cache.refreshArchivedThreads(context); - this.archivedSnapshots.delete(this.contextKeyFor(context)); - return threads; - } - observeActiveThreadsResult(observer: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void { this.activeObservers.add(observer); const unsubscribe = this.queries.observeActiveThreadsResult((result) => { diff --git a/tests/features/threads/workflows/thread-title-service.test.ts b/tests/features/threads/workflows/thread-title-service.test.ts index ffc16f7e..a168fd51 100644 --- a/tests/features/threads/workflows/thread-title-service.test.ts +++ b/tests/features/threads/workflows/thread-title-service.test.ts @@ -23,7 +23,7 @@ describe("ThreadTitleService", () => { await expect(service.generateTitle("thread")).resolves.toBe("Generated title"); expect(withClient).not.toHaveBeenCalled(); - expect(generateThreadTitle).toHaveBeenCalledWith(titleContext("visible request", "visible response")); + expect(generateThreadTitle).toHaveBeenCalledWith(titleContext("visible request", "visible response"), expect.any(AbortSignal)); }); it("throws the existing unavailable-context error when no context can be resolved", async () => { @@ -51,6 +51,36 @@ describe("ThreadTitleService", () => { titleContext("summary turn", "summary answer"), ); }); + + it("cancels stale title work and starts later work with a fresh signal", async () => { + let resolveOldTitle!: (title: string | null) => void; + const generateTitle = vi + .fn() + .mockImplementationOnce( + (_context: ThreadTitleContext, _signal: AbortSignal) => + new Promise((resolve) => { + resolveOldTitle = resolve; + }), + ) + .mockResolvedValueOnce("Fresh title"); + const service = titleService({ + visibleContext: () => titleContext("request", "response"), + transport: { + persistedContext: vi.fn().mockResolvedValue(null), + generateTitle, + }, + }); + + const staleTitle = service.generateTitle("thread"); + await Promise.resolve(); + const staleSignal = generateTitle.mock.calls[0]?.[1]; + service.invalidate(); + + expect(staleSignal?.aborted).toBe(true); + await expect(service.generateTitle("thread")).resolves.toBe("Fresh title"); + resolveOldTitle("Stale title"); + await expect(staleTitle).rejects.toThrow("Thread title generation cancelled."); + }); }); function titleService(options: Partial = {}): ThreadTitleService { diff --git a/tests/main.test.ts b/tests/main.test.ts index 3780822e..f4fbe9bf 100644 --- a/tests/main.test.ts +++ b/tests/main.test.ts @@ -637,7 +637,7 @@ describe("CodexPanelPlugin workspace panel reconciliation", () => { let resolveFirst!: (threads: Thread[]) => void; const runWithAppServerClient = vi.spyOn(connectedView.surface, "runWithAppServerClient"); const plugin = await pluginWithLeaves([connectedLeaf]); - plugin.settings.codexPath = "codex-a"; + await publishCodexPath(plugin, "codex-a"); runWithAppServerClient.mockImplementation((operation) => operation( threadListClient(() => @@ -651,8 +651,9 @@ describe("CodexPanelPlugin workspace panel reconciliation", () => { ); const first = threadCatalog(plugin).refreshActive(); + const staleFirst = expect(first).rejects.toThrow("Codex app-server resource context changed while loading."); await flushMicrotasks(); - plugin.settings.codexPath = "codex-b"; + await publishCodexPath(plugin, "codex-b"); const second = threadCatalog(plugin).refreshActive(); await flushMicrotasks(); @@ -661,9 +662,9 @@ describe("CodexPanelPlugin workspace panel reconciliation", () => { expect(threadCatalog(plugin).activeSnapshot()).toEqual([thread("second")]); resolveFirst([thread("first")]); - await expect(first).rejects.toThrow("Codex app-server query context changed while loading shared queries."); + await staleFirst; expect(threadCatalog(plugin).activeSnapshot()).toEqual([thread("second")]); - plugin.settings.codexPath = "codex-a"; + await publishCodexPath(plugin, "codex-a"); expect(threadCatalog(plugin).activeSnapshot()).toBeNull(); }); @@ -680,7 +681,7 @@ describe("CodexPanelPlugin workspace panel reconciliation", () => { operation(threadListClient(() => Promise.resolve([thread("matching-context")]))), ); const plugin = await pluginWithLeaves([connectedLeaf]); - plugin.settings.codexPath = "codex-b"; + await publishCodexPath(plugin, "codex-b"); await expect(threadCatalog(plugin).refreshActive()).resolves.toEqual([thread("matching-context")]); @@ -746,28 +747,51 @@ describe("CodexPanelPlugin workspace panel reconciliation", () => { const result = deferred(); vi.spyOn(connectedView.surface, "runWithAppServerClient").mockReturnValue(result.promise); const plugin = await pluginWithLeaves([connectedLeaf]); - plugin.settings.codexPath = "codex-a"; + await publishCodexPath(plugin, "codex-a"); const operation = plugin.runtime.withClient(() => Promise.resolve("unused")); - plugin.settings.codexPath = "codex-b"; + await publishCodexPath(plugin, "codex-b"); result.resolve("stale-result"); - await expect(operation).rejects.toThrow("Codex app-server query context changed while loading shared queries."); + await expect(operation).rejects.toThrow("Codex app-server resource context changed while loading."); }); it("rejects a short-lived result when the app-server context changes during the operation", async () => { const result = deferred(); withShortLivedAppServerClientMock.mockReturnValue(result.promise); const plugin = await pluginWithLeaves([]); - plugin.settings.codexPath = "codex-a"; + await publishCodexPath(plugin, "codex-a"); const operation = plugin.runtime.withClient(() => Promise.resolve("unused"), { serverRequests: { kind: "reject", message: "test" }, }); - plugin.settings.codexPath = "codex-b"; + await publishCodexPath(plugin, "codex-b"); result.resolve("stale-result"); - await expect(operation).rejects.toThrow("Codex app-server query context changed while loading shared queries."); + await expect(operation).rejects.toThrow("Codex app-server resource context changed while loading."); + }); + + it("publishes a persisted context and its new resource lease atomically", async () => { + const plugin = await pluginWithLeaves([]); + const firstLease = plugin.runtime.appServerContextLease(); + const save = deferred(); + const saveSettings = vi.spyOn(plugin, "saveSettings").mockReturnValue(save.promise); + + const publication = plugin.runtime.settingTabHost().publishSettings({ ...plugin.settings, codexPath: "codex-next" }); + await Promise.resolve(); + + expect(saveSettings).toHaveBeenCalledWith(expect.objectContaining({ codexPath: "codex-next" })); + expect(plugin.settings.codexPath).toBe(firstLease.context.codexPath); + expect(plugin.runtime.appServerContextLease()).toBe(firstLease); + + save.resolve(undefined); + await publication; + + expect(plugin.settings.codexPath).toBe("codex-next"); + expect(plugin.runtime.appServerContextLease()).toEqual({ + context: { codexPath: "codex-next", vaultPath: "/vault" }, + generation: firstLease.generation + 1, + }); }); it("coordinates context replacement with every open Threads view", async () => { @@ -781,13 +805,8 @@ describe("CodexPanelPlugin workspace panel reconciliation", () => { const threadsLeaf = leaf(); threadsLeaf.view = threadsView; const plugin = await pluginWithLeaves([], { threadsLeaves: [threadsLeaf] }); - const settingsHost = plugin.runtime.settingTabHost(); - - settingsHost.prepareAppServerContextChange(); + await publishCodexPath(plugin, "codex-next"); expect(prepareAppServerContextChange).toHaveBeenCalledOnce(); - - plugin.settings.codexPath = "codex-next"; - settingsHost.refreshOpenViews(); expect(refreshSettings).toHaveBeenCalledOnce(); }); @@ -796,7 +815,7 @@ describe("CodexPanelPlugin workspace panel reconciliation", () => { const closeAll = vi.fn(); plugin.runtime.setSelectionRewriteController({ closeAll }); - plugin.runtime.settingTabHost().prepareAppServerContextChange(); + await publishCodexPath(plugin, "codex-next"); expect(closeAll).toHaveBeenCalledOnce(); }); @@ -860,10 +879,16 @@ async function pluginWithLeaves(leaves: ReturnType[], options: { th } as never, {} as never, ); + plugin.settings = { ...DEFAULT_SETTINGS }; plugin.vaultPath = "/vault"; + plugin.runtime.initialize(); return plugin; } +async function publishCodexPath(plugin: CodexPanelPlugin, codexPath: string): Promise { + await plugin.runtime.settingTabHost().publishSettings({ ...plugin.settings, codexPath }); +} + type TestLeaf = ReturnType; function leaf(options: { state?: Record } = {}) { @@ -937,6 +962,7 @@ function chatHostFixture(): CodexChatHost { openSideChat: vi.fn(), }, appServerQueries: { + contextLease: () => ({ context: { codexPath: settings.codexPath, vaultPath: "/vault" }, generation: 1 }), appServerMetadataSnapshot: vi.fn(() => null), refreshAppServerMetadata: vi.fn(() => Promise.resolve(null)), refreshSkills: vi.fn(() => Promise.resolve(null)), diff --git a/tests/settings/settings-tab.test.ts b/tests/settings/settings-tab.test.ts index 1a064326..4a4de9b3 100644 --- a/tests/settings/settings-tab.test.ts +++ b/tests/settings/settings-tab.test.ts @@ -14,7 +14,7 @@ import { createSettingsAppServerDynamicData } from "../../src/settings/app-serve import type { SettingsDynamicDataAccess } from "../../src/settings/dynamic-data"; import { SettingsDynamicSectionsController, type SettingsDynamicSectionsSnapshot } from "../../src/settings/dynamic-sections-controller"; import type { CodexPanelSettingTabHost } from "../../src/settings/host"; -import { DEFAULT_SETTINGS } from "../../src/settings/model"; +import { type CodexPanelSettings, DEFAULT_SETTINGS } from "../../src/settings/model"; import { CodexPanelSettingTab } from "../../src/settings/tab.obsidian"; import { notices } from "../mocks/obsidian"; import { deferred } from "../support/async"; @@ -359,7 +359,6 @@ describe("settings tab", () => { it("clears dynamic sections when the Codex executable changes", async () => { const saveSettings = vi.fn().mockResolvedValue(undefined); - const notifyContextChanged = vi.fn(); const refreshOpenViews = vi.fn(); const oldClient = settingsClient({ models: [model("gpt-old")], @@ -376,7 +375,7 @@ describe("settings tab", () => { .fn() .mockResolvedValueOnce([panelThread({ id: "thread-old", preview: "Old archived", archived: true })]) .mockResolvedValueOnce([panelThread({ id: "thread-new", preview: "New archived", archived: true })]); - const tab = newSettingsTab({ saveSettings, notifyContextChanged, refreshOpenViews, fetchModels, refreshModels, refreshArchived }); + const tab = newSettingsTab({ saveSettings, refreshOpenViews, fetchModels, refreshModels, refreshArchived }); tab.display(); await flushPromises(); @@ -391,14 +390,12 @@ describe("settings tab", () => { await flushPromises(); expect(saveSettings).not.toHaveBeenCalled(); - expect(notifyContextChanged).not.toHaveBeenCalled(); expect(refreshOpenViews).not.toHaveBeenCalled(); codexInput.dispatchEvent(new FocusEvent("blur")); await flushPromises(); expect(saveSettings).toHaveBeenCalledOnce(); - expect(notifyContextChanged).toHaveBeenCalledOnce(); expect(refreshOpenViews).toHaveBeenCalledOnce(); expect(withShortLivedAppServerClientMock).toHaveBeenCalledTimes(1); expect(tab.containerEl.textContent).not.toContain("gpt-old"); @@ -1149,14 +1146,13 @@ function expectRequestTimes(client: SettingsRequestClient, method: string, times function newSettingsTab( options: { - saveSettings?: CodexPanelSettingTabHost["saveSettings"]; + saveSettings?: (settings: CodexPanelSettings) => Promise; prepareAppServerContextChange?: () => void; sendShortcut?: "enter" | "mod-enter"; modelsSnapshot?: ModelMetadata[]; fetchModels?: () => Promise; refreshModels?: () => Promise; observeModels?: SettingsDynamicDataAccess["observeModelsResult"]; - notifyContextChanged?: () => void; refreshOpenViews?: () => void; archivedThreads?: Thread[]; archivedSnapshot?: Thread[] | null; @@ -1177,14 +1173,13 @@ function newSettingsTab( function settingsTabHost( options: { - saveSettings?: CodexPanelSettingTabHost["saveSettings"]; + saveSettings?: (settings: CodexPanelSettings) => Promise; prepareAppServerContextChange?: () => void; sendShortcut?: "enter" | "mod-enter"; modelsSnapshot?: ModelMetadata[]; fetchModels?: () => Promise; refreshModels?: () => Promise; observeModels?: SettingsDynamicDataAccess["observeModelsResult"]; - notifyContextChanged?: () => void; refreshOpenViews?: () => void; archivedThreads?: Thread[]; archivedSnapshot?: Thread[] | null; @@ -1214,7 +1209,6 @@ function settingsTabHost( fetchModels: options.fetchModels ?? vi.fn().mockResolvedValue(options.modelsSnapshot ?? []), refreshModels: options.refreshModels ?? vi.fn().mockResolvedValue(options.modelsSnapshot ?? []), observeModelsResult: options.observeModels ?? vi.fn(() => () => undefined), - notifyContextChanged: options.notifyContextChanged ?? vi.fn(), }; const threadCatalog = { archivedSnapshot: vi.fn(() => options.archivedSnapshot ?? null), @@ -1235,9 +1229,15 @@ function settingsTabHost( appServerQueries, threadCatalog, }), - saveSettings: options.saveSettings ?? vi.fn().mockResolvedValue(undefined), - prepareAppServerContextChange: options.prepareAppServerContextChange ?? vi.fn(), - refreshOpenViews: options.refreshOpenViews ?? vi.fn(), + publishSettings: async (nextSettings) => { + const previousSettings = { ...settings }; + await (options.saveSettings ?? vi.fn().mockResolvedValue(undefined))(nextSettings); + const appServerContextReplaced = previousSettings.codexPath !== nextSettings.codexPath; + if (appServerContextReplaced) options.prepareAppServerContextChange?.(); + Object.assign(settings, nextSettings); + if (appServerContextReplaced || previousSettings.showToolbar !== nextSettings.showToolbar) options.refreshOpenViews?.(); + return { appServerContextReplaced }; + }, }; }