mirror of
https://github.com/murashit/codex-panel.git
synced 2026-07-22 06:57:10 +00:00
refactor(query): require complete runtime-owned context
This commit is contained in:
murashit
parent
381646ebfb
commit
54dcbd8759
6 changed files with 114 additions and 212 deletions
|
|
@ -40,12 +40,10 @@ import {
|
|||
activeThreadsQueryKey,
|
||||
appServerModelsQueryKey,
|
||||
appServerPermissionProfilesQueryKey,
|
||||
appServerQueryContextIsComplete,
|
||||
appServerRateLimitsQueryKey,
|
||||
appServerRuntimeConfigQueryKey,
|
||||
appServerSkillsQueryKey,
|
||||
archivedThreadsQueryKey,
|
||||
cloneAppServerQueryContext,
|
||||
} from "./keys";
|
||||
import { readPermissionProfileMetadataProbe, readRateLimitMetadataProbe, readSkillMetadataProbe } from "./metadata-probes";
|
||||
import type { ObservedPaginatedResult, ObservedPaginatedResultListener, ObservedResult, ObservedResultListener } from "./observed-result";
|
||||
|
|
@ -55,11 +53,7 @@ import { applyThreadListMutation, type ThreadListKind, type ThreadListMutation }
|
|||
const MODELS_STALE_TIME_MS = 60_000;
|
||||
|
||||
export interface AppServerQueryClientRunner {
|
||||
runWithClient<T>(
|
||||
context: AppServerQueryContext,
|
||||
operation: (client: AppServerClient) => Promise<T>,
|
||||
options?: AppServerClientAccessOptions,
|
||||
): Promise<T>;
|
||||
runWithClient<T>(operation: (client: AppServerClient) => Promise<T>, options?: AppServerClientAccessOptions): Promise<T>;
|
||||
}
|
||||
|
||||
interface AppServerQueryOptions<T> {
|
||||
|
|
@ -79,15 +73,15 @@ type MetadataResourceKind = "skills" | "permissionProfiles" | "rateLimits";
|
|||
type MetadataResourceValue = readonly SkillMetadata[] | readonly RuntimePermissionProfileSummary[] | RateLimitSnapshot | null;
|
||||
|
||||
export class AppServerQueryCache {
|
||||
private readonly context: AppServerQueryContext;
|
||||
private readonly context: Readonly<AppServerQueryContext>;
|
||||
private readonly client: QueryClient;
|
||||
private readonly clientRunner: AppServerQueryClientRunner | null;
|
||||
private readonly clientRunner: AppServerQueryClientRunner;
|
||||
private disposed = false;
|
||||
|
||||
constructor(context: AppServerQueryContext, options: { client?: QueryClient; clientRunner?: AppServerQueryClientRunner } = {}) {
|
||||
this.context = cloneAppServerQueryContext(context);
|
||||
constructor(context: AppServerQueryContext, options: { client?: QueryClient; clientRunner: AppServerQueryClientRunner }) {
|
||||
this.context = Object.freeze({ ...context });
|
||||
this.client = options.client ?? createAppServerQueryClient();
|
||||
this.clientRunner = options.clientRunner ?? null;
|
||||
this.clientRunner = options.clientRunner;
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
|
|
@ -98,14 +92,13 @@ export class AppServerQueryCache {
|
|||
|
||||
activeThreadsSnapshot(): readonly Thread[] | null {
|
||||
if (this.disposed) return null;
|
||||
if (!appServerQueryContextIsComplete(this.context)) return null;
|
||||
const data = this.client.getQueryData<ActiveThreadData>(activeThreadsQueryKey(this.context));
|
||||
const threads = activeThreadsFromData(data);
|
||||
return threads ? cloneThreads(threads) : null;
|
||||
}
|
||||
|
||||
recentActiveThreadsSnapshot(): readonly Thread[] | null {
|
||||
if (this.disposed || !appServerQueryContextIsComplete(this.context)) return null;
|
||||
if (this.disposed) return null;
|
||||
const data = this.client.getQueryData<ActiveThreadData>(activeThreadsQueryKey(this.context));
|
||||
const threads = recentActiveThreadsFromData(data);
|
||||
return threads ? cloneThreads(threads) : null;
|
||||
|
|
@ -143,7 +136,6 @@ export class AppServerQueryCache {
|
|||
|
||||
async fetchActiveThreads(options: { force?: boolean } = {}): Promise<readonly Thread[]> {
|
||||
this.assertUsable();
|
||||
if (!appServerQueryContextIsComplete(this.context)) return [];
|
||||
const key = activeThreadsQueryKey(this.context);
|
||||
if (options.force) {
|
||||
if (this.client.getQueryState(key)?.fetchMeta?.fetchMore?.direction === "forward") {
|
||||
|
|
@ -167,7 +159,6 @@ export class AppServerQueryCache {
|
|||
|
||||
async fetchActiveThreadSearchInventory(): Promise<readonly Thread[]> {
|
||||
this.assertUsable();
|
||||
if (!appServerQueryContextIsComplete(this.context)) return [];
|
||||
const key = activeThreadSearchInventoryQueryKey(this.context);
|
||||
const options = {
|
||||
queryKey: key,
|
||||
|
|
@ -179,13 +170,11 @@ export class AppServerQueryCache {
|
|||
|
||||
hasMoreActiveThreads(): boolean {
|
||||
if (this.disposed) return false;
|
||||
if (!appServerQueryContextIsComplete(this.context)) return false;
|
||||
return activeThreadDataHasMore(this.client.getQueryData<ActiveThreadData>(activeThreadsQueryKey(this.context)));
|
||||
}
|
||||
|
||||
async loadMoreActiveThreads(): Promise<readonly Thread[]> {
|
||||
this.assertUsable();
|
||||
if (!appServerQueryContextIsComplete(this.context)) return [];
|
||||
const current = this.activeThreadsSnapshot() ?? (await this.fetchActiveThreads());
|
||||
const observer = new InfiniteQueryObserver(this.client, {
|
||||
...this.activeThreadsQueryOptions(),
|
||||
|
|
@ -210,7 +199,6 @@ export class AppServerQueryCache {
|
|||
|
||||
async fetchArchivedThreads(options: { force?: boolean } = {}): Promise<readonly Thread[]> {
|
||||
this.assertUsable();
|
||||
if (!appServerQueryContextIsComplete(this.context)) return [];
|
||||
const key = archivedThreadsQueryKey(this.context);
|
||||
if (options.force) {
|
||||
await this.client.invalidateQueries({ queryKey: key, refetchType: "none" });
|
||||
|
|
@ -223,7 +211,7 @@ export class AppServerQueryCache {
|
|||
|
||||
applyThreadListMutations(mutations: readonly ThreadListMutation[]): void {
|
||||
this.assertUsable();
|
||||
if (!appServerQueryContextIsComplete(this.context) || mutations.length === 0) return;
|
||||
if (mutations.length === 0) return;
|
||||
const activeMutations = mutations.filter((mutation) => mutation.list === "active");
|
||||
if (activeMutations.length > 0) {
|
||||
const key = activeThreadsQueryKey(this.context);
|
||||
|
|
@ -266,7 +254,6 @@ export class AppServerQueryCache {
|
|||
}
|
||||
|
||||
private threadListSnapshot(kind: ThreadListKind): readonly Thread[] | null {
|
||||
if (!appServerQueryContextIsComplete(this.context)) return null;
|
||||
if (kind === "active") return this.activeThreadsSnapshot();
|
||||
const threads = this.client.getQueryData<readonly Thread[]>(archivedThreadsQueryKey(this.context));
|
||||
return threads ? cloneThreads(threads) : null;
|
||||
|
|
@ -274,7 +261,6 @@ export class AppServerQueryCache {
|
|||
|
||||
appServerMetadataSnapshot(): SharedServerMetadata | null {
|
||||
if (this.disposed) return null;
|
||||
if (!appServerQueryContextIsComplete(this.context)) return null;
|
||||
const runtimeConfig = this.client.getQueryData<RuntimeConfigSnapshot>(appServerRuntimeConfigQueryKey(this.context));
|
||||
if (!runtimeConfig) return null;
|
||||
const skills = this.metadataResourceState("skills");
|
||||
|
|
@ -359,7 +345,6 @@ export class AppServerQueryCache {
|
|||
|
||||
async refreshAppServerMetadata(): Promise<void> {
|
||||
this.assertUsable();
|
||||
if (!appServerQueryContextIsComplete(this.context)) return;
|
||||
const runtimeResult = this.fetchRuntimeConfig().then(
|
||||
() => ({ ok: true as const }),
|
||||
(error: unknown) => ({ ok: false as const, error }),
|
||||
|
|
@ -379,21 +364,18 @@ export class AppServerQueryCache {
|
|||
|
||||
async refreshSkills(): Promise<void> {
|
||||
this.assertUsable();
|
||||
if (!appServerQueryContextIsComplete(this.context)) return;
|
||||
await this.refreshNotifiedMetadataResource("skills");
|
||||
this.assertUsable();
|
||||
}
|
||||
|
||||
async refreshRateLimits(): Promise<void> {
|
||||
this.assertUsable();
|
||||
if (!appServerQueryContextIsComplete(this.context)) return;
|
||||
await this.refreshNotifiedMetadataResource("rateLimits");
|
||||
this.assertUsable();
|
||||
}
|
||||
|
||||
modelsSnapshot(): readonly ModelMetadata[] | null {
|
||||
if (this.disposed) return null;
|
||||
if (!appServerQueryContextIsComplete(this.context)) return null;
|
||||
const models = this.client.getQueryData<readonly ModelMetadata[]>(appServerModelsQueryKey(this.context));
|
||||
return models ? cloneModelMetadata(models) : null;
|
||||
}
|
||||
|
|
@ -405,7 +387,6 @@ export class AppServerQueryCache {
|
|||
|
||||
async fetchModels(options: { force?: boolean } = {}): Promise<readonly ModelMetadata[]> {
|
||||
this.assertUsable();
|
||||
if (!appServerQueryContextIsComplete(this.context)) return [];
|
||||
const key = appServerModelsQueryKey(this.context);
|
||||
if (options.force) {
|
||||
await this.client.invalidateQueries({ queryKey: key, refetchType: "none" });
|
||||
|
|
@ -617,9 +598,7 @@ export class AppServerQueryCache {
|
|||
|
||||
private runWithClient<T>(operation: (client: AppServerClient) => Promise<T>, options: AppServerClientAccessOptions = {}): Promise<T> {
|
||||
this.assertUsable();
|
||||
const runner = this.clientRunner;
|
||||
if (!runner) throw new Error("Codex app-server query client runner is not configured.");
|
||||
return runner.runWithClient(this.context, operation, options);
|
||||
return this.clientRunner.runWithClient(operation, options);
|
||||
}
|
||||
|
||||
private async readThroughQueryCancellation<T>(read: () => Promise<T>, fallback?: () => T | null): Promise<T> {
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
export interface AppServerQueryContext {
|
||||
codexPath: string;
|
||||
vaultPath: string;
|
||||
readonly codexPath: string;
|
||||
readonly vaultPath: string;
|
||||
}
|
||||
|
||||
type AppServerQueryScope = readonly ["app-server", string, string];
|
||||
|
|
@ -13,22 +13,6 @@ 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);
|
||||
}
|
||||
|
||||
export function cloneAppServerQueryContext(context: AppServerQueryContext): AppServerQueryContext {
|
||||
return { ...context };
|
||||
}
|
||||
|
||||
function appServerQueryContextRawEquals(left: AppServerQueryContext, right: AppServerQueryContext): boolean {
|
||||
return left.codexPath === right.codexPath && left.vaultPath === right.vaultPath;
|
||||
}
|
||||
|
||||
export function appServerQueryContextMatches(left: AppServerQueryContext, right: AppServerQueryContext): boolean {
|
||||
return appServerQueryContextIsComplete(left) && appServerQueryContextIsComplete(right) && appServerQueryContextRawEquals(left, right);
|
||||
}
|
||||
|
||||
function appServerQueryScope(context: AppServerQueryContext): AppServerQueryScope {
|
||||
return ["app-server", context.codexPath, context.vaultPath];
|
||||
}
|
||||
|
|
@ -64,7 +48,3 @@ export function appServerPermissionProfilesQueryKey(context: AppServerQueryConte
|
|||
export function appServerRateLimitsQueryKey(context: AppServerQueryContext): AppServerRateLimitsQueryKey {
|
||||
return [...appServerQueryScope(context), "rate-limits"];
|
||||
}
|
||||
|
||||
function nonEmptyString(value: string): boolean {
|
||||
return value.trim().length > 0;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,8 +8,7 @@ import type { ThreadListMutation } from "./thread-list-mutation";
|
|||
|
||||
export interface AppServerResourceStoreOptions {
|
||||
context: AppServerQueryContext;
|
||||
cacheFactory?: (context: AppServerQueryContext) => AppServerQueryCache;
|
||||
clientRunner?: AppServerQueryClientRunner;
|
||||
clientRunner: AppServerQueryClientRunner;
|
||||
}
|
||||
|
||||
export class StaleAppServerResourceContextError extends Error {
|
||||
|
|
@ -23,43 +22,34 @@ export function isStaleAppServerResourceContextError(error: unknown): error is S
|
|||
return error instanceof StaleAppServerResourceContextError;
|
||||
}
|
||||
|
||||
interface ResourceObserver<T> {
|
||||
readonly observe: (cache: AppServerQueryCache, listener: (value: T) => void, options: { emitCurrent?: boolean }) => () => void;
|
||||
readonly listener: (value: T) => void;
|
||||
readonly options: { emitCurrent?: boolean };
|
||||
unsubscribeQuery: (() => void) | null;
|
||||
}
|
||||
|
||||
export class AppServerResourceStore {
|
||||
private readonly observers = new Set<ResourceObserver<unknown>>();
|
||||
private cache: AppServerQueryCache | null;
|
||||
private readonly cache: AppServerQueryCache;
|
||||
private readonly observerUnsubscribes = new Set<() => void>();
|
||||
private disposed = false;
|
||||
|
||||
constructor(options: AppServerResourceStoreOptions) {
|
||||
const cacheFactory =
|
||||
options.cacheFactory ??
|
||||
((context) => new AppServerQueryCache(context, options.clientRunner ? { clientRunner: options.clientRunner } : {}));
|
||||
this.cache = cacheFactory(Object.freeze({ ...options.context }));
|
||||
const context = Object.freeze({ ...options.context });
|
||||
this.cache = new AppServerQueryCache(context, { clientRunner: options.clientRunner });
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
if (this.disposed) return;
|
||||
this.disposed = true;
|
||||
this.unsubscribeObservers();
|
||||
this.cache?.dispose();
|
||||
this.cache = null;
|
||||
for (const unsubscribe of this.observerUnsubscribes) unsubscribe();
|
||||
this.observerUnsubscribes.clear();
|
||||
this.cache.dispose();
|
||||
}
|
||||
|
||||
activeThreadsSnapshot(): readonly Thread[] | null {
|
||||
return this.cache?.activeThreadsSnapshot() ?? null;
|
||||
return this.cache.activeThreadsSnapshot();
|
||||
}
|
||||
|
||||
recentActiveThreadsSnapshot(): readonly Thread[] | null {
|
||||
return this.cache?.recentActiveThreadsSnapshot() ?? null;
|
||||
return this.cache.recentActiveThreadsSnapshot();
|
||||
}
|
||||
|
||||
archivedThreadsSnapshot(): readonly Thread[] | null {
|
||||
return this.cache?.archivedThreadsSnapshot() ?? null;
|
||||
return this.cache.archivedThreadsSnapshot();
|
||||
}
|
||||
|
||||
fetchActiveThreadSearchInventory(): Promise<readonly Thread[]> {
|
||||
|
|
@ -67,7 +57,7 @@ export class AppServerResourceStore {
|
|||
}
|
||||
|
||||
hasMoreActiveThreads(): boolean {
|
||||
return this.cache?.hasMoreActiveThreads() ?? false;
|
||||
return this.cache.hasMoreActiveThreads();
|
||||
}
|
||||
|
||||
loadMoreActiveThreads(): Promise<readonly Thread[]> {
|
||||
|
|
@ -87,30 +77,23 @@ export class AppServerResourceStore {
|
|||
}
|
||||
|
||||
applyThreadListMutations(mutations: readonly ThreadListMutation[]): void {
|
||||
this.currentCache().applyThreadListMutations(mutations);
|
||||
this.assertActive();
|
||||
this.cache.applyThreadListMutations(mutations);
|
||||
}
|
||||
|
||||
observeActiveThreadsResult(
|
||||
listener: ObservedPaginatedResultListener<readonly Thread[]>,
|
||||
options?: { emitCurrent?: boolean },
|
||||
): () => void {
|
||||
return this.observeCurrentContext(
|
||||
(cache, contextListener, observeOptions) => cache.observeActiveThreadsResult(contextListener, observeOptions),
|
||||
listener,
|
||||
options,
|
||||
);
|
||||
return this.observe((contextListener) => this.cache.observeActiveThreadsResult(contextListener, options), listener);
|
||||
}
|
||||
|
||||
observeArchivedThreadsResult(listener: ObservedResultListener<readonly Thread[]>, options?: { emitCurrent?: boolean }): () => void {
|
||||
return this.observeCurrentContext(
|
||||
(cache, contextListener, observeOptions) => cache.observeArchivedThreadsResult(contextListener, observeOptions),
|
||||
listener,
|
||||
options,
|
||||
);
|
||||
return this.observe((contextListener) => this.cache.observeArchivedThreadsResult(contextListener, options), listener);
|
||||
}
|
||||
|
||||
appServerMetadataSnapshot(): SharedServerMetadata | null {
|
||||
return this.cache?.appServerMetadataSnapshot() ?? null;
|
||||
return this.cache.appServerMetadataSnapshot();
|
||||
}
|
||||
|
||||
refreshAppServerMetadata(): Promise<void> {
|
||||
|
|
@ -129,15 +112,11 @@ export class AppServerResourceStore {
|
|||
listener: (resource: SharedServerMetadataResource) => void,
|
||||
options?: { emitCurrent?: boolean },
|
||||
): () => void {
|
||||
return this.observeCurrentContext(
|
||||
(cache, contextListener, observeOptions) => cache.observeAppServerMetadataResources(contextListener, observeOptions),
|
||||
listener,
|
||||
options,
|
||||
);
|
||||
return this.observe((contextListener) => this.cache.observeAppServerMetadataResources(contextListener, options), listener);
|
||||
}
|
||||
|
||||
modelsSnapshot(): readonly ModelMetadata[] | null {
|
||||
return this.cache?.modelsSnapshot() ?? null;
|
||||
return this.cache.modelsSnapshot();
|
||||
}
|
||||
|
||||
fetchModels(): Promise<readonly ModelMetadata[]> {
|
||||
|
|
@ -149,73 +128,39 @@ export class AppServerResourceStore {
|
|||
}
|
||||
|
||||
observeModelsResult(listener: ObservedResultListener<readonly ModelMetadata[]>, options?: { emitCurrent?: boolean }): () => void {
|
||||
return this.observeCurrentContext(
|
||||
(cache, contextListener, observeOptions) => cache.observeModelsResult(contextListener, observeOptions),
|
||||
listener,
|
||||
options,
|
||||
);
|
||||
return this.observe((contextListener) => this.cache.observeModelsResult(contextListener, options), listener);
|
||||
}
|
||||
|
||||
private runForCurrentContext<T>(operation: (cache: AppServerQueryCache) => Promise<T>): Promise<T> {
|
||||
const cache = this.currentCache();
|
||||
return this.runWhileActive(cache, () => operation(cache));
|
||||
this.assertActive();
|
||||
return this.runWhileActive(() => operation(this.cache));
|
||||
}
|
||||
|
||||
private async runWhileActive<T>(cache: AppServerQueryCache, operation: () => Promise<T>): Promise<T> {
|
||||
private async runWhileActive<T>(operation: () => Promise<T>): Promise<T> {
|
||||
let result: T;
|
||||
try {
|
||||
result = await operation();
|
||||
} catch (error) {
|
||||
if (this.cache !== cache) throw new StaleAppServerResourceContextError();
|
||||
if (this.disposed) throw new StaleAppServerResourceContextError();
|
||||
throw error;
|
||||
}
|
||||
if (this.cache !== cache) throw new StaleAppServerResourceContextError();
|
||||
if (this.disposed) throw new StaleAppServerResourceContextError();
|
||||
return result;
|
||||
}
|
||||
|
||||
private observeCurrentContext<T>(
|
||||
observe: ResourceObserver<T>["observe"],
|
||||
listener: ResourceObserver<T>["listener"],
|
||||
options: { emitCurrent?: boolean } = {},
|
||||
): () => void {
|
||||
const observer: ResourceObserver<T> = {
|
||||
observe,
|
||||
listener,
|
||||
options,
|
||||
unsubscribeQuery: null,
|
||||
};
|
||||
this.observers.add(observer as ResourceObserver<unknown>);
|
||||
this.bindObserver(observer);
|
||||
private observe<T>(subscribe: (listener: (value: T) => void) => () => void, listener: (value: T) => void): () => void {
|
||||
this.assertActive();
|
||||
const unsubscribe = subscribe((value) => {
|
||||
if (!this.disposed) listener(value);
|
||||
});
|
||||
this.observerUnsubscribes.add(unsubscribe);
|
||||
return () => {
|
||||
this.observers.delete(observer as ResourceObserver<unknown>);
|
||||
observer.unsubscribeQuery?.();
|
||||
observer.unsubscribeQuery = null;
|
||||
if (!this.observerUnsubscribes.delete(unsubscribe)) return;
|
||||
unsubscribe();
|
||||
};
|
||||
}
|
||||
|
||||
private bindObserver<T>(observer: ResourceObserver<T>): void {
|
||||
if (this.disposed || !this.cache) return;
|
||||
const cache = this.cache;
|
||||
const observeOptions: { emitCurrent?: boolean } = {};
|
||||
if (observer.options.emitCurrent !== undefined) observeOptions.emitCurrent = observer.options.emitCurrent;
|
||||
observer.unsubscribeQuery = observer.observe(
|
||||
cache,
|
||||
(value) => {
|
||||
if (!this.disposed) observer.listener(value);
|
||||
},
|
||||
observeOptions,
|
||||
);
|
||||
}
|
||||
|
||||
private unsubscribeObservers(): void {
|
||||
for (const observer of this.observers) {
|
||||
observer.unsubscribeQuery?.();
|
||||
observer.unsubscribeQuery = null;
|
||||
}
|
||||
}
|
||||
|
||||
private currentCache(): AppServerQueryCache {
|
||||
if (!this.cache) throw new Error("Codex app-server resource store is not initialized.");
|
||||
return this.cache;
|
||||
private assertActive(): void {
|
||||
if (this.disposed) throw new StaleAppServerResourceContextError();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ import type { App } from "obsidian";
|
|||
import type { AppServerClient } from "./app-server/connection/client";
|
||||
import type { AppServerClientAccess, AppServerClientAccessOptions } from "./app-server/connection/client-access";
|
||||
import { withShortLivedAppServerClient } from "./app-server/connection/short-lived-client";
|
||||
import { type AppServerQueryContext, appServerQueryContextIsComplete, appServerQueryContextMatches } from "./app-server/query/keys";
|
||||
import type { AppServerQueryContext } from "./app-server/query/keys";
|
||||
import { AppServerResourceStore, StaleAppServerResourceContextError } from "./app-server/query/resource-store";
|
||||
import {
|
||||
type EphemeralStructuredTurnClient,
|
||||
|
|
@ -64,7 +64,7 @@ export class CodexExecutionRuntime implements AppServerClientAccess {
|
|||
this.resourceStore = new AppServerResourceStore({
|
||||
context: this.context,
|
||||
clientRunner: {
|
||||
runWithClient: (context, operation, clientOptions) => this.runWithAppServerClient(context, operation, clientOptions),
|
||||
runWithClient: (operation, clientOptions) => this.runWithAppServerClient(operation, clientOptions),
|
||||
},
|
||||
});
|
||||
this.threadCatalog = createThreadCatalog({
|
||||
|
|
@ -116,7 +116,7 @@ export class CodexExecutionRuntime implements AppServerClientAccess {
|
|||
}
|
||||
|
||||
withClient<T>(operation: (client: AppServerClient) => Promise<T>, options: AppServerClientAccessOptions = {}): Promise<T> {
|
||||
return this.runWithAppServerClient(this.context, operation, options);
|
||||
return this.runWithAppServerClient(operation, options);
|
||||
}
|
||||
|
||||
attachChatView(view: ChatRuntimeView): void {
|
||||
|
|
@ -264,16 +264,15 @@ export class CodexExecutionRuntime implements AppServerClientAccess {
|
|||
}
|
||||
|
||||
private async runWithAppServerClient<T>(
|
||||
context: AppServerQueryContext,
|
||||
operation: (client: AppServerClient) => Promise<T>,
|
||||
options: AppServerClientAccessOptions = {},
|
||||
): Promise<T> {
|
||||
this.assertCurrent(context);
|
||||
this.assertActive();
|
||||
const guardedOperation = (client: AppServerClient): Promise<T> => {
|
||||
this.assertCurrent(context);
|
||||
this.assertActive();
|
||||
return operation(client);
|
||||
};
|
||||
const result = await withShortLivedAppServerClient(context.codexPath, context.vaultPath, guardedOperation, options, {
|
||||
const result = await withShortLivedAppServerClient(this.context.codexPath, this.context.vaultPath, guardedOperation, options, {
|
||||
created: (client) => {
|
||||
if (this.disposed) {
|
||||
client.disconnect();
|
||||
|
|
@ -285,7 +284,7 @@ export class CodexExecutionRuntime implements AppServerClientAccess {
|
|||
this.shortLivedClients.delete(client);
|
||||
},
|
||||
});
|
||||
this.assertCurrent(context);
|
||||
this.assertActive();
|
||||
return result;
|
||||
}
|
||||
|
||||
|
|
@ -335,13 +334,6 @@ export class CodexExecutionRuntime implements AppServerClientAccess {
|
|||
});
|
||||
}
|
||||
|
||||
private assertCurrent(context: AppServerQueryContext): void {
|
||||
if (this.disposed || !appServerQueryContextIsComplete(context)) throw new StaleAppServerResourceContextError();
|
||||
if (!appServerQueryContextMatches(this.context, context)) {
|
||||
throw new StaleAppServerResourceContextError();
|
||||
}
|
||||
}
|
||||
|
||||
private assertActive(): void {
|
||||
if (this.disposed) throw new StaleAppServerResourceContextError();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,16 +8,6 @@ import type { RuntimePermissionProfileSummary } from "../../src/domain/runtime/p
|
|||
import type { Thread } from "../../src/domain/threads/model";
|
||||
|
||||
describe("AppServerQueryCache", () => {
|
||||
it("does not share or store snapshots before the cache context is complete", async () => {
|
||||
const context = cacheContext({ codexPath: "" });
|
||||
const cache = new AppServerQueryCache(context);
|
||||
|
||||
await expect(cache.refreshActiveThreads()).resolves.toEqual([]);
|
||||
expect(cache.activeThreadsSnapshot()).toBeNull();
|
||||
expect(cache.appServerMetadataSnapshot()).toBeNull();
|
||||
expect(cache.modelsSnapshot()).toBeNull();
|
||||
});
|
||||
|
||||
it("stores successful empty thread list snapshots as shared cache truth", async () => {
|
||||
const fetchThreads = vi.fn().mockResolvedValue([]);
|
||||
const cache = cacheWithThreads(fetchThreads);
|
||||
|
|
@ -444,7 +434,7 @@ describe("AppServerQueryCache", () => {
|
|||
});
|
||||
|
||||
it("freezes its lease context before starting requests", async () => {
|
||||
const context = cacheContext({ codexPath: "codex-captured" });
|
||||
const context = { codexPath: "codex-captured", vaultPath: "/vault" };
|
||||
const capturedContext = { ...context };
|
||||
const refresh = deferred<readonly ReturnType<typeof thread>[]>();
|
||||
const fetchThreads = vi.fn(() => refresh.promise);
|
||||
|
|
@ -783,14 +773,15 @@ function cacheWithThreads(
|
|||
fetchThreads: (context: AppServerQueryContext, archived: boolean) => Promise<readonly ReturnType<typeof thread>[]>,
|
||||
context: AppServerQueryContext = cacheContext(),
|
||||
): AppServerQueryCache {
|
||||
const runtimeContext = { ...context };
|
||||
return new AppServerQueryCache(context, {
|
||||
clientRunner: {
|
||||
runWithClient: async (context, operation) => {
|
||||
runWithClient: async (operation) => {
|
||||
return operation({
|
||||
request: async (method: string, params: { archived?: boolean }) => {
|
||||
if (method !== "thread/list") throw new Error(`Unexpected app-server request: ${method}`);
|
||||
return {
|
||||
data: await fetchThreads(context, params.archived ?? false),
|
||||
data: await fetchThreads(runtimeContext, params.archived ?? false),
|
||||
nextCursor: null,
|
||||
};
|
||||
},
|
||||
|
|
@ -813,7 +804,7 @@ function cacheWithRequestHandlers(
|
|||
};
|
||||
return new AppServerQueryCache(context, {
|
||||
clientRunner: {
|
||||
runWithClient: async (_context, operation) => operation(requestClient as never),
|
||||
runWithClient: async (operation) => operation(requestClient as never),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,58 +1,73 @@
|
|||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import type { AppServerQueryCache } from "../../src/app-server/query/cache";
|
||||
import type { AppServerQueryContext } from "../../src/app-server/query/keys";
|
||||
import { AppServerResourceStore } from "../../src/app-server/query/resource-store";
|
||||
import type { AppServerQueryClientRunner } from "../../src/app-server/query/cache";
|
||||
import { AppServerResourceStore, StaleAppServerResourceContextError } from "../../src/app-server/query/resource-store";
|
||||
|
||||
describe("AppServerResourceStore", () => {
|
||||
it("constructs its only query cache from an immutable context", () => {
|
||||
const cacheFactory = vi.fn((_context: AppServerQueryContext) => cacheWith());
|
||||
const context = { codexPath: "codex-a", vaultPath: "/vault" };
|
||||
it("uses its required runtime-owned client runner", async () => {
|
||||
const runWithClient = vi.fn(async (operation) =>
|
||||
operation({
|
||||
request: vi.fn().mockResolvedValue({ data: [], nextCursor: null }),
|
||||
} as never),
|
||||
);
|
||||
const store = createStore({ runWithClient });
|
||||
|
||||
new AppServerResourceStore({ context, cacheFactory });
|
||||
context.codexPath = "changed";
|
||||
await expect(store.fetchActiveThreads()).resolves.toEqual([]);
|
||||
|
||||
expect(cacheFactory).toHaveBeenCalledWith({ codexPath: "codex-a", vaultPath: "/vault" });
|
||||
expect(Object.isFrozen(cacheFactory.mock.calls[0]?.[0])).toBe(true);
|
||||
expect(runWithClient).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("disposes its only query cache once", () => {
|
||||
const cache = cacheWith();
|
||||
const store = new AppServerResourceStore({
|
||||
context: { codexPath: "codex", vaultPath: "/vault" },
|
||||
cacheFactory: () => cache,
|
||||
it("rejects new work after disposal", async () => {
|
||||
const store = createStore({
|
||||
runWithClient: vi.fn(() => Promise.resolve([])) as AppServerQueryClientRunner["runWithClient"],
|
||||
});
|
||||
|
||||
store.dispose();
|
||||
store.dispose();
|
||||
|
||||
expect(cache.dispose).toHaveBeenCalledOnce();
|
||||
expect(() => store.fetchModels()).toThrow(StaleAppServerResourceContextError);
|
||||
});
|
||||
|
||||
it("rejects a completion after its runtime-owned store is disposed", async () => {
|
||||
let resolveFetch: (models: readonly []) => void = () => undefined;
|
||||
const pending = new Promise<readonly []>((resolve) => {
|
||||
resolveFetch = resolve;
|
||||
});
|
||||
const store = createStore({
|
||||
runWithClient: vi.fn(() => pending) as AppServerQueryClientRunner["runWithClient"],
|
||||
});
|
||||
|
||||
const fetch = store.fetchModels();
|
||||
store.dispose();
|
||||
resolveFetch([]);
|
||||
|
||||
await expect(fetch).rejects.toBeInstanceOf(StaleAppServerResourceContextError);
|
||||
});
|
||||
|
||||
it("does not notify observers after disposal", async () => {
|
||||
let resolveFetch: (models: readonly []) => void = () => undefined;
|
||||
const pending = new Promise<readonly []>((resolve) => {
|
||||
resolveFetch = resolve;
|
||||
});
|
||||
const store = createStore({
|
||||
runWithClient: vi.fn(() => pending) as AppServerQueryClientRunner["runWithClient"],
|
||||
});
|
||||
const listener = vi.fn();
|
||||
store.observeModelsResult(listener, { emitCurrent: false });
|
||||
|
||||
const fetch = store.fetchModels();
|
||||
listener.mockClear();
|
||||
store.dispose();
|
||||
resolveFetch([]);
|
||||
await expect(fetch).rejects.toBeInstanceOf(StaleAppServerResourceContextError);
|
||||
|
||||
expect(listener).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
function cacheWith(overrides: Partial<AppServerQueryCache> = {}): AppServerQueryCache {
|
||||
return {
|
||||
dispose: vi.fn(),
|
||||
activeThreadsSnapshot: vi.fn(() => null),
|
||||
recentActiveThreadsSnapshot: vi.fn(() => null),
|
||||
archivedThreadsSnapshot: vi.fn(() => null),
|
||||
fetchActiveThreadSearchInventory: vi.fn(() => Promise.resolve([])),
|
||||
fetchActiveThreads: vi.fn(() => Promise.resolve([])),
|
||||
hasMoreActiveThreads: vi.fn(() => false),
|
||||
loadMoreActiveThreads: vi.fn(() => Promise.resolve([])),
|
||||
refreshActiveThreads: vi.fn(() => Promise.resolve([])),
|
||||
refreshArchivedThreads: vi.fn(() => Promise.resolve([])),
|
||||
observeActiveThreadsResult: vi.fn(() => () => undefined),
|
||||
observeArchivedThreadsResult: vi.fn(() => () => undefined),
|
||||
appServerMetadataSnapshot: vi.fn(() => null),
|
||||
refreshAppServerMetadata: vi.fn(() => Promise.resolve()),
|
||||
refreshSkills: vi.fn(() => Promise.resolve()),
|
||||
refreshRateLimits: vi.fn(() => Promise.resolve()),
|
||||
observeAppServerMetadataResources: vi.fn(() => () => undefined),
|
||||
modelsSnapshot: vi.fn(() => null),
|
||||
fetchModels: vi.fn(() => Promise.resolve([])),
|
||||
refreshModels: vi.fn(() => Promise.resolve([])),
|
||||
observeModelsResult: vi.fn(() => () => undefined),
|
||||
...overrides,
|
||||
} as unknown as AppServerQueryCache;
|
||||
function createStore(clientRunner: AppServerQueryClientRunner): AppServerResourceStore {
|
||||
return new AppServerResourceStore({
|
||||
context: { codexPath: "codex", vaultPath: "/vault" },
|
||||
clientRunner,
|
||||
});
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue