diff --git a/src/app-server/query/cache.ts b/src/app-server/query/cache.ts index 53e50539..98d775c5 100644 --- a/src/app-server/query/cache.ts +++ b/src/app-server/query/cache.ts @@ -1,11 +1,14 @@ import { QueryClient, QueryObserver, type QueryObserverResult } from "@tanstack/query-core"; -import type { ModelMetadata } from "../../domain/catalog/metadata"; +import type { ModelMetadata, SkillMetadata } from "../../domain/catalog/metadata"; +import { cloneRuntimeConfigSnapshot, type RuntimeConfigSnapshot } from "../../domain/runtime/config"; +import type { RateLimitSnapshot } from "../../domain/runtime/metrics"; +import type { RuntimePermissionProfileSummary } from "../../domain/runtime/permissions"; import { createServerDiagnostics, + type DiagnosticProbeResult, diagnosticProbeError, diagnosticProbeOk, diagnosticsWithProbe, - metadataResourceDiagnostics, } from "../../domain/server/diagnostics"; import type { SharedServerMetadata } from "../../domain/server/metadata"; import type { Thread } from "../../domain/threads/model"; @@ -18,9 +21,13 @@ import { listThreads, readThreadPage } from "../services/threads"; import { type AppServerQueryContext, activeThreadsQueryKey, - appServerMetadataQueryKey, appServerModelsQueryKey, + appServerPermissionProfilesQueryKey, appServerQueryContextIsComplete, + appServerQueryContextKey, + appServerRateLimitsQueryKey, + appServerRuntimeConfigQueryKey, + appServerSkillsQueryKey, archivedThreadsQueryKey, cloneAppServerQueryContext, } from "./keys"; @@ -49,15 +56,31 @@ interface AppServerQueryOptions { } type ThreadListKind = "active" | "archived"; -export type MetadataResourceKind = "skills" | "rateLimits"; +interface MetadataResourceSnapshot { + readonly value: T; + readonly probe: DiagnosticProbeResult; +} + +type MetadataResourceKind = "skills" | "permissionProfiles" | "rateLimits"; +type MetadataNotificationResourceKind = "skills" | "rateLimits"; +type MetadataResourceValue = readonly SkillMetadata[] | readonly RuntimePermissionProfileSummary[] | RateLimitSnapshot | null; + +interface MetadataNotificationRefresh { + dirty: boolean; + readonly generation: number; + promise: Promise; +} export class AppServerQueryCache { private readonly client: QueryClient; private readonly clientRunner: AppServerQueryClientRunner | null; private readonly activeThreadCursors = new Map(); private readonly activeThreadRevisions = new Map(); - private readonly metadataRevisions = new Map(); - private readonly metadataWriteRevisions = new Map(); + private readonly metadataRefreshes = new Map(); + private readonly metadataProjectionListeners = new Map void>>(); + private readonly metadataResourceFetches = new Map>>(); + private readonly metadataNotificationRefreshes = new Map(); + private generation = 0; constructor(options: { client?: QueryClient; clientRunner?: AppServerQueryClientRunner } = {}) { this.client = options.client ?? createAppServerQueryClient(); @@ -67,8 +90,11 @@ export class AppServerQueryCache { clear(): void { this.activeThreadCursors.clear(); this.activeThreadRevisions.clear(); - this.metadataRevisions.clear(); - this.metadataWriteRevisions.clear(); + this.metadataRefreshes.clear(); + this.metadataProjectionListeners.clear(); + this.metadataResourceFetches.clear(); + this.metadataNotificationRefreshes.clear(); + this.generation += 1; this.client.clear(); } @@ -183,8 +209,22 @@ export class AppServerQueryCache { appServerMetadataSnapshot(context: AppServerQueryContext): SharedServerMetadata | null { if (!appServerQueryContextIsComplete(context)) return null; - const metadata = this.client.getQueryData(appServerMetadataQueryKey(context)); - return metadata ? cloneSharedServerMetadata(metadata) : null; + const runtimeConfig = this.client.getQueryData(appServerRuntimeConfigQueryKey(context)); + if (!runtimeConfig) return null; + const skills = this.metadataResourceState(context, "skills"); + const permissionProfiles = this.metadataResourceState(context, "permissionProfiles"); + const rateLimits = this.metadataResourceState(context, "rateLimits"); + const diagnostics = [this.modelsProbe(context), skills.probe, permissionProfiles.probe, rateLimits.probe].reduce( + (current, probe) => diagnosticsWithProbe(current, probe), + createServerDiagnostics(), + ); + return cloneSharedServerMetadata({ + runtimeConfig, + availableSkills: skills.value ?? [], + availablePermissionProfiles: permissionProfiles.value ?? [], + rateLimit: rateLimits.value ?? null, + serverDiagnostics: diagnostics, + }); } observeAppServerMetadataResult( @@ -192,69 +232,86 @@ export class AppServerQueryCache { listener: ObservedResultListener, options: { emitCurrent?: boolean } = {}, ): () => void { - return this.observeQueryResult(this.appServerMetadataQueryOptions(context), cloneSharedServerMetadata, listener, options); - } - - async refreshAppServerMetadata( - context: AppServerQueryContext, - options: { forceSkills?: boolean } = {}, - ): Promise { const refreshContext = cloneAppServerQueryContext(context); - if (!appServerQueryContextIsComplete(refreshContext)) { - return null; - } - this.beginMetadataResourceRefresh(refreshContext, "skills"); - this.beginMetadataResourceRefresh(refreshContext, "rateLimits"); - const key = appServerMetadataQueryKey(refreshContext); - await Promise.all([ - this.client.invalidateQueries({ queryKey: key }), - this.client.invalidateQueries({ queryKey: appServerModelsQueryKey(refreshContext) }), - ]); - const metadata = await this.client.fetchQuery(this.appServerMetadataQueryOptions(refreshContext, options)); - return cloneSharedServerMetadata(metadata); + const generation = this.generation; + const observers = [ + new QueryObserver(this.client, { ...this.runtimeConfigQueryOptions(refreshContext), enabled: false }), + new QueryObserver(this.client, { ...this.skillsQueryOptions(refreshContext), enabled: false }), + new QueryObserver(this.client, { ...this.permissionProfilesQueryOptions(refreshContext), enabled: false }), + new QueryObserver(this.client, { ...this.rateLimitsQueryOptions(refreshContext), enabled: false }), + new QueryObserver(this.client, { ...this.modelsQueryOptions(refreshContext), enabled: false }), + ]; + let queued = false; + let disposed = false; + const emit = (): void => { + queued = false; + if (disposed || generation !== this.generation) return; + const metadata = this.appServerMetadataSnapshot(refreshContext); + const runtimeState = this.client.getQueryState(appServerRuntimeConfigQueryKey(refreshContext)); + listener({ + value: metadata, + error: runtimeState?.error instanceof Error ? runtimeState.error : null, + isFetching: observers.some((observer) => observer.getCurrentResult().isFetching), + }); + }; + const schedule = (): void => { + if (disposed || generation !== this.generation) return; + if ((this.metadataRefreshes.get(appServerQueryContextKey(refreshContext)) ?? 0) > 0 || queued) return; + queued = true; + queueMicrotask(emit); + }; + const contextKey = appServerQueryContextKey(refreshContext); + const projectionListeners = this.metadataProjectionListeners.get(contextKey) ?? new Set<() => void>(); + projectionListeners.add(schedule); + this.metadataProjectionListeners.set(contextKey, projectionListeners); + const unsubscribers = observers.map((observer) => observer.subscribe(schedule)); + if (options.emitCurrent ?? true) emit(); + return () => { + disposed = true; + for (const unsubscribe of unsubscribers) unsubscribe(); + projectionListeners.delete(schedule); + if (projectionListeners.size === 0) this.metadataProjectionListeners.delete(contextKey); + }; } - writeAppServerMetadata(context: AppServerQueryContext, metadata: SharedServerMetadata): SharedServerMetadata | null { - if (!appServerQueryContextIsComplete(context)) return null; - const next = metadataWithLastKnownGood(metadata, this.appServerMetadataSnapshot(context)); - this.client.setQueryData(appServerMetadataQueryKey(context), cloneSharedServerMetadata(next)); - this.bumpMetadataRevision(context); - this.bumpMetadataWriteRevision(context); - return cloneSharedServerMetadata(next); + async refreshAppServerMetadata(context: AppServerQueryContext): Promise { + const refreshContext = cloneAppServerQueryContext(context); + if (!appServerQueryContextIsComplete(refreshContext)) return null; + return this.runMetadataRefresh(refreshContext, async () => { + const runtimeResult = this.fetchRuntimeConfig(refreshContext, true).then( + () => ({ ok: true as const }), + (error: unknown) => ({ ok: false as const, error }), + ); + const [, runtime] = await Promise.all([ + Promise.allSettled([ + this.fetchMetadataResource(refreshContext, "skills", true), + this.fetchMetadataResource(refreshContext, "permissionProfiles", true), + this.fetchMetadataResource(refreshContext, "rateLimits", true), + this.fetchModels(refreshContext, { force: true }), + ]), + runtimeResult, + ]); + if (!runtime.ok) throw runtime.error; + return this.appServerMetadataSnapshot(refreshContext); + }); } - updateAppServerMetadata( - context: AppServerQueryContext, - updater: (metadata: SharedServerMetadata | null) => SharedServerMetadata | null, - resource?: MetadataResourceKind, - ): SharedServerMetadata | null { - if (!appServerQueryContextIsComplete(context)) return null; - const next = updater(this.appServerMetadataSnapshot(context)); - if (!next) return null; - const merged = metadataWithLastKnownGood(next, this.appServerMetadataSnapshot(context)); - this.client.setQueryData(appServerMetadataQueryKey(context), cloneSharedServerMetadata(merged)); - if (resource) this.beginMetadataResourceRefresh(context, resource); - else this.bumpMetadataRevision(context); - this.bumpMetadataWriteRevision(context); - return cloneSharedServerMetadata(merged); + async refreshSkills(context: AppServerQueryContext): Promise { + const refreshContext = cloneAppServerQueryContext(context); + if (!appServerQueryContextIsComplete(refreshContext)) return null; + return this.runMetadataRefresh(refreshContext, async () => { + const current = await this.refreshMetadataResourceNotification(refreshContext, "skills"); + return current ? this.appServerMetadataSnapshot(refreshContext) : null; + }); } - beginMetadataResourceRefresh(context: AppServerQueryContext, resource: MetadataResourceKind): number { - const key = this.metadataResourceRevisionKey(context, resource); - const revision = (this.metadataRevisions.get(key) ?? 0) + 1; - this.metadataRevisions.delete(key); - this.metadataRevisions.set(key, revision); - while (this.metadataRevisions.size > 16) { - for (const oldestKey of this.metadataRevisions.keys()) { - this.metadataRevisions.delete(oldestKey); - break; - } - } - return revision; - } - - metadataResourceRefreshIsCurrent(context: AppServerQueryContext, resource: MetadataResourceKind, revision: number): boolean { - return this.metadataRevisions.get(this.metadataResourceRevisionKey(context, resource)) === revision; + async refreshRateLimits(context: AppServerQueryContext): Promise { + const refreshContext = cloneAppServerQueryContext(context); + if (!appServerQueryContextIsComplete(refreshContext)) return null; + return this.runMetadataRefresh(refreshContext, async () => { + const current = await this.refreshMetadataResourceNotification(refreshContext, "rateLimits"); + return current ? this.appServerMetadataSnapshot(refreshContext) : null; + }); } modelsSnapshot(context: AppServerQueryContext): readonly ModelMetadata[] | null { @@ -315,83 +372,215 @@ export class AppServerQueryCache { return kind === "archived" ? archivedThreadsQueryKey(context) : activeThreadsQueryKey(context); } - private appServerMetadataQueryOptions( - context: AppServerQueryContext, - options: { forceSkills?: boolean } = {}, - ): AppServerQueryOptions { + private runtimeConfigQueryOptions(context: AppServerQueryContext): AppServerQueryOptions { const refreshContext = cloneAppServerQueryContext(context); return { - queryKey: appServerMetadataQueryKey(refreshContext), - queryFn: async (): Promise => { - const previous = this.appServerMetadataSnapshot(refreshContext); - const revision = this.metadataRevision(refreshContext); - const metadata = await this.runWithClient(refreshContext, async (client) => { - const runtimeConfig = runtimeConfigSnapshotFromAppServerConfig(await readEffectiveConfig(client, refreshContext.vaultPath)); - const [modelProbe, skills, permissionProfiles, rateLimit] = await Promise.all([ - this.readModelMetadataProbe(refreshContext, client), - readSkillMetadataProbe(client, refreshContext.vaultPath, options.forceSkills ?? false), - readPermissionProfileMetadataProbe(client, refreshContext.vaultPath), - readRateLimitMetadataProbe(client), - ]); - const diagnostics = [modelProbe, skills.probe, permissionProfiles.probe, rateLimit.probe].reduce( - (current, probe) => diagnosticsWithProbe(current, probe), - previous?.serverDiagnostics ?? createServerDiagnostics(), - ); - return metadataWithLastKnownGood( - { - runtimeConfig, - availableSkills: skills.value, - availablePermissionProfiles: permissionProfiles.value, - rateLimit: rateLimit.value, - serverDiagnostics: diagnostics, - }, - previous, - ); - }); - return this.metadataRevision(refreshContext) === revision ? metadata : (this.appServerMetadataSnapshot(refreshContext) ?? metadata); - }, + queryKey: appServerRuntimeConfigQueryKey(refreshContext), + queryFn: async (): Promise => + this.runWithClient(refreshContext, async (client) => + runtimeConfigSnapshotFromAppServerConfig(await readEffectiveConfig(client, refreshContext.vaultPath)), + ), staleTime: APP_SERVER_METADATA_STALE_TIME_MS, }; } + private skillsQueryOptions( + context: AppServerQueryContext, + forceReload = false, + ): AppServerQueryOptions> { + const refreshContext = cloneAppServerQueryContext(context); + return { + queryKey: appServerSkillsQueryKey(refreshContext), + queryFn: async () => + this.runWithClient(refreshContext, async (client) => + successfulMetadataResource(await readSkillMetadataProbe(client, refreshContext.vaultPath, forceReload)), + ), + staleTime: APP_SERVER_METADATA_STALE_TIME_MS, + }; + } + + private permissionProfilesQueryOptions( + context: AppServerQueryContext, + ): AppServerQueryOptions> { + const refreshContext = cloneAppServerQueryContext(context); + return { + queryKey: appServerPermissionProfilesQueryKey(refreshContext), + queryFn: async () => + this.runWithClient(refreshContext, async (client) => + successfulMetadataResource(await readPermissionProfileMetadataProbe(client, refreshContext.vaultPath)), + ), + staleTime: APP_SERVER_METADATA_STALE_TIME_MS, + }; + } + + private rateLimitsQueryOptions( + context: AppServerQueryContext, + ): AppServerQueryOptions> { + const refreshContext = cloneAppServerQueryContext(context); + return { + queryKey: appServerRateLimitsQueryKey(refreshContext), + queryFn: async () => + this.runWithClient(refreshContext, async (client) => successfulMetadataResource(await readRateLimitMetadataProbe(client))), + staleTime: APP_SERVER_METADATA_STALE_TIME_MS, + }; + } + + private async fetchRuntimeConfig(context: AppServerQueryContext, force: boolean): Promise { + const key = appServerRuntimeConfigQueryKey(context); + if (force) await this.client.invalidateQueries({ queryKey: key }); + return cloneRuntimeConfigSnapshot(await this.client.fetchQuery(this.runtimeConfigQueryOptions(context))); + } + + private fetchMetadataResource( + context: AppServerQueryContext, + resource: MetadataResourceKind, + force: boolean, + reloadSkills = false, + ): Promise { + const key = this.metadataResourceKey(context, resource); + const refresh = (async (): Promise => { + if (resource === "skills") { + const options = this.skillsQueryOptions(context, reloadSkills); + if (force) await this.client.invalidateQueries({ queryKey: options.queryKey }); + await this.client.fetchQuery(options); + return; + } + if (resource === "permissionProfiles") { + const options = this.permissionProfilesQueryOptions(context); + if (force) await this.client.invalidateQueries({ queryKey: options.queryKey }); + await this.client.fetchQuery(options); + return; + } + const options = this.rateLimitsQueryOptions(context); + if (force) await this.client.invalidateQueries({ queryKey: options.queryKey }); + await this.client.fetchQuery(options); + })(); + const active = this.metadataResourceFetches.get(key) ?? new Set>(); + active.add(refresh); + this.metadataResourceFetches.set(key, active); + const cleanup = (): void => { + active.delete(refresh); + if (active.size === 0 && this.metadataResourceFetches.get(key) === active) this.metadataResourceFetches.delete(key); + }; + void refresh.then(cleanup, cleanup); + return refresh; + } + + private refreshMetadataResourceNotification( + context: AppServerQueryContext, + resource: MetadataNotificationResourceKind, + ): Promise { + const key = this.metadataResourceKey(context, resource); + const generation = this.generation; + const current = this.metadataNotificationRefreshes.get(key); + if (current?.generation === generation) { + current.dirty = true; + return current.promise.then(() => generation === this.generation); + } + + const activeFetches = [...(this.metadataResourceFetches.get(key) ?? [])]; + const refresh: MetadataNotificationRefresh = { + dirty: false, + generation, + promise: Promise.resolve(), + }; + refresh.promise = (async (): Promise => { + if (activeFetches.length > 0) await Promise.allSettled(activeFetches); + if (generation !== this.generation) return; + for (;;) { + refresh.dirty = false; + await this.fetchMetadataResource(context, resource, true, resource === "skills").catch(() => undefined); + if (generation !== this.generation) return; + if (!metadataNotificationRefreshIsDirty(refresh)) return; + } + })(); + this.metadataNotificationRefreshes.set(key, refresh); + const cleanup = (): void => { + if (this.metadataNotificationRefreshes.get(key) === refresh) this.metadataNotificationRefreshes.delete(key); + }; + void refresh.promise.then(cleanup, cleanup); + return refresh.promise.then(() => generation === this.generation); + } + + private metadataResourceKey(context: AppServerQueryContext, resource: MetadataResourceKind): string { + return `${appServerQueryContextKey(context)}\u0000${resource}`; + } + + private metadataResourceState( + context: AppServerQueryContext, + resource: "skills", + ): { value: readonly SkillMetadata[] | null; probe: DiagnosticProbeResult }; + private metadataResourceState( + context: AppServerQueryContext, + resource: "permissionProfiles", + ): { value: readonly RuntimePermissionProfileSummary[] | null; probe: DiagnosticProbeResult }; + private metadataResourceState( + context: AppServerQueryContext, + resource: "rateLimits", + ): { value: RateLimitSnapshot | null; probe: DiagnosticProbeResult }; + private metadataResourceState( + context: AppServerQueryContext, + resource: MetadataResourceKind, + ): { value: MetadataResourceValue; probe: DiagnosticProbeResult } { + const key = + resource === "skills" + ? appServerSkillsQueryKey(context) + : resource === "permissionProfiles" + ? appServerPermissionProfilesQueryKey(context) + : appServerRateLimitsQueryKey(context); + const state = this.client.getQueryState>(key); + const failedProbe = diagnosticProbeFromError(state?.error); + return { + value: state?.data?.value ?? null, + probe: failedProbe ?? state?.data?.probe ?? createServerDiagnostics().probes[resource], + }; + } + + private modelsProbe(context: AppServerQueryContext): DiagnosticProbeResult { + const state = this.client.getQueryState(appServerModelsQueryKey(context)); + return ( + diagnosticProbeFromError(state?.error) ?? + (state?.data + ? diagnosticProbeOk("models", `${String(state.data.length)} models`, state.dataUpdatedAt) + : createServerDiagnostics().probes.models) + ); + } + + private async runMetadataRefresh(context: AppServerQueryContext, operation: () => Promise): Promise { + const key = appServerQueryContextKey(context); + this.metadataRefreshes.set(key, (this.metadataRefreshes.get(key) ?? 0) + 1); + try { + return await operation(); + } finally { + const remaining = (this.metadataRefreshes.get(key) ?? 1) - 1; + if (remaining > 0) { + this.metadataRefreshes.set(key, remaining); + } else { + this.metadataRefreshes.delete(key); + for (const listener of this.metadataProjectionListeners.get(key) ?? []) listener(); + } + } + } + private modelsQueryOptions(context: AppServerQueryContext): AppServerQueryOptions { const refreshContext = cloneAppServerQueryContext(context); return { queryKey: appServerModelsQueryKey(refreshContext), queryFn: async (): Promise => { - return cloneModelMetadata( - await this.runWithClient(refreshContext, (client) => listModelMetadata(client), { - serverRequests: { kind: "reject", message: "Codex model list refresh does not handle server requests." }, - }), - ); + try { + return cloneModelMetadata( + await this.runWithClient(refreshContext, (client) => listModelMetadata(client), { + serverRequests: { kind: "reject", message: "Codex model list refresh does not handle server requests." }, + }), + ); + } catch (error) { + throw new MetadataResourceQueryError(diagnosticProbeError("models", error, Date.now())); + } }, staleTime: MODELS_STALE_TIME_MS, }; } - private async readModelMetadataProbe( - context: AppServerQueryContext, - client: AppServerClient, - ): Promise { - try { - const models = cloneModelMetadata(await this.client.fetchQuery(this.modelsQueryOptionsWithClient(context, client))); - return diagnosticProbeOk("models", `${String(models.length)} models`, Date.now()); - } catch (error) { - return diagnosticProbeError("models", error, Date.now()); - } - } - - private modelsQueryOptionsWithClient( - context: AppServerQueryContext, - client: AppServerClient, - ): AppServerQueryOptions { - const refreshContext = cloneAppServerQueryContext(context); - return { - ...this.modelsQueryOptions(refreshContext), - queryFn: async (): Promise => cloneModelMetadata(await listModelMetadata(client)), - }; - } - private observeQueryResult( queryOptions: AppServerQueryOptions, clone: (value: T) => T, @@ -455,48 +644,26 @@ export class AppServerQueryCache { const key = this.activeThreadCursorKey(context); this.activeThreadRevisions.set(key, this.activeThreadRevision(context) + 1); } +} - private metadataRevision(context: AppServerQueryContext): number { - return this.metadataWriteRevisions.get(this.metadataWriteRevisionKey(context)) ?? 0; - } - - private bumpMetadataRevision(context: AppServerQueryContext): void { - this.beginMetadataResourceRefresh(context, "skills"); - this.beginMetadataResourceRefresh(context, "rateLimits"); - } - - private bumpMetadataWriteRevision(context: AppServerQueryContext): void { - const key = this.metadataWriteRevisionKey(context); - const revision = this.metadataRevision(context) + 1; - this.metadataWriteRevisions.delete(key); - this.metadataWriteRevisions.set(key, revision); - while (this.metadataWriteRevisions.size > 8) { - for (const oldestKey of this.metadataWriteRevisions.keys()) { - this.metadataWriteRevisions.delete(oldestKey); - break; - } - } - } - - private metadataWriteRevisionKey(context: AppServerQueryContext): string { - return JSON.stringify(appServerMetadataQueryKey(context)); - } - - private metadataResourceRevisionKey(context: AppServerQueryContext, resource: MetadataResourceKind): string { - return JSON.stringify([...appServerMetadataQueryKey(context), resource]); +class MetadataResourceQueryError extends Error { + constructor(readonly probe: DiagnosticProbeResult) { + super(probe.message ?? `Codex app-server ${probe.id} query failed.`); + this.name = "MetadataResourceQueryError"; } } -function metadataWithLastKnownGood(metadata: SharedServerMetadata, previous: SharedServerMetadata | null): SharedServerMetadata { - const probes = metadata.serverDiagnostics.probes; - return cloneSharedServerMetadata({ - ...metadata, - availableSkills: probes.skills.status === "ok" ? metadata.availableSkills : (previous?.availableSkills ?? []), - availablePermissionProfiles: - probes.permissionProfiles.status === "ok" ? metadata.availablePermissionProfiles : (previous?.availablePermissionProfiles ?? []), - rateLimit: probes.rateLimits.status === "ok" ? metadata.rateLimit : (previous?.rateLimit ?? null), - serverDiagnostics: metadataResourceDiagnostics(metadata.serverDiagnostics), - }); +function successfulMetadataResource(result: MetadataResourceSnapshot): MetadataResourceSnapshot { + if (result.probe.status !== "ok") throw new MetadataResourceQueryError(result.probe); + return result; +} + +function diagnosticProbeFromError(error: unknown): DiagnosticProbeResult | null { + return error instanceof MetadataResourceQueryError ? error.probe : null; +} + +function metadataNotificationRefreshIsDirty(refresh: MetadataNotificationRefresh): boolean { + return refresh.dirty; } function createAppServerQueryClient(): QueryClient { diff --git a/src/app-server/query/keys.ts b/src/app-server/query/keys.ts index c490edf0..a2254c89 100644 --- a/src/app-server/query/keys.ts +++ b/src/app-server/query/keys.ts @@ -6,8 +6,11 @@ export interface AppServerQueryContext { type AppServerQueryScope = readonly ["app-server", string, string]; export type AppServerActiveThreadsQueryKey = readonly [...AppServerQueryScope, "threads", "active"]; export type AppServerArchivedThreadsQueryKey = readonly [...AppServerQueryScope, "threads", "archived"]; -export type AppServerMetadataQueryKey = readonly [...AppServerQueryScope, "metadata"]; export type AppServerModelsQueryKey = readonly [...AppServerQueryScope, "models"]; +export type AppServerRuntimeConfigQueryKey = readonly [...AppServerQueryScope, "runtime-config"]; +export type AppServerSkillsQueryKey = readonly [...AppServerQueryScope, "skills"]; +export type AppServerPermissionProfilesQueryKey = readonly [...AppServerQueryScope, "permission-profiles"]; +export type AppServerRateLimitsQueryKey = readonly [...AppServerQueryScope, "rate-limits"]; export function appServerQueryContextIsComplete(context: AppServerQueryContext): boolean { return nonEmptyString(context.codexPath) && nonEmptyString(context.vaultPath); @@ -41,14 +44,26 @@ export function archivedThreadsQueryKey(context: AppServerQueryContext): AppServ return [...appServerQueryScope(context), "threads", "archived"]; } -export function appServerMetadataQueryKey(context: AppServerQueryContext): AppServerMetadataQueryKey { - return [...appServerQueryScope(context), "metadata"]; -} - export function appServerModelsQueryKey(context: AppServerQueryContext): AppServerModelsQueryKey { return [...appServerQueryScope(context), "models"]; } +export function appServerRuntimeConfigQueryKey(context: AppServerQueryContext): AppServerRuntimeConfigQueryKey { + return [...appServerQueryScope(context), "runtime-config"]; +} + +export function appServerSkillsQueryKey(context: AppServerQueryContext): AppServerSkillsQueryKey { + return [...appServerQueryScope(context), "skills"]; +} + +export function appServerPermissionProfilesQueryKey(context: AppServerQueryContext): AppServerPermissionProfilesQueryKey { + return [...appServerQueryScope(context), "permission-profiles"]; +} + +export function appServerRateLimitsQueryKey(context: AppServerQueryContext): AppServerRateLimitsQueryKey { + return [...appServerQueryScope(context), "rate-limits"]; +} + function nonEmptyString(value: string): boolean { return value.trim().length > 0; } diff --git a/src/app-server/query/shared-queries.ts b/src/app-server/query/shared-queries.ts index 07221f56..98906961 100644 --- a/src/app-server/query/shared-queries.ts +++ b/src/app-server/query/shared-queries.ts @@ -1,7 +1,7 @@ import type { ModelMetadata } from "../../domain/catalog/metadata"; import type { SharedServerMetadata } from "../../domain/server/metadata"; import type { Thread } from "../../domain/threads/model"; -import type { AppServerQueryCache, MetadataResourceKind } from "./cache"; +import type { AppServerQueryCache } from "./cache"; import { type AppServerQueryContext, appServerQueryContextKey, @@ -105,23 +105,16 @@ export class AppServerSharedQueries { return this.options.cache.appServerMetadataSnapshot(this.context()); } - updateAppServerMetadata( - updater: (metadata: SharedServerMetadata | null) => SharedServerMetadata | null, - resource?: MetadataResourceKind, - ): SharedServerMetadata | null { - return this.options.cache.updateAppServerMetadata(this.context(), updater, resource); + refreshAppServerMetadata(): Promise { + return this.runForCurrentContext((context) => this.options.cache.refreshAppServerMetadata(context)); } - beginAppServerMetadataResourceRefresh(resource: MetadataResourceKind): () => boolean { - const context = this.context(); - const revision = this.options.cache.beginMetadataResourceRefresh(context, resource); - return () => - appServerQueryContextMatches(this.context(), context) && - this.options.cache.metadataResourceRefreshIsCurrent(context, resource, revision); + refreshSkills(): Promise { + return this.runForCurrentContext((context) => this.options.cache.refreshSkills(context)); } - refreshAppServerMetadata(options: { forceSkills?: boolean } = {}): Promise { - return this.runForCurrentContext((context) => this.options.cache.refreshAppServerMetadata(context, options)); + refreshRateLimits(): Promise { + return this.runForCurrentContext((context) => this.options.cache.refreshRateLimits(context)); } observeAppServerMetadataResult(listener: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void { diff --git a/src/domain/server/diagnostics.ts b/src/domain/server/diagnostics.ts index e69ac29e..0a5f0292 100644 --- a/src/domain/server/diagnostics.ts +++ b/src/domain/server/diagnostics.ts @@ -66,10 +66,6 @@ export function diagnosticsWithToolInventory(diagnostics: Diagnostics, toolInven }; } -export function metadataResourceDiagnostics(diagnostics: Diagnostics): Diagnostics { - return diagnosticsWithMetadataResourceProbes(createServerDiagnostics(), diagnostics); -} - export function diagnosticsWithMetadataResourceProbes(diagnostics: Diagnostics, metadataDiagnostics: Diagnostics): Diagnostics { return METADATA_RESOURCE_PROBE_IDS.reduce((current, id) => diagnosticsWithProbe(current, metadataDiagnostics.probes[id]), diagnostics); } diff --git a/src/features/chat/app-server/inbound/notification-plan.ts b/src/features/chat/app-server/inbound/notification-plan.ts index da32d031..bccccf9e 100644 --- a/src/features/chat/app-server/inbound/notification-plan.ts +++ b/src/features/chat/app-server/inbound/notification-plan.ts @@ -105,10 +105,10 @@ function planDiagnosticStatus(notification: DiagnosticStatusNotification): ChatN case "account/rateLimits/updated": return effectPlan({ type: "apply-app-server-resource-event", - event: { type: "rate-limits-updated", preserveExistingOnFailure: true }, + event: { type: "rate-limits-updated" }, }); case "skills/changed": - return effectPlan({ type: "apply-app-server-resource-event", event: { type: "skills-changed", forceReload: true } }); + return effectPlan({ type: "apply-app-server-resource-event", event: { type: "skills-changed" } }); case "app/list/updated": return effectPlan({ type: "refresh-server-diagnostics" }); case "mcpServer/oauthLogin/completed": diff --git a/src/features/chat/app-server/transports/metadata-transports.ts b/src/features/chat/app-server/transports/metadata-transports.ts index 5712a528..f73de128 100644 --- a/src/features/chat/app-server/transports/metadata-transports.ts +++ b/src/features/chat/app-server/transports/metadata-transports.ts @@ -1,15 +1,11 @@ import type { AppServerClient } from "../../../../app-server/connection/client"; -import { readRateLimitMetadataProbe, readSkillMetadataProbe } from "../../../../app-server/query/metadata-probes"; +import { readRateLimitMetadataProbe } from "../../../../app-server/query/metadata-probes"; import { listModelMetadata } from "../../../../app-server/services/catalog"; import type { AppServerRequestClient } from "../../../../app-server/services/request-client"; import { readToolInventory } from "../../../../app-server/services/tool-inventory"; import type { DiagnosticProbeId, DiagnosticProbeResult } from "../../../../domain/server/diagnostics"; import { diagnosticProbeError, diagnosticProbeOk } from "../../../../domain/server/diagnostics"; -import type { - MetadataResourceTransport, - ServerDiagnosticsSnapshot, - ServerDiagnosticsTransport, -} from "../../application/connection/metadata-transport"; +import type { ServerDiagnosticsSnapshot, ServerDiagnosticsTransport } from "../../application/connection/metadata-transport"; interface CurrentChatAppServerClientHost { currentClient(): AppServerClient | null; @@ -20,7 +16,6 @@ interface ChatAppServerMetadataTransportHost extends CurrentChatAppServerClientH } export interface ChatMetadataTransports { - readonly metadataResource: MetadataResourceTransport; readonly serverDiagnostics: ServerDiagnosticsTransport; } @@ -30,19 +25,10 @@ interface DiagnosticProbeSnapshot { export function createChatMetadataTransports(host: ChatAppServerMetadataTransportHost): ChatMetadataTransports { return { - metadataResource: createChatMetadataResourceTransport(host), serverDiagnostics: createChatServerDiagnosticsTransport(host), }; } -function createChatMetadataResourceTransport(host: ChatAppServerMetadataTransportHost): MetadataResourceTransport { - return { - readSkillMetadata: (forceReload) => - withCapturedChatAppServerClient(host, (client) => readSkillMetadataProbe(client, host.vaultPath, forceReload)), - readRateLimitMetadata: () => withCapturedChatAppServerClient(host, (client) => readRateLimitMetadataProbe(client)), - }; -} - function createChatServerDiagnosticsTransport(host: ChatAppServerMetadataTransportHost): ServerDiagnosticsTransport { return { readServerDiagnostics: async (request): Promise => { @@ -76,15 +62,6 @@ function createChatServerDiagnosticsTransport(host: ChatAppServerMetadataTranspo }; } -async function withCapturedChatAppServerClient( - host: CurrentChatAppServerClientHost, - operation: (client: AppServerClient | null) => Promise, -): Promise { - const client = host.currentClient(); - const result = await operation(client); - return client !== null && host.currentClient() !== client ? null : result; -} - async function readRateLimitDiagnosticProbe(client: AppServerRequestClient): Promise { const result = await readRateLimitMetadataProbe(client); return { probe: result.probe }; diff --git a/src/features/chat/application/connection/metadata-transport.ts b/src/features/chat/application/connection/metadata-transport.ts index 428978c6..cde3c2e4 100644 --- a/src/features/chat/application/connection/metadata-transport.ts +++ b/src/features/chat/application/connection/metadata-transport.ts @@ -1,22 +1,8 @@ import type { SkillMetadata } from "../../../../domain/catalog/metadata"; -import type { RateLimitSnapshot } from "../../../../domain/runtime/metrics"; import type { DiagnosticProbeResult, Diagnostics } from "../../../../domain/server/diagnostics"; import type { McpServerStatusSummary } from "../../../../domain/server/mcp-status"; import type { ToolInventorySnapshot } from "../../../../domain/server/tool-inventory"; -interface MetadataProbeResult { - value: T; - probe: Diagnostics["probes"][K]; -} - -type SkillMetadataProbeResult = MetadataProbeResult; -export type RateLimitMetadataProbeResult = MetadataProbeResult; - -export interface MetadataResourceTransport { - readSkillMetadata(forceReload?: boolean): Promise; - readRateLimitMetadata(): Promise; -} - interface ServerDiagnosticsReadRequest { threadId: string | null; initialDiagnostics: Diagnostics; diff --git a/src/features/chat/application/connection/server-metadata-actions.ts b/src/features/chat/application/connection/server-metadata-actions.ts index 878bfcc1..7b213f9f 100644 --- a/src/features/chat/application/connection/server-metadata-actions.ts +++ b/src/features/chat/application/connection/server-metadata-actions.ts @@ -1,29 +1,23 @@ import { cloneServerDiagnostics, diagnosticsWithMetadataResourceProbes, - diagnosticsWithProbe, upsertMcpServerDiagnostic, } from "../../../../domain/server/diagnostics"; import type { McpServerStartupStatus } from "../../../../domain/server/mcp-status"; import type { SharedServerMetadata } from "../../../../domain/server/metadata"; import type { ChatStateStore } from "../state/store"; -import type { MetadataResourceTransport, RateLimitMetadataProbeResult } from "./metadata-transport"; export type AppServerResourceEvent = - | { type: "skills-changed"; forceReload: boolean } - | { type: "rate-limits-updated"; preserveExistingOnFailure?: boolean } + | { type: "skills-changed" } + | { type: "rate-limits-updated" } | { type: "mcp-startup-status-updated"; name: string; status: McpServerStartupStatus; message: string | null }; export interface ServerMetadataActionsHost { stateStore: ChatStateStore; - metadataResourceTransport: MetadataResourceTransport; - beginAppServerMetadataResourceRefresh: (resource: "skills" | "rateLimits") => () => boolean; - updateAppServerMetadata: ( - updater: (metadata: SharedServerMetadata | null) => SharedServerMetadata | null, - resource?: "skills" | "rateLimits", - ) => SharedServerMetadata | null; appServerMetadataSnapshot: () => SharedServerMetadata | null; - refreshAppServerMetadata: (options?: { forceSkills?: boolean }) => Promise; + refreshAppServerMetadata: () => Promise; + refreshSkills: () => Promise; + refreshRateLimits: () => Promise; isStaleSharedQueryError: (error: unknown) => boolean; } @@ -41,15 +35,11 @@ export function createServerMetadataActions(host: ServerMetadataActionsHost): Se refreshAppServerMetadata: () => refreshAppServerMetadata(host), applyAppServerResourceEvent: async (event) => { if (event.type === "skills-changed") { - await refreshSkillResource(host, event.forceReload, host.beginAppServerMetadataResourceRefresh("skills")); + await refreshMetadataResource(host, host.refreshSkills); return; } if (event.type === "rate-limits-updated") { - await refreshRateLimitResource( - host, - { preserveExistingOnFailure: event.preserveExistingOnFailure === true }, - host.beginAppServerMetadataResourceRefresh("rateLimits"), - ); + await refreshMetadataResource(host, host.refreshRateLimits); return; } await applyAppServerResourceEvent(host, event); @@ -104,77 +94,16 @@ function applyCurrentAppServerMetadataSnapshot(host: ServerMetadataActionsHost): if (metadata) applyAppServerMetadata(host, metadata); } -async function refreshSkillResource( +async function refreshMetadataResource( host: ServerMetadataActionsHost, - forceReload = false, - isCurrent: () => boolean = () => true, -): Promise { - const skills = await host.metadataResourceTransport.readSkillMetadata(forceReload); - if (!skills || !isCurrent()) return null; - const next = host.updateAppServerMetadata((metadata) => { - if (!metadata) return null; - return { - ...metadata, - ...(skills.probe.status === "ok" ? { availableSkills: skills.value } : {}), - serverDiagnostics: diagnosticsWithProbe(cloneServerDiagnostics(metadata.serverDiagnostics), skills.probe), - }; - }, "skills"); - if (next) { - applyAppServerMetadata(host, next); - return next; - } - const diagnostics = diagnosticsWithProbe(currentMetadataDiagnostics(host), skills.probe); - host.stateStore.dispatch( - skills.probe.status === "ok" - ? { - type: "connection/metadata-applied", - availableSkills: skills.value, - serverDiagnostics: diagnostics, - } - : { type: "connection/metadata-applied", serverDiagnostics: diagnostics }, - ); - return null; -} - -async function refreshRateLimitResource( - host: ServerMetadataActionsHost, - options: { preserveExistingOnFailure?: boolean } = {}, - isCurrent: () => boolean = () => true, + refresh: () => Promise, ): Promise { - const rateLimit = await host.metadataResourceTransport.readRateLimitMetadata(); - if (!rateLimit || !isCurrent()) return; - const preserveExistingOnFailure = options.preserveExistingOnFailure === true; - const next = updateRateLimitMetadata(host, rateLimit, { preserveRateLimitOnFailure: preserveExistingOnFailure }); - if (next) { - applyAppServerMetadata(host, next); - return; + try { + const metadata = await refresh(); + if (metadata) applyAppServerMetadata(host, metadata); + } catch (error) { + if (!host.isStaleSharedQueryError(error)) throw error; } - const diagnostics = diagnosticsWithProbe(currentMetadataDiagnostics(host), rateLimit.probe); - host.stateStore.dispatch( - preserveExistingOnFailure && rateLimit.probe.status !== "ok" - ? { type: "connection/metadata-applied", serverDiagnostics: diagnostics } - : { - type: "connection/metadata-applied", - rateLimit: rateLimit.value, - serverDiagnostics: diagnostics, - }, - ); -} - -function updateRateLimitMetadata( - host: ServerMetadataActionsHost, - rateLimit: RateLimitMetadataProbeResult, - options: { preserveRateLimitOnFailure: boolean }, -): SharedServerMetadata | null { - return host.updateAppServerMetadata((metadata) => { - if (!metadata) return null; - const diagnostics = diagnosticsWithProbe(cloneServerDiagnostics(metadata.serverDiagnostics), rateLimit.probe); - return { - ...metadata, - ...(rateLimit.probe.status === "ok" || !options.preserveRateLimitOnFailure ? { rateLimit: rateLimit.value } : {}), - serverDiagnostics: diagnostics, - }; - }, "rateLimits"); } function applyMcpStartupStatusEvent( diff --git a/src/features/chat/host/bundles/connection-bundle.ts b/src/features/chat/host/bundles/connection-bundle.ts index 4a688646..b452109b 100644 --- a/src/features/chat/host/bundles/connection-bundle.ts +++ b/src/features/chat/host/bundles/connection-bundle.ts @@ -124,12 +124,10 @@ export function createConnectionBundle( const serverRequestResponders = createServerRequestResponderRegistry(); const serverMetadata = createServerMetadataActions({ stateStore, - metadataResourceTransport: appServer.metadataResource, - beginAppServerMetadataResourceRefresh: (resource) => - environment.plugin.appServerQueries.beginAppServerMetadataResourceRefresh(resource), - updateAppServerMetadata: (updater, resource) => environment.plugin.appServerQueries.updateAppServerMetadata(updater, resource), appServerMetadataSnapshot: () => environment.plugin.appServerQueries.appServerMetadataSnapshot(), - refreshAppServerMetadata: (options) => environment.plugin.appServerQueries.refreshAppServerMetadata(options), + refreshAppServerMetadata: () => environment.plugin.appServerQueries.refreshAppServerMetadata(), + refreshSkills: () => environment.plugin.appServerQueries.refreshSkills(), + refreshRateLimits: () => environment.plugin.appServerQueries.refreshRateLimits(), isStaleSharedQueryError: isStaleAppServerSharedQueryContextError, }); const serverDiagnostics = createServerDiagnosticsActions({ diff --git a/src/features/chat/host/contracts.ts b/src/features/chat/host/contracts.ts index 70fb9644..53b1a2f8 100644 --- a/src/features/chat/host/contracts.ts +++ b/src/features/chat/host/contracts.ts @@ -54,13 +54,10 @@ interface WorkspacePanels { type ChatThreadCatalog = ThreadCatalogActiveReader & ThreadCatalogEventSink & ThreadCatalogConnectionEventSink; interface ChatAppServerQueries { - beginAppServerMetadataResourceRefresh(resource: "skills" | "rateLimits"): () => boolean; - updateAppServerMetadata( - updater: (metadata: SharedServerMetadata | null) => SharedServerMetadata | null, - resource?: "skills" | "rateLimits", - ): SharedServerMetadata | null; appServerMetadataSnapshot(): SharedServerMetadata | null; - refreshAppServerMetadata(options?: { forceSkills?: boolean }): Promise; + refreshAppServerMetadata(): Promise; + refreshSkills(): Promise; + refreshRateLimits(): Promise; observeAppServerMetadataResult(listener: ObservedResultListener, options?: { emitCurrent?: boolean }): () => void; modelsSnapshot(): readonly ModelMetadata[] | null; fetchModels(): Promise; diff --git a/tests/app-server/query-cache.test.ts b/tests/app-server/query-cache.test.ts index 5739c86b..dfd380f4 100644 --- a/tests/app-server/query-cache.test.ts +++ b/tests/app-server/query-cache.test.ts @@ -2,19 +2,8 @@ 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 { SkillMetadata } from "../../src/domain/catalog/metadata"; -import { emptyRuntimeConfigSnapshot, type RuntimeConfigSnapshot } from "../../src/domain/runtime/config"; import type { RateLimitSnapshot } from "../../src/domain/runtime/metrics"; import type { RuntimePermissionProfileSummary } from "../../src/domain/runtime/permissions"; -import { - createServerDiagnostics, - diagnosticProbeError, - diagnosticProbeOk, - diagnosticsWithProbe, - diagnosticsWithToolInventory, - upsertMcpServerDiagnostic, -} from "../../src/domain/server/diagnostics"; -import type { SharedServerMetadata } from "../../src/domain/server/metadata"; describe("AppServerQueryCache", () => { it("garbage-collects inactive query contexts", async () => { @@ -32,181 +21,16 @@ describe("AppServerQueryCache", () => { } }); - it("preserves successful metadata resource values when probes fail", () => { - const cache = new AppServerQueryCache(); - const context = cacheContext(); - const goodMetadata = metadata({ - availableSkills: [skillMetadata("writer")], - availablePermissionProfiles: [permissionProfile(":workspace")], - rateLimit: rateLimit(42), - }); - - cache.writeAppServerMetadata(context, goodMetadata); - cache.writeAppServerMetadata( - context, - metadata({ - availableSkills: [skillMetadata("failed-skill")], - availablePermissionProfiles: [permissionProfile(":failed")], - rateLimit: rateLimit(90), - skillsProbeStatus: "failed", - permissionProfilesProbeStatus: "failed", - rateLimitProbeStatus: "failed", - }), - ); - - expect(cache.appServerMetadataSnapshot(context)?.availableSkills.map((skill) => skill.name)).toEqual(["writer"]); - expect(cache.appServerMetadataSnapshot(context)?.availablePermissionProfiles.map((profile) => profile.id)).toEqual([":workspace"]); - expect(cache.appServerMetadataSnapshot(context)?.rateLimit?.primary?.usedPercent).toBe(42); - expect(cache.appServerMetadataSnapshot(context)?.serverDiagnostics.probes.skills.status).toBe("failed"); - expect(cache.appServerMetadataSnapshot(context)?.serverDiagnostics.probes.rateLimits.status).toBe("failed"); - - cache.writeAppServerMetadata(context, metadata({ modelProbeStatus: "failed" })); - expect(cache.appServerMetadataSnapshot(context)?.serverDiagnostics.probes.models.status).toBe("failed"); - }); - - it("model-checks last-known-good metadata fallback for all probe status combinations", () => { - for (const statuses of metadataProbeStatusCombinations()) { - const cache = new AppServerQueryCache(); - const context = cacheContext(); - cache.writeAppServerMetadata( - context, - metadata({ - availableSkills: [skillMetadata("previous-skill")], - availablePermissionProfiles: [permissionProfile(":previous")], - rateLimit: rateLimit(11), - }), - ); - - cache.writeAppServerMetadata( - context, - metadata({ - availableSkills: [skillMetadata("next-skill")], - availablePermissionProfiles: [permissionProfile(":next")], - rateLimit: rateLimit(22), - modelProbeStatus: statuses.models, - skillsProbeStatus: statuses.skills, - permissionProfilesProbeStatus: statuses.permissionProfiles, - rateLimitProbeStatus: statuses.rateLimits, - }), - ); - - const snapshot = cache.appServerMetadataSnapshot(context); - const statusCase = metadataProbeStatusCase(statuses); - expect( - snapshot?.availableSkills.map((skill) => skill.name), - statusCase, - ).toEqual([statuses.skills === "ok" ? "next-skill" : "previous-skill"]); - expect( - snapshot?.availablePermissionProfiles.map((profile) => profile.id), - statusCase, - ).toEqual([statuses.permissionProfiles === "ok" ? ":next" : ":previous"]); - expect(snapshot?.rateLimit?.primary?.usedPercent, statusCase).toBe(statuses.rateLimits === "ok" ? 22 : 11); - } - }); - - it("does not accept failed metadata resource payloads as cache truth", () => { - const cache = new AppServerQueryCache(); - const context = cacheContext(); - - cache.writeAppServerMetadata( - context, - metadata({ - availableSkills: [skillMetadata("writer")], - availablePermissionProfiles: [permissionProfile(":workspace")], - rateLimit: rateLimit(90), - modelProbeStatus: "failed", - skillsProbeStatus: "failed", - permissionProfilesProbeStatus: "failed", - rateLimitProbeStatus: "failed", - }), - ); - - const cached = cache.appServerMetadataSnapshot(context); - expect(cached?.runtimeConfig).not.toBeNull(); - expect(cached?.serverDiagnostics.probes.models.status).toBe("failed"); - expect(cached?.availableSkills).toEqual([]); - expect(cached?.availablePermissionProfiles).toEqual([]); - expect(cached?.rateLimit).toBeNull(); - expect(cache.modelsSnapshot(context)).toBeNull(); - }); - - it("keeps connection and thread diagnostics out of shared metadata snapshots", () => { - const cache = new AppServerQueryCache(); - const context = cacheContext(); - const polluted = metadata(); - polluted.serverDiagnostics = diagnosticsWithToolInventory( - upsertMcpServerDiagnostic(polluted.serverDiagnostics, { - name: "github", - startupStatus: "ready", - authStatus: null, - toolCount: 1, - message: null, - }), - { - checkedAt: 1, - plugins: [], - pluginMarketplaceErrors: [], - pluginsError: null, - mcpServers: [], - mcpDiagnostics: [], - mcpError: null, - skills: [], - skillsError: null, - }, - ); - - cache.writeAppServerMetadata(context, polluted); - - expect(cache.appServerMetadataSnapshot(context)?.serverDiagnostics).toMatchObject({ - mcpServers: [], - toolInventory: null, - probes: { - models: { status: "ok" }, - skills: { status: "ok" }, - permissionProfiles: { status: "ok" }, - rateLimits: { status: "ok" }, - plugins: { status: "unknown" }, - mcpServers: { status: "unknown" }, - }, - }); - }); - it("does not share or store snapshots before the cache context is complete", async () => { const cache = new AppServerQueryCache(); const context = cacheContext({ codexPath: "" }); await expect(cache.fetchActiveThreads(context)).resolves.toEqual([]); - cache.writeAppServerMetadata(context, metadata()); - expect(cache.activeThreadsSnapshot(context)).toBeNull(); expect(cache.appServerMetadataSnapshot(context)).toBeNull(); expect(cache.modelsSnapshot(context)).toBeNull(); }); - it("does not let metadata writes overwrite the model query", async () => { - const cache = cacheWithRequestHandlers({ - "model/list": vi.fn().mockResolvedValue({ data: [catalogModel("gpt-model-query")] }), - }); - const context = cacheContext(); - - await cache.fetchModels(context); - cache.writeAppServerMetadata(context, metadata({ modelProbeStatus: "failed" })); - - expect(cache.modelsSnapshot(context)?.map((model) => model.model)).toEqual(["gpt-model-query"]); - }); - - it("does not reuse metadata or model snapshots across app-server cache contexts", () => { - const cache = new AppServerQueryCache(); - const context = cacheContext(); - - cache.writeAppServerMetadata(context, metadata()); - - expect(cache.appServerMetadataSnapshot(cacheContext({ vaultPath: "/other-vault" }))).toBeNull(); - expect(cache.appServerMetadataSnapshot(cacheContext({ codexPath: "/opt/codex" }))).toBeNull(); - expect(cache.modelsSnapshot(cacheContext({ vaultPath: "/other-vault" }))).toBeNull(); - expect(cache.modelsSnapshot(cacheContext({ codexPath: "/opt/codex" }))).toBeNull(); - }); - it("stores successful empty thread list snapshots as shared cache truth", async () => { const fetchThreads = vi.fn().mockResolvedValue([]); const cache = cacheWithThreads(fetchThreads); @@ -355,6 +179,259 @@ describe("AppServerQueryCache", () => { expect(cache.modelsSnapshot(context)?.map((model) => model.model)).toEqual(["gpt-meta"]); }); + it("publishes one derived metadata projection after a coordinated refresh", async () => { + const context = cacheContext(); + const skills = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); + const cache = cacheWithRequestHandlers({ + "config/read": vi.fn().mockResolvedValue({}), + "model/list": vi.fn().mockResolvedValue({ data: [] }), + "skills/list": vi.fn(() => skills.promise), + "permissionProfile/list": vi.fn().mockResolvedValue({ data: [], nextCursor: null }), + "account/rateLimits/read": vi.fn().mockResolvedValue({ rateLimits: appServerRateLimit(0), rateLimitsByLimitId: null }), + }); + const listener = vi.fn(); + const unsubscribe = cache.observeAppServerMetadataResult(context, listener, { emitCurrent: false }); + + const refresh = cache.refreshAppServerMetadata(context); + await flushMicrotasks(); + expect(listener).not.toHaveBeenCalled(); + + skills.resolve({ data: [{ skills: [catalogSkill("writer")] }] }); + await refresh; + await flushMicrotasks(); + + expect(listener).toHaveBeenCalledOnce(); + expect(listener).toHaveBeenCalledWith( + expect.objectContaining({ value: expect.objectContaining({ availableSkills: [expect.objectContaining({ name: "writer" })] }) }), + ); + unsubscribe(); + }); + + it("coalesces a burst of skills notifications into one forced trailing refresh", async () => { + const first = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); + const second = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); + const listSkills = vi + .fn() + .mockImplementationOnce(() => first.promise) + .mockImplementationOnce(() => second.promise); + const cache = cacheWithRequestHandlers({ "skills/list": listSkills }); + const context = cacheContext(); + + const refreshes = Array.from({ length: 10 }, () => cache.refreshSkills(context)); + await flushMicrotasks(); + expect(listSkills).toHaveBeenCalledOnce(); + expect(listSkills).toHaveBeenNthCalledWith(1, { cwds: ["/vault"], forceReload: true }); + + first.resolve({ data: [{ skills: [catalogSkill("old")] }] }); + await vi.waitFor(() => expect(listSkills).toHaveBeenCalledTimes(2)); + expect(listSkills).toHaveBeenNthCalledWith(2, { cwds: ["/vault"], forceReload: true }); + second.resolve({ data: [{ skills: [catalogSkill("new")] }] }); + await Promise.all(refreshes); + expect(listSkills).toHaveBeenCalledTimes(2); + }); + + it("coalesces a burst of rate-limit notifications into one trailing refresh", async () => { + const first = deferred<{ rateLimits: ReturnType; rateLimitsByLimitId: null }>(); + const second = deferred<{ rateLimits: ReturnType; rateLimitsByLimitId: null }>(); + const readRateLimits = vi + .fn() + .mockImplementationOnce(() => first.promise) + .mockImplementationOnce(() => second.promise); + const cache = cacheWithRequestHandlers({ "account/rateLimits/read": readRateLimits }); + const context = cacheContext(); + + const refreshes = Array.from({ length: 10 }, () => cache.refreshRateLimits(context)); + await flushMicrotasks(); + expect(readRateLimits).toHaveBeenCalledOnce(); + + first.resolve({ rateLimits: appServerRateLimit(10), rateLimitsByLimitId: null }); + await vi.waitFor(() => expect(readRateLimits).toHaveBeenCalledTimes(2)); + second.resolve({ rateLimits: appServerRateLimit(20), rateLimitsByLimitId: null }); + await Promise.all(refreshes); + }); + + it("deduplicates metadata resource RPCs across concurrent full refreshes", async () => { + const config = deferred>(); + const models = deferred<{ data: CatalogModel[] }>(); + const skills = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); + const profiles = deferred<{ data: RuntimePermissionProfileSummary[]; nextCursor: null }>(); + const limits = deferred<{ rateLimits: ReturnType; rateLimitsByLimitId: null }>(); + const handlers = { + "config/read": vi.fn(() => config.promise), + "model/list": vi.fn(() => models.promise), + "skills/list": vi.fn(() => skills.promise), + "permissionProfile/list": vi.fn(() => profiles.promise), + "account/rateLimits/read": vi.fn(() => limits.promise), + }; + const cache = cacheWithRequestHandlers(handlers); + const context = cacheContext(); + + const first = cache.refreshAppServerMetadata(context); + const second = cache.refreshAppServerMetadata(context); + await flushMicrotasks(); + + for (const handler of Object.values(handlers)) expect(handler).toHaveBeenCalledOnce(); + config.resolve({}); + models.resolve({ data: [] }); + skills.resolve({ data: [{ skills: [] }] }); + profiles.resolve({ data: [], nextCursor: null }); + limits.resolve({ rateLimits: appServerRateLimit(0), rateLimitsByLimitId: null }); + await Promise.all([first, second]); + }); + + it("runs a forced skills request after a notification overlaps a full refresh", async () => { + const first = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); + const second = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); + const listSkills = vi + .fn() + .mockImplementationOnce(() => first.promise) + .mockImplementationOnce(() => second.promise); + const cache = cacheWithRequestHandlers({ + "config/read": vi.fn().mockResolvedValue({}), + "model/list": vi.fn().mockResolvedValue({ data: [] }), + "skills/list": listSkills, + "permissionProfile/list": vi.fn().mockResolvedValue({ data: [], nextCursor: null }), + "account/rateLimits/read": vi.fn().mockResolvedValue({ rateLimits: appServerRateLimit(0), rateLimitsByLimitId: null }), + }); + const context = cacheContext(); + + const fullRefresh = cache.refreshAppServerMetadata(context); + await flushMicrotasks(); + const notificationRefresh = cache.refreshSkills(context); + first.resolve({ data: [{ skills: [catalogSkill("old")] }] }); + await vi.waitFor(() => expect(listSkills).toHaveBeenCalledTimes(2)); + expect(listSkills).toHaveBeenNthCalledWith(2, { cwds: ["/vault"], forceReload: true }); + second.resolve({ data: [{ skills: [catalogSkill("new")] }] }); + await Promise.all([fullRefresh, notificationRefresh]); + expect(cache.appServerMetadataSnapshot(context)?.availableSkills.map((skill) => skill.name)).toEqual(["new"]); + }); + + it("schedules one more skills read when a notification arrives during the trailing refresh", async () => { + const first = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); + const second = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); + const third = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); + const listSkills = vi + .fn() + .mockImplementationOnce(() => first.promise) + .mockImplementationOnce(() => second.promise) + .mockImplementationOnce(() => third.promise); + const cache = cacheWithRequestHandlers({ "skills/list": listSkills }); + const context = cacheContext(); + + const leading = cache.refreshSkills(context); + const trailing = cache.refreshSkills(context); + first.resolve({ data: [{ skills: [catalogSkill("first")] }] }); + await vi.waitFor(() => expect(listSkills).toHaveBeenCalledTimes(2)); + const afterTrailing = cache.refreshSkills(context); + second.resolve({ data: [{ skills: [catalogSkill("second")] }] }); + await vi.waitFor(() => expect(listSkills).toHaveBeenCalledTimes(3)); + third.resolve({ data: [{ skills: [catalogSkill("third")] }] }); + + await Promise.all([leading, trailing, afterTrailing]); + expect(listSkills).toHaveBeenCalledTimes(3); + }); + + it("does not start trailing notification work after the cache is cleared", async () => { + const first = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); + const listSkills = vi.fn(() => first.promise); + const cache = cacheWithRequestHandlers({ "skills/list": listSkills }); + const context = cacheContext(); + + const leading = cache.refreshSkills(context); + const trailing = cache.refreshSkills(context); + await flushMicrotasks(); + cache.clear(); + first.resolve({ data: [{ skills: [catalogSkill("stale")] }] }); + await Promise.all([leading, trailing]); + await flushMicrotasks(); + + expect(listSkills).toHaveBeenCalledOnce(); + expect(cache.appServerMetadataSnapshot(context)).toBeNull(); + }); + + it("rejects an initial metadata refresh when runtime config fails after optional resources settle", async () => { + const skills = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); + const cache = cacheWithRequestHandlers({ + "config/read": vi.fn().mockRejectedValue(new Error("config offline")), + "model/list": vi.fn().mockResolvedValue({ data: [] }), + "skills/list": vi.fn(() => skills.promise), + "permissionProfile/list": vi.fn().mockResolvedValue({ data: [], nextCursor: null }), + "account/rateLimits/read": vi.fn().mockResolvedValue({ rateLimits: appServerRateLimit(0), rateLimitsByLimitId: null }), + }); + const context = cacheContext(); + let settled = false; + const refresh = cache.refreshAppServerMetadata(context).finally(() => { + settled = true; + }); + await flushMicrotasks(); + expect(settled).toBe(false); + + skills.resolve({ data: [{ skills: [] }] }); + await expect(refresh).rejects.toThrow("config offline"); + expect(cache.appServerMetadataSnapshot(context)).toBeNull(); + }); + + it("rejects runtime config refresh failures while preserving prior config and refreshed optional resources", async () => { + const readConfig = vi.fn().mockResolvedValueOnce({}).mockRejectedValueOnce(new Error("config offline")); + const listSkills = vi + .fn() + .mockResolvedValueOnce({ data: [{ skills: [catalogSkill("old")] }] }) + .mockResolvedValueOnce({ data: [{ skills: [catalogSkill("new")] }] }); + const cache = cacheWithRequestHandlers({ + "config/read": readConfig, + "model/list": vi.fn().mockResolvedValue({ data: [] }), + "skills/list": listSkills, + "permissionProfile/list": vi.fn().mockResolvedValue({ data: [], nextCursor: null }), + "account/rateLimits/read": vi.fn().mockResolvedValue({ rateLimits: appServerRateLimit(0), rateLimitsByLimitId: null }), + }); + const context = cacheContext(); + await cache.refreshAppServerMetadata(context); + + await expect(cache.refreshAppServerMetadata(context)).rejects.toThrow("config offline"); + + const snapshot = cache.appServerMetadataSnapshot(context); + expect(snapshot?.runtimeConfig).not.toBeNull(); + expect(snapshot?.availableSkills.map((skill) => skill.name)).toEqual(["new"]); + }); + + it("does not emit a queued metadata projection after unsubscribe", async () => { + vi.useFakeTimers({ toFake: ["queueMicrotask"] }); + try { + const cache = metadataCacheWithSuccessfulHandlers(); + const context = cacheContext(); + await cache.refreshAppServerMetadata(context); + const listener = vi.fn(); + const unsubscribe = cache.observeAppServerMetadataResult(context, listener, { emitCurrent: false }); + + await cache.refreshSkills(context); + unsubscribe(); + vi.runAllTicks(); + + expect(listener).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it("does not emit a queued metadata projection after the cache is cleared", async () => { + vi.useFakeTimers({ toFake: ["queueMicrotask"] }); + try { + const cache = metadataCacheWithSuccessfulHandlers(); + const context = cacheContext(); + await cache.refreshAppServerMetadata(context); + const listener = vi.fn(); + cache.observeAppServerMetadataResult(context, listener, { emitCurrent: false }); + + await cache.refreshSkills(context); + cache.clear(); + vi.runAllTicks(); + + expect(listener).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + it("shares in-flight model fetches between metadata and models queries", async () => { const context = cacheContext(); const modelRefresh = deferred<{ data: CatalogModel[] }>(); @@ -409,22 +486,26 @@ describe("AppServerQueryCache", () => { .fn() .mockResolvedValueOnce({ data: [catalogModel("gpt-cached")] }) .mockRejectedValueOnce(new Error("models offline")); + const listSkills = vi + .fn() + .mockResolvedValueOnce({ data: [{ skills: [catalogSkill("cached-skill")] }] }) + .mockRejectedValueOnce(new Error("skills offline")); + const listProfiles = vi + .fn() + .mockResolvedValueOnce({ data: [permissionProfile(":cached")], nextCursor: null }) + .mockRejectedValueOnce(new Error("profiles offline")); + const readRateLimits = vi + .fn() + .mockResolvedValueOnce({ rateLimits: appServerRateLimit(17), rateLimitsByLimitId: null }) + .mockRejectedValueOnce(new Error("limits offline")); const cache = cacheWithRequestHandlers({ "config/read": vi.fn().mockResolvedValue({}), "model/list": listModels, - "skills/list": vi.fn().mockRejectedValue(new Error("skills offline")), - "permissionProfile/list": vi.fn().mockRejectedValue(new Error("profiles offline")), - "account/rateLimits/read": vi.fn().mockRejectedValue(new Error("limits offline")), + "skills/list": listSkills, + "permissionProfile/list": listProfiles, + "account/rateLimits/read": readRateLimits, }); - cache.writeAppServerMetadata( - context, - metadata({ - availableSkills: [skillMetadata("cached-skill")], - availablePermissionProfiles: [permissionProfile(":cached")], - rateLimit: rateLimit(17), - }), - ); - await cache.fetchModels(context); + await cache.refreshAppServerMetadata(context); const refreshed = await cache.refreshAppServerMetadata(context); @@ -441,32 +522,6 @@ describe("AppServerQueryCache", () => { expect(cache.appServerMetadataSnapshot(context)).toEqual(refreshed); }); - it("does not overwrite a newer sparse metadata write with an in-flight full refresh", async () => { - const skills = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>(); - const cache = cacheWithRequestHandlers({ - "config/read": vi.fn().mockResolvedValue({}), - "model/list": vi.fn().mockResolvedValue({ data: [] }), - "skills/list": vi.fn(() => skills.promise), - "permissionProfile/list": vi.fn().mockResolvedValue({ data: [], nextCursor: null }), - "account/rateLimits/read": vi.fn().mockResolvedValue({ rateLimits: appServerRateLimit(0), rateLimitsByLimitId: null }), - }); - const context = cacheContext(); - cache.writeAppServerMetadata(context, metadata({ availableSkills: [skillMetadata("initial")] })); - for (let index = 0; index < 5; index += 1) cache.beginMetadataResourceRefresh(context, "skills"); - const refresh = cache.refreshAppServerMetadata(context); - await flushMicrotasks(); - - cache.updateAppServerMetadata( - context, - () => metadata({ availableSkills: [skillMetadata("event")], rateLimit: rateLimit(42) }), - "rateLimits", - ); - skills.resolve({ data: [{ skills: [catalogSkill("old-full")] }] }); - - await expect(refresh).resolves.toMatchObject({ availableSkills: [{ name: "event" }] }); - expect(cache.appServerMetadataSnapshot(context)?.availableSkills.map((skill) => skill.name)).toEqual(["event"]); - }); - it("stores an in-flight app-server snapshot as raw thread-list truth", async () => { const context = cacheContext(); const refresh = deferred[]>(); @@ -525,93 +580,20 @@ function cacheWithRequestHandlers(handlers: Record }); } -function metadata( - overrides: { - availableSkills?: readonly SkillMetadata[]; - availablePermissionProfiles?: readonly RuntimePermissionProfileSummary[]; - rateLimit?: RateLimitSnapshot | null; - runtimeConfig?: RuntimeConfigSnapshot | null; - modelProbeStatus?: "ok" | "failed"; - skillsProbeStatus?: "ok" | "failed"; - permissionProfilesProbeStatus?: "ok" | "failed"; - rateLimitProbeStatus?: "ok" | "failed"; - } = {}, -): SharedServerMetadata { - let diagnostics = createServerDiagnostics(); - diagnostics = diagnosticsWithProbe( - diagnostics, - overrides.modelProbeStatus === "failed" - ? diagnosticProbeError("models", new Error("offline"), 1) - : diagnosticProbeOk("models", "1 models", 1), - ); - diagnostics = diagnosticsWithProbe( - diagnostics, - overrides.skillsProbeStatus === "failed" - ? diagnosticProbeError("skills", new Error("offline"), 1) - : diagnosticProbeOk("skills", "0 skills", 1), - ); - diagnostics = diagnosticsWithProbe( - diagnostics, - overrides.rateLimitProbeStatus === "failed" - ? diagnosticProbeError("rateLimits", new Error("offline"), 1) - : diagnosticProbeOk("rateLimits", "available", 1), - ); - diagnostics = diagnosticsWithProbe( - diagnostics, - overrides.permissionProfilesProbeStatus === "failed" - ? diagnosticProbeError("permissionProfiles", new Error("offline"), 1) - : diagnosticProbeOk("permissionProfiles", "0 profiles", 1), - ); - return { - runtimeConfig: overrides.runtimeConfig ?? emptyRuntimeConfigSnapshot(), - availableSkills: overrides.availableSkills ?? [], - availablePermissionProfiles: overrides.availablePermissionProfiles ?? [], - rateLimit: overrides.rateLimit ?? null, - serverDiagnostics: diagnostics, - }; -} - -type MetadataProbeStatus = "ok" | "failed"; - -interface MetadataProbeStatuses { - readonly models: MetadataProbeStatus; - readonly skills: MetadataProbeStatus; - readonly permissionProfiles: MetadataProbeStatus; - readonly rateLimits: MetadataProbeStatus; -} - -function metadataProbeStatusCombinations(): MetadataProbeStatuses[] { - const statuses: readonly MetadataProbeStatus[] = ["ok", "failed"]; - return statuses.flatMap((models) => - statuses.flatMap((skills) => - statuses.flatMap((permissionProfiles) => statuses.map((rateLimits) => ({ models, skills, permissionProfiles, rateLimits }))), - ), - ); -} - -function metadataProbeStatusCase(statuses: MetadataProbeStatuses): string { - return `models=${statuses.models} skills=${statuses.skills} permissionProfiles=${statuses.permissionProfiles} rateLimits=${statuses.rateLimits}`; -} - -function skillMetadata(name: string): SkillMetadata { - return { name, description: "", path: `/tmp/${name}`, enabled: true }; +function metadataCacheWithSuccessfulHandlers(): AppServerQueryCache { + return cacheWithRequestHandlers({ + "config/read": vi.fn().mockResolvedValue({}), + "model/list": vi.fn().mockResolvedValue({ data: [] }), + "skills/list": vi.fn().mockResolvedValue({ data: [{ skills: [] }] }), + "permissionProfile/list": vi.fn().mockResolvedValue({ data: [], nextCursor: null }), + "account/rateLimits/read": vi.fn().mockResolvedValue({ rateLimits: appServerRateLimit(0), rateLimitsByLimitId: null }), + }); } function permissionProfile(id: string): RuntimePermissionProfileSummary { return { id, description: null, allowed: true }; } -function rateLimit(usedPercent: number): RateLimitSnapshot { - return { - limitId: "codex", - limitName: "Codex", - primary: { usedPercent, windowDurationMins: 300, resetsAt: 1 }, - secondary: null, - individualLimit: null, - rateLimitReachedType: null, - }; -} - function catalogModel(model: string): CatalogModel { return { id: model, diff --git a/tests/app-server/shared-queries.test.ts b/tests/app-server/shared-queries.test.ts index 997b16f5..76812544 100644 --- a/tests/app-server/shared-queries.test.ts +++ b/tests/app-server/shared-queries.test.ts @@ -113,7 +113,6 @@ describe("AppServerSharedQueries", () => { const queries = new AppServerSharedQueries({ cache: cacheWith({ appServerMetadataSnapshot: () => metadata, - updateAppServerMetadata: () => metadata, modelsSnapshot: () => models, observeAppServerMetadataResult: (_context, listener) => { metadataObserver = listener; @@ -131,7 +130,6 @@ describe("AppServerSharedQueries", () => { queries.observeAppServerMetadataResult(metadataListener); queries.observeModelsResult(modelListener); - queries.updateAppServerMetadata(() => metadata); metadataObserver(observedResult(metadata)); modelObserver(observedResult(models)); @@ -149,8 +147,9 @@ function cacheWith(overrides: Partial): AppServerQueryCache refreshActiveThreads: vi.fn(() => Promise.resolve([])), observeActiveThreadsResult: vi.fn(() => () => undefined), appServerMetadataSnapshot: vi.fn(() => null), - updateAppServerMetadata: 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([])), diff --git a/tests/features/chat/app-server/inbound/handler.test.ts b/tests/features/chat/app-server/inbound/handler.test.ts index 1c5f33e1..25b08eed 100644 --- a/tests/features/chat/app-server/inbound/handler.test.ts +++ b/tests/features/chat/app-server/inbound/handler.test.ts @@ -208,7 +208,7 @@ describe("ChatInboundHandler", () => { params: {}, } satisfies Extract); - expect(applyAppServerResourceEvent).toHaveBeenCalledWith({ type: "skills-changed", forceReload: true }); + expect(applyAppServerResourceEvent).toHaveBeenCalledWith({ type: "skills-changed" }); }); it("stores the latest aggregated turn diff for the active turn", () => { @@ -701,10 +701,7 @@ describe("ChatInboundHandler", () => { } satisfies Extract); expect(state.connection.rateLimit).toBeNull(); - expect(applyAppServerResourceEvent).toHaveBeenCalledWith({ - type: "rate-limits-updated", - preserveExistingOnFailure: true, - }); + expect(applyAppServerResourceEvent).toHaveBeenCalledWith({ type: "rate-limits-updated" }); }); it("routes MCP startup status through the app-server resource event boundary", () => { diff --git a/tests/features/chat/app-server/transports.test.ts b/tests/features/chat/app-server/transports.test.ts index 5c8751a6..19421a77 100644 --- a/tests/features/chat/app-server/transports.test.ts +++ b/tests/features/chat/app-server/transports.test.ts @@ -343,21 +343,6 @@ describe("chat app-server transports", () => { expect(request).toHaveBeenCalledWith("thread/settings/update", { threadId: "thread", model: "gpt-5.5" }); }); - it("reads sparse skill metadata through the current app-server client", async () => { - const request = vi.fn().mockResolvedValue({ data: [{ cwd: "/vault", skills: [] }] }); - const client = { request } as unknown as AppServerClient; - const transport = createTestGateway({ - currentClient: () => client, - }).metadataResource; - - await expect(transport.readSkillMetadata(true)).resolves.toMatchObject({ - value: [], - probe: { status: "ok" }, - }); - - expect(request).toHaveBeenCalledWith("skills/list", { cwds: ["/vault"], forceReload: true }); - }); - it("reads diagnostics probes and tool inventory at the app-server boundary", async () => { const request = vi.fn((method: string) => { switch (method) { diff --git a/tests/features/chat/application/connection/server-actions.test.ts b/tests/features/chat/application/connection/server-actions.test.ts index 353b60f9..a4a44725 100644 --- a/tests/features/chat/application/connection/server-actions.test.ts +++ b/tests/features/chat/application/connection/server-actions.test.ts @@ -6,7 +6,6 @@ import type { RateLimitSnapshot } from "../../../../../src/domain/runtime/metric import { createServerDiagnostics, type DiagnosticProbeResult, - diagnosticProbeError, diagnosticProbeOk, diagnosticsWithProbe, diagnosticsWithToolInventory, @@ -15,7 +14,6 @@ import { import type { McpServerStatusSummary } from "../../../../../src/domain/server/mcp-status"; import type { SharedServerMetadata } from "../../../../../src/domain/server/metadata"; import type { ToolInventorySnapshot } from "../../../../../src/domain/server/tool-inventory"; -import type { MetadataResourceTransport } from "../../../../../src/features/chat/application/connection/metadata-transport"; import { createServerDiagnosticsActions } from "../../../../../src/features/chat/application/connection/server-diagnostics-actions"; import { createServerMetadataActions } from "../../../../../src/features/chat/application/connection/server-metadata-actions"; import { createChatStateStore } from "../../../../../src/features/chat/application/state/store"; @@ -29,7 +27,6 @@ describe("server metadata actions", () => { const metadata = serverMetadataFixture({ availableSkills: [skillFixture("writer")] }); const actions = createServerMetadataActions({ stateStore, - metadataResourceTransport: metadataResourceTransport(), ...metadataCacheHost(), refreshAppServerMetadata: vi.fn().mockResolvedValue(metadata), isStaleSharedQueryError: () => false, @@ -57,7 +54,6 @@ describe("server metadata actions", () => { }); const actions = createServerMetadataActions({ stateStore, - metadataResourceTransport: metadataResourceTransport(), ...metadataCacheHost(), refreshAppServerMetadata: vi.fn().mockResolvedValue(metadata), isStaleSharedQueryError: () => false, @@ -77,7 +73,6 @@ describe("server metadata actions", () => { const cache = { current: serverMetadataFixture() as SharedServerMetadata | null }; const actions = createServerMetadataActions({ stateStore, - metadataResourceTransport: metadataResourceTransport(), ...metadataCacheHost(cache), refreshAppServerMetadata: vi.fn().mockResolvedValue(null), isStaleSharedQueryError: () => false, @@ -99,7 +94,6 @@ describe("server metadata actions", () => { const stale = new Error("stale"); const actions = createServerMetadataActions({ stateStore, - metadataResourceTransport: metadataResourceTransport(), ...metadataCacheHost(), refreshAppServerMetadata: vi.fn().mockRejectedValue(stale), isStaleSharedQueryError: (error) => error === stale, @@ -111,149 +105,55 @@ describe("server metadata actions", () => { expect(stateStore.getState().connection.runtimeConfig).toBeNull(); }); - it("does not apply stale sparse skill refreshes", async () => { + it("does not apply stale skill refreshes", async () => { const stateStore = createChatStateStore(chatStateFixture()); - const updateAppServerMetadata = vi.fn(() => null); + const stale = new Error("stale"); const actions = createServerMetadataActions({ stateStore, - metadataResourceTransport: metadataResourceTransport({ readSkillMetadata: vi.fn().mockResolvedValue(null) }), - beginAppServerMetadataResourceRefresh: () => () => true, - appServerMetadataSnapshot: () => null, - updateAppServerMetadata, - refreshAppServerMetadata: vi.fn().mockResolvedValue(null), - isStaleSharedQueryError: () => false, + ...metadataCacheHost(), + refreshSkills: vi.fn().mockRejectedValue(stale), + isStaleSharedQueryError: (error) => error === stale, }); - await actions.applyAppServerResourceEvent({ type: "skills-changed", forceReload: true }); + await actions.applyAppServerResourceEvent({ type: "skills-changed" }); expect(stateStore.getState().connection.availableSkills).toEqual([]); - expect(updateAppServerMetadata).not.toHaveBeenCalled(); }); - it("keeps previous skills when sparse skill refresh fails", async () => { + it("applies authoritative skills refreshed after a change event", async () => { let state = chatStateFixture(); const previousSkills = [skillFixture("writer")]; state = chatStateWith(state, { connection: { availableSkills: previousSkills } }); const stateStore = createChatStateStore(state); + const refreshed = serverMetadataFixture({ + availableSkills: [skillFixture("editor")], + serverDiagnostics: diagnosticsWithProbe(createServerDiagnostics(), diagnosticProbeOk("skills", "1 skill", 1)), + }); + const refreshSkills = vi.fn().mockResolvedValue(refreshed); const actions = createServerMetadataActions({ stateStore, - metadataResourceTransport: metadataResourceTransport({ - readSkillMetadata: vi.fn().mockResolvedValue({ - value: [], - probe: diagnosticProbeError("skills", new Error("offline"), 1), - }), - }), ...metadataCacheHost(), - refreshAppServerMetadata: vi.fn().mockResolvedValue(null), - isStaleSharedQueryError: () => false, + refreshSkills, }); - await actions.applyAppServerResourceEvent({ type: "skills-changed", forceReload: true }); + await actions.applyAppServerResourceEvent({ type: "skills-changed" }); - expect(stateStore.getState().connection.availableSkills).toEqual(previousSkills); - expect(stateStore.getState().connection.serverDiagnostics.probes.skills).toMatchObject({ status: "failed" }); + expect(refreshSkills).toHaveBeenCalledOnce(); + expect(stateStore.getState().connection.availableSkills.map((skill) => skill.name)).toEqual(["editor"]); }); - it("ignores an older skill refresh that completes after a newer refresh", async () => { - const older = deferred>>(); - const newer = deferred>>(); - const stateStore = createChatStateStore(chatStateFixture()); - const actions = createServerMetadataActions({ - stateStore, - metadataResourceTransport: metadataResourceTransport({ - readSkillMetadata: vi - .fn() - .mockImplementationOnce(() => older.promise) - .mockImplementationOnce(() => newer.promise), - }), - ...metadataCacheHost(), - refreshAppServerMetadata: vi.fn().mockResolvedValue(null), - isStaleSharedQueryError: () => false, - }); - - const first = actions.applyAppServerResourceEvent({ type: "skills-changed", forceReload: true }); - const second = actions.applyAppServerResourceEvent({ type: "skills-changed", forceReload: true }); - newer.resolve({ value: [skillFixture("new")], probe: diagnosticProbeOk("skills", "new", 2) }); - await second; - older.resolve({ value: [skillFixture("old")], probe: diagnosticProbeOk("skills", "old", 1) }); - await first; - - expect(stateStore.getState().connection.availableSkills.map((skill) => skill.name)).toEqual(["new"]); - }); - - it("shares skill refresh ordering across panel action instances", async () => { - const older = deferred>>(); - const cache = { current: serverMetadataFixture() as SharedServerMetadata | null }; - const sharedCache = metadataCacheHost(cache); - const createActions = (readSkillMetadata: MetadataResourceTransport["readSkillMetadata"]) => - createServerMetadataActions({ - stateStore: createChatStateStore(chatStateFixture()), - metadataResourceTransport: metadataResourceTransport({ readSkillMetadata }), - ...sharedCache, - refreshAppServerMetadata: vi.fn().mockResolvedValue(null), - isStaleSharedQueryError: () => false, - }); - const firstPanel = createActions(vi.fn(() => older.promise)); - const secondPanel = createActions( - vi.fn().mockResolvedValue({ value: [skillFixture("new")], probe: diagnosticProbeOk("skills", "new", 2) }), - ); - - const first = firstPanel.applyAppServerResourceEvent({ type: "skills-changed", forceReload: true }); - await secondPanel.applyAppServerResourceEvent({ type: "skills-changed", forceReload: true }); - older.resolve({ value: [skillFixture("old")], probe: diagnosticProbeOk("skills", "old", 1) }); - await first; - - expect(cache.current?.availableSkills.map((skill) => skill.name)).toEqual(["new"]); - }); - - it("publishes refreshed rate limits from sparse update notifications", async () => { + it("publishes refreshed rate limits from update notifications", async () => { const stateStore = createChatStateStore(chatStateFixture()); const rateLimit = rateLimitFixture({ primary: { usedPercent: 64, windowDurationMins: 300, resetsAt: null } }); - const cachedMetadata = { current: serverMetadataFixture() as SharedServerMetadata | null }; const actions = createServerMetadataActions({ stateStore, - metadataResourceTransport: metadataResourceTransport({ - readRateLimitMetadata: vi.fn().mockResolvedValue({ - value: rateLimit, - probe: diagnosticProbeOk("rateLimits", "available", 1), - }), - }), - ...metadataCacheHost(cachedMetadata), - refreshAppServerMetadata: vi.fn().mockResolvedValue(null), - isStaleSharedQueryError: () => false, + ...metadataCacheHost(), + refreshRateLimits: vi.fn().mockResolvedValue(serverMetadataFixture({ rateLimit })), }); - await actions.applyAppServerResourceEvent({ type: "rate-limits-updated", preserveExistingOnFailure: true }); + await actions.applyAppServerResourceEvent({ type: "rate-limits-updated" }); expect(stateStore.getState().connection.rateLimit).toMatchObject({ primary: { usedPercent: 64 } }); - expect(cachedMetadata.current?.rateLimit).toStrictEqual(rateLimit); - }); - - it("keeps the previous rate limit snapshot when sparse update refresh fails", async () => { - let state = chatStateFixture(); - const previousRateLimit = rateLimitFixture({ - limitName: "Codex", - primary: { usedPercent: 42, windowDurationMins: 300, resetsAt: null }, - }); - state = chatStateWith(state, { connection: { rateLimit: previousRateLimit } }); - const stateStore = createChatStateStore(state); - const actions = createServerMetadataActions({ - stateStore, - metadataResourceTransport: metadataResourceTransport({ - readRateLimitMetadata: vi.fn().mockResolvedValue({ - value: null, - probe: diagnosticProbeError("rateLimits", new Error("offline"), 1), - }), - }), - ...metadataCacheHost(), - refreshAppServerMetadata: vi.fn().mockResolvedValue(null), - isStaleSharedQueryError: () => false, - }); - - await actions.applyAppServerResourceEvent({ type: "rate-limits-updated", preserveExistingOnFailure: true }); - - expect(stateStore.getState().connection.rateLimit).toBe(previousRateLimit); - expect(stateStore.getState().connection.serverDiagnostics.probes.rateLimits).toMatchObject({ status: "failed" }); }); }); @@ -267,13 +167,13 @@ describe("server diagnostics actions", () => { diagnosticProbeOk("skills", "1 skills", 1), ), }); - const metadataCache = metadataCacheHost({ current: null }); + const cache = { current: null as SharedServerMetadata | null }; + const metadataCache = metadataCacheHost(cache); const metadata = createServerMetadataActions({ stateStore, - metadataResourceTransport: metadataResourceTransport(), ...metadataCache, refreshAppServerMetadata: vi.fn().mockImplementation(async () => { - metadataCache.updateAppServerMetadata(() => refreshedMetadata); + cache.current = refreshedMetadata; return refreshedMetadata; }), isStaleSharedQueryError: () => false, @@ -371,7 +271,6 @@ describe("server diagnostics actions", () => { const metadataCache = metadataCacheHost({ current: serverMetadataFixture() }); const metadata = createServerMetadataActions({ stateStore, - metadataResourceTransport: metadataResourceTransport(), ...metadataCache, refreshAppServerMetadata: vi.fn().mockResolvedValue(null), isStaleSharedQueryError: () => false, @@ -471,14 +370,6 @@ describe("server diagnostics actions", () => { }); }); -function metadataResourceTransport(overrides: Partial = {}): MetadataResourceTransport { - return { - readSkillMetadata: vi.fn().mockResolvedValue({ value: [], probe: diagnosticProbeOk("skills", "0 skills", 1) }), - readRateLimitMetadata: vi.fn().mockResolvedValue({ value: null, probe: diagnosticProbeOk("rateLimits", "unavailable", 1) }), - ...overrides, - }; -} - function serverDiagnosticsSnapshot( overrides: { resourceProbes?: DiagnosticProbeResult[]; mcpServerStatuses?: McpServerStatusSummary[] | null } = {}, ) { @@ -543,26 +434,18 @@ function serverMetadataFixture(overrides: Partial = {}): S } function metadataCacheHost(cache: { current: SharedServerMetadata | null } = { current: null }): { - beginAppServerMetadataResourceRefresh: (resource: "skills" | "rateLimits") => () => boolean; appServerMetadataSnapshot: () => SharedServerMetadata | null; - updateAppServerMetadata: ( - updater: (metadata: SharedServerMetadata | null) => SharedServerMetadata | null, - resource?: "skills" | "rateLimits", - ) => SharedServerMetadata | null; + refreshAppServerMetadata: () => Promise; + refreshSkills: () => Promise; + refreshRateLimits: () => Promise; + isStaleSharedQueryError: (error: unknown) => boolean; } { - const generations = { skills: 0, rateLimits: 0 }; return { - beginAppServerMetadataResourceRefresh: (resource) => { - const generation = ++generations[resource]; - return () => generation === generations[resource]; - }, appServerMetadataSnapshot: () => cache.current, - updateAppServerMetadata: (updater, resource) => { - const next = updater(cache.current); - cache.current = next; - if (resource) generations[resource] += 1; - return next; - }, + refreshAppServerMetadata: vi.fn().mockResolvedValue(null), + refreshSkills: vi.fn().mockResolvedValue(null), + refreshRateLimits: vi.fn().mockResolvedValue(null), + isStaleSharedQueryError: () => false, }; } diff --git a/tests/features/chat/host/session-runtime.test.ts b/tests/features/chat/host/session-runtime.test.ts index 558c40bf..de1fbfaf 100644 --- a/tests/features/chat/host/session-runtime.test.ts +++ b/tests/features/chat/host/session-runtime.test.ts @@ -366,10 +366,10 @@ describe("ChatPanelSessionRuntime actions", () => { overrides: Partial = {}, ): ChatPanelEnvironment["plugin"]["appServerQueries"] { return { - beginAppServerMetadataResourceRefresh: vi.fn(() => () => true), - updateAppServerMetadata: vi.fn(() => null), appServerMetadataSnapshot: vi.fn(() => null), refreshAppServerMetadata: vi.fn().mockResolvedValue(null), + refreshSkills: vi.fn().mockResolvedValue(null), + refreshRateLimits: vi.fn().mockResolvedValue(null), modelsSnapshot: vi.fn(() => null), fetchModels: vi.fn().mockResolvedValue([]), refreshModels: vi.fn().mockResolvedValue([]), diff --git a/tests/features/chat/host/view-connection.test.ts b/tests/features/chat/host/view-connection.test.ts index 46630846..9a13f5a7 100644 --- a/tests/features/chat/host/view-connection.test.ts +++ b/tests/features/chat/host/view-connection.test.ts @@ -1279,7 +1279,6 @@ interface ChatHostFixtureOverrides { refreshThreadsViewLiveState?: CodexChatHost["workspace"]["refreshThreadsViewLiveState"]; openSideChat?: CodexChatHost["workspace"]["openSideChat"]; applyThreadCatalogEvent?: CodexChatHost["threadCatalog"]["apply"]; - updateAppServerMetadata?: CodexChatHost["appServerQueries"]["updateAppServerMetadata"]; refreshActive?: CodexChatHost["threadCatalog"]["refreshActive"]; activeSnapshot?: CodexChatHost["threadCatalog"]["activeSnapshot"]; appServerMetadataSnapshot?: CodexChatHost["appServerQueries"]["appServerMetadataSnapshot"]; @@ -1287,6 +1286,8 @@ interface ChatHostFixtureOverrides { fetchModels?: CodexChatHost["appServerQueries"]["fetchModels"]; refreshModels?: CodexChatHost["appServerQueries"]["refreshModels"]; refreshAppServerMetadata?: CodexChatHost["appServerQueries"]["refreshAppServerMetadata"]; + refreshSkills?: CodexChatHost["appServerQueries"]["refreshSkills"]; + refreshRateLimits?: CodexChatHost["appServerQueries"]["refreshRateLimits"]; } function chatHost(overrides: ChatHostFixtureOverrides = {}): TestCodexChatHost { @@ -1308,7 +1309,7 @@ function chatHost(overrides: ChatHostFixtureOverrides = {}): TestCodexChatHost { for (const listener of metadataResultListeners) listener(queryResult(nextMetadata)); return nextMetadata; }; - const loadAppServerMetadata = async (): Promise => { + const loadAppServerMetadata = async (reloadSkills = false): Promise => { const client = connectionMock.state.client as TestAppServerClient | null; if (!client || !connectionMock.state.connected) return null; const connectionStillCurrent = () => connectionMock.state.client === client && connectionMock.state.connected; @@ -1326,7 +1327,7 @@ function chatHost(overrides: ChatHostFixtureOverrides = {}): TestCodexChatHost { if (!connectionStillCurrent()) return null; models = fetchedModels; for (const listener of modelResultListeners) listener(queryResult(fetchedModels)); - const skillsResponse = (await client.request("skills/list", { cwds: [vaultPath], forceReload: false })) as { + const skillsResponse = (await client.request("skills/list", { cwds: [vaultPath], forceReload: reloadSkills })) as { data: { skills: { name: string; description?: string; path?: string; enabled?: boolean }[] }[]; }; if (!connectionStillCurrent()) return null; @@ -1408,14 +1409,6 @@ function chatHost(overrides: ChatHostFixtureOverrides = {}): TestCodexChatHost { openSideChat: overrides.openSideChat ?? vi.fn().mockResolvedValue(undefined), }, appServerQueries: { - beginAppServerMetadataResourceRefresh: () => () => true, - updateAppServerMetadata: - overrides.updateAppServerMetadata ?? - ((updater) => { - const nextMetadata = updater(metadata); - if (!nextMetadata) return null; - return applyMetadataToCache(nextMetadata); - }), appServerMetadataSnapshot: overrides.appServerMetadataSnapshot ?? vi.fn(() => metadata), refreshAppServerMetadata: overrides.refreshAppServerMetadata ?? @@ -1423,6 +1416,18 @@ function chatHost(overrides: ChatHostFixtureOverrides = {}): TestCodexChatHost { const nextMetadata = await loadAppServerMetadata(); return nextMetadata ? applyMetadataToCache(nextMetadata) : null; }), + refreshSkills: + overrides.refreshSkills ?? + vi.fn(async () => { + const nextMetadata = await loadAppServerMetadata(true); + return nextMetadata ? applyMetadataToCache(nextMetadata) : null; + }), + refreshRateLimits: + overrides.refreshRateLimits ?? + vi.fn(async () => { + const nextMetadata = await loadAppServerMetadata(); + return nextMetadata ? applyMetadataToCache(nextMetadata) : null; + }), modelsSnapshot: overrides.modelsSnapshot ?? vi.fn(() => models), fetchModels: overrides.fetchModels ?? vi.fn(async () => models ?? []), refreshModels: overrides.refreshModels ?? vi.fn(async () => models ?? []), diff --git a/tests/main.test.ts b/tests/main.test.ts index 4fc3e12a..3780822e 100644 --- a/tests/main.test.ts +++ b/tests/main.test.ts @@ -937,10 +937,10 @@ function chatHostFixture(): CodexChatHost { openSideChat: vi.fn(), }, appServerQueries: { - beginAppServerMetadataResourceRefresh: vi.fn(() => () => true), - updateAppServerMetadata: vi.fn(() => null), appServerMetadataSnapshot: vi.fn(() => null), refreshAppServerMetadata: vi.fn(() => Promise.resolve(null)), + refreshSkills: vi.fn(() => Promise.resolve(null)), + refreshRateLimits: vi.fn(() => Promise.resolve(null)), modelsSnapshot: vi.fn(() => null), fetchModels: vi.fn(() => Promise.resolve([])), refreshModels: vi.fn(() => Promise.resolve([])),