obsidian-plugin/murashit_codex-panel
1
0
Fork 0
mirror of https://github.com/murashit/codex-panel.git synced 2026-07-22 06:57:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactor(app-server): make TanStack Query authoritative

Browse source
This commit is contained in:
murashit 2026-07-18 22:11:38 +09:00
parent ea665af455
commit cf0548a0d0
25 changed files with 449 additions and 546 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
docs/design.md
View file

@ -48,7 +48,9 @@ Obsidian and app-server boundaries stay outside Preact components. External life
Chat-visible state should have one authoritative owner. Components should consume narrow projections of that state rather than mirror it into another reactive store.
Each shared app-server resource should likewise have one authoritative query record. UI subscriptions should stay narrow enough that unrelated updates do not disturb independent panel regions.
TanStack Query is the single panel-side owner of cached app-server resources. Features may project authoritative event results into that state, but should not introduce parallel cache or synchronization mechanisms. Partial read models are reconciled at explicit lifecycle boundaries rather than kept globally and continuously consistent.
Reads with different completeness requirements have separate lifecycles. Complete operation-local reads must not replace bounded shared history, and transient activity should come from its owning live state rather than forcing history refreshes.
Imperative UI bridges are allowed only where the host platform requires them. They should remain narrow boundary adapters, not a second UI composition system inside Preact-owned surfaces.

425
src/app-server/query/cache.ts
View file

@ -1,4 +1,13 @@
import { QueryClient, QueryObserver, type QueryObserverResult } from "@tanstack/query-core";
import {
CancelledError,
type InfiniteData,
InfiniteQueryObserver,
type InfiniteQueryObserverOptions,
type InfiniteQueryObserverResult,
QueryClient,
QueryObserver,
type QueryObserverResult,
} from "@tanstack/query-core";
import type { ModelMetadata, SkillMetadata } from "../../domain/catalog/metadata";
import { cloneRuntimeConfigSnapshot, type RuntimeConfigSnapshot } from "../../domain/runtime/config";
import type { RateLimitSnapshot } from "../../domain/runtime/metrics";
@ -17,7 +26,7 @@ import type { AppServerClientAccessOptions } from "../connection/client-access";
import { runtimeConfigSnapshotFromAppServerConfig } from "../protocol/runtime-config";
import { listModelMetadata } from "../services/catalog";
import { readEffectiveConfig } from "../services/runtime-metadata";
import { listThreads, readThreadPage } from "../services/threads";
import { listThreads, readThreadPage, type ThreadPage } from "../services/threads";
import {
type AppServerQueryContextIdentity as AppServerQueryContext,
activeThreadsQueryKey,
@ -35,11 +44,7 @@ import type { ObservedResult, ObservedResultListener } from "./observed-result";
import { cloneModelMetadata, cloneSharedServerMetadata, cloneSharedServerMetadataResource, cloneThreads } from "./snapshots";
import { applyThreadListMutation, type ThreadListKind, type ThreadListMutation } from "./thread-list-mutation";
const THREAD_LIST_STALE_TIME_MS = 10_000;
const APP_SERVER_METADATA_STALE_TIME_MS = 10_000;
const MODELS_STALE_TIME_MS = 60_000;
const APP_SERVER_QUERY_GC_TIME_MS = 5 * 60_000;
const FULL_ACTIVE_THREAD_FETCH_ATTEMPTS = 2;
export interface AppServerQueryClientRunner {
runWithClient<T>(
@ -52,35 +57,25 @@ export interface AppServerQueryClientRunner {
interface AppServerQueryOptions<T> {
readonly queryKey: readonly unknown[];
readonly queryFn: () => Promise<T>;
readonly staleTime: number;
readonly staleTime?: number;
}
type ActiveThreadPageParam = string | null;
type ActiveThreadListData = InfiniteData<ThreadPage, ActiveThreadPageParam>;
type ActiveThreadsQueryKey = ReturnType<typeof activeThreadsQueryKey>;
interface MetadataResourceSnapshot<T> {
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<void>;
}
export class AppServerQueryCache {
private readonly context: AppServerQueryContext;
private readonly client: QueryClient;
private readonly clientRunner: AppServerQueryClientRunner | null;
private activeThreadCursorKnown = false;
private activeThreadCursor: string | null = null;
private activeThreadRevision = 0;
private readonly threadListEpochs: Record<ThreadListKind, number> = { active: 0, archived: 0 };
private readonly metadataResourceFetches = new Map<MetadataResourceKind, Set<Promise<void>>>();
private readonly metadataNotificationRefreshes = new Map<MetadataNotificationResourceKind, MetadataNotificationRefresh>();
private generation = 0;
private disposed = false;
constructor(context: AppServerQueryContext, options: { client?: QueryClient; clientRunner?: AppServerQueryClientRunner } = {}) {
@ -92,14 +87,6 @@ export class AppServerQueryCache {
dispose(): void {
if (this.disposed) return;
this.disposed = true;
this.generation += 1;
this.activeThreadCursorKnown = false;
this.activeThreadCursor = null;
this.activeThreadRevision = 0;
this.threadListEpochs.active = 0;
this.threadListEpochs.archived = 0;
this.metadataResourceFetches.clear();
this.metadataNotificationRefreshes.clear();
this.client.clear();
}
@ -115,70 +102,58 @@ export class AppServerQueryCache {
observeActiveThreadsResult(listener: ObservedResultListener<readonly Thread[]>, options: { emitCurrent?: boolean } = {}): () => void {
this.assertUsable();
return this.observeQueryResult(this.threadListQueryOptions("active"), cloneThreads, listener, options);
const observer = this.activeThreadListObserver();
const emit = (result: InfiniteQueryObserverResult<ActiveThreadListData>): void => {
listener(this.projectObservedResult(result, flattenActiveThreadList));
};
const unsubscribe = observer.subscribe(emit);
if (options.emitCurrent ?? true) emit(observer.getCurrentResult());
return unsubscribe;
}
observeArchivedThreadsResult(listener: ObservedResultListener<readonly Thread[]>, options: { emitCurrent?: boolean } = {}): () => void {
this.assertUsable();
return this.observeQueryResult(this.threadListQueryOptions("archived"), cloneThreads, listener, options);
}
async fetchActiveThreads(options: { force?: boolean } = {}): Promise<readonly Thread[]> {
this.assertUsable();
return this.fetchThreadList("active", options);
}
async fetchArchivedThreads(options: { force?: boolean } = {}): Promise<readonly Thread[]> {
this.assertUsable();
return this.fetchThreadList("archived", options);
return this.observeQueryResult(this.archivedThreadListQueryOptions(), cloneThreads, listener, options);
}
async refreshActiveThreads(): Promise<readonly Thread[]> {
return this.fetchActiveThreads({ force: true });
this.assertUsable();
return this.refreshActiveThreadList();
}
async fetchAllActiveThreads(): Promise<readonly Thread[]> {
this.assertUsable();
if (!appServerQueryContextIsComplete(this.context)) return [];
const snapshot = this.activeThreadsSnapshot();
if (snapshot && this.activeThreadCursorKnown && !this.activeThreadCursor) return snapshot;
for (let attempt = 0; attempt < FULL_ACTIVE_THREAD_FETCH_ATTEMPTS; attempt += 1) {
const revision = this.activeThreadRevision;
const threads = await this.runWithClient((client) => listThreads(client, this.context.vaultPath));
if (this.activeThreadRevision !== revision) continue;
this.storeThreadList("active", threads);
this.rememberActiveThreadCursor(null);
return cloneThreads(threads);
}
throw new Error("Active thread inventory changed while it was being fetched.");
return cloneThreads(await this.runWithClient((client) => listThreads(client, this.context.vaultPath)));
}
hasMoreActiveThreads(): boolean {
if (this.disposed) return false;
if (!appServerQueryContextIsComplete(this.context)) return false;
return Boolean(this.activeThreadCursor);
return Boolean(this.activeThreadListData()?.pages.at(-1)?.nextCursor);
}
async loadMoreActiveThreads(): Promise<readonly Thread[]> {
this.assertUsable();
if (!appServerQueryContextIsComplete(this.context)) return [];
const current = this.activeThreadsSnapshot() ?? (await this.fetchActiveThreads());
const cursor = this.activeThreadCursor;
if (!cursor) return current;
const revision = this.activeThreadRevision;
const page = await this.runWithClient((client) => readThreadPage(client, this.context.vaultPath, { cursor, archived: false }));
if (page.nextCursor === cursor) throw new Error("Codex app-server returned a repeated thread list cursor.");
if (this.activeThreadRevision !== revision) return this.activeThreadsSnapshot() ?? current;
const latest = this.activeThreadsSnapshot() ?? current;
const existingIds = new Set(latest.map((thread) => thread.id));
const threads = [...latest, ...page.threads.filter((thread) => !existingIds.has(thread.id))];
this.storeThreadList("active", threads);
this.rememberActiveThreadCursor(page.nextCursor);
return cloneThreads(threads);
const current = this.activeThreadsSnapshot() ?? (await this.fetchActiveThreadList());
if (!this.hasMoreActiveThreads()) return current;
const observer = this.activeThreadListObserver();
try {
const result = await observer.fetchNextPage({ cancelRefetch: false, throwOnError: true });
this.assertUsable();
return result.data ? flattenActiveThreadList(result.data) : current;
} catch (error) {
if (error instanceof CancelledError) return this.activeThreadsSnapshot() ?? current;
throw error;
} finally {
observer.destroy();
}
}
async refreshArchivedThreads(): Promise<readonly Thread[]> {
return this.fetchArchivedThreads({ force: true });
this.assertUsable();
return this.refreshArchivedThreadList();
}
applyThreadListMutations(mutations: readonly ThreadListMutation[]): void {
@ -188,17 +163,20 @@ export class AppServerQueryCache {
const relevant = mutations.filter((mutation) => mutation.list === kind);
if (relevant.length === 0) continue;
const key = this.threadListQueryKey(kind);
const key = kind === "active" ? activeThreadsQueryKey(this.context) : archivedThreadsQueryKey(this.context);
const before = this.threadListSnapshot(kind);
const wasFetching = this.client.getQueryState(key)?.fetchStatus === "fetching";
this.threadListEpochs[kind] += 1;
const after = relevant.reduce<readonly Thread[] | null>(applyThreadListMutation, before);
if (after) this.storeThreadList(kind, after);
const partialSnapshotCreated = before === null && after !== null;
const refreshRequested = relevant.some((mutation) => mutation.kind === "refresh");
if (before !== null || wasFetching || partialSnapshotCreated || refreshRequested) {
void this.fetchThreadList(kind, { force: true }).catch(() => {
const activeFetchIsNextPage = wasFetching && kind === "active" && this.activeThreadListIsFetchingNextPage();
void this.client.cancelQueries({ queryKey: key, exact: true });
if (kind === "active") {
for (const mutation of relevant) this.projectActiveThreadListMutation(mutation);
} else {
const projected = before ? relevant.reduce<readonly Thread[] | null>(applyThreadListMutation, before) : null;
if (projected) this.client.setQueryData(archivedThreadsQueryKey(this.context), cloneThreads(projected));
}
if (wasFetching && !activeFetchIsNextPage) {
const refresh = kind === "active" ? this.refreshActiveThreadList() : this.refreshArchivedThreadList();
void refresh.catch(() => {
// Query observers retain refresh failures while the event projection remains last-known-good state.
});
}
@ -207,26 +185,79 @@ export class AppServerQueryCache {
private threadListSnapshot(kind: ThreadListKind): readonly Thread[] | null {
if (!appServerQueryContextIsComplete(this.context)) return null;
const threads = this.client.getQueryData<readonly Thread[]>(this.threadListQueryKey(kind));
if (kind === "active") {
const data = this.activeThreadListData();
return data ? flattenActiveThreadList(data) : null;
}
const threads = this.client.getQueryData<readonly Thread[]>(archivedThreadsQueryKey(this.context));
return threads ? cloneThreads(threads) : null;
}
private async fetchThreadList(kind: ThreadListKind, options: { force?: boolean } = {}): Promise<readonly Thread[]> {
private async refreshActiveThreadList(): Promise<readonly Thread[]> {
if (!appServerQueryContextIsComplete(this.context)) return [];
const key = this.threadListQueryKey(kind);
if (options.force) {
await this.client.invalidateQueries({ queryKey: key });
this.assertUsable();
const key = activeThreadsQueryKey(this.context);
if (this.activeThreadListIsFetchingNextPage()) await this.client.cancelQueries({ queryKey: key, exact: true });
await this.client.invalidateQueries({ queryKey: key, refetchType: "none" });
this.assertUsable();
return this.fetchActiveThreadList();
}
private async fetchActiveThreadList(): Promise<readonly Thread[]> {
if (!appServerQueryContextIsComplete(this.context)) return [];
const queryOptions = { ...this.activeThreadListQueryOptions(), pages: 1 };
let data: ActiveThreadListData;
try {
data = await this.client.fetchInfiniteQuery(queryOptions);
} catch (error) {
if (!(error instanceof CancelledError) || this.activeThreadListData()) throw error;
data = await this.client.fetchInfiniteQuery(queryOptions);
}
this.assertUsable();
return flattenActiveThreadList(data);
}
private async refreshArchivedThreadList(): Promise<readonly Thread[]> {
if (!appServerQueryContextIsComplete(this.context)) return [];
const key = archivedThreadsQueryKey(this.context);
await this.client.invalidateQueries({ queryKey: key, refetchType: "none" });
this.assertUsable();
return this.fetchArchivedThreadList();
}
private async fetchArchivedThreadList(): Promise<readonly Thread[]> {
if (!appServerQueryContextIsComplete(this.context)) return [];
let threads: readonly Thread[];
try {
threads = await this.client.fetchQuery(this.archivedThreadListQueryOptions());
} catch (error) {
if (!(error instanceof CancelledError) || this.archivedThreadsSnapshot()) throw error;
threads = await this.client.fetchQuery(this.archivedThreadListQueryOptions());
}
const threads = await this.client.fetchQuery(this.threadListQueryOptions(kind));
this.assertUsable();
return cloneThreads(threads);
}
private storeThreadList(kind: ThreadListKind, threads: readonly Thread[]): void {
if (!appServerQueryContextIsComplete(this.context)) return;
this.client.setQueryData(this.threadListQueryKey(kind), cloneThreads(threads));
if (kind === "active") this.bumpActiveThreadRevision();
private projectActiveThreadListMutation(mutation: ThreadListMutation): void {
this.client.setQueryData<ActiveThreadListData>(activeThreadsQueryKey(this.context), (data) => {
if (!data) return data;
if (mutation.kind === "upsert") {
const existing = data.pages.some((page) => page.threads.some((thread) => thread.id === mutation.thread.id));
const pages = data.pages.map((page) => ({
...page,
threads: page.threads.map((thread) => (thread.id === mutation.thread.id ? mutation.thread : thread)),
}));
const first = pages[0];
if (!existing && first) pages[0] = { ...first, threads: [mutation.thread, ...first.threads] };
return { ...data, pages };
}
return {
...data,
pages: data.pages.map((page) => ({
...page,
threads: applyThreadListMutation(page.threads, mutation) ?? page.threads,
})),
};
});
}
appServerMetadataSnapshot(): SharedServerMetadata | null {
@ -255,10 +286,9 @@ export class AppServerQueryCache {
options: { emitCurrent?: boolean } = {},
): () => void {
this.assertUsable();
const generation = this.generation;
let disposed = false;
const emit = (resource: SharedServerMetadataResource): void => {
if (!disposed && generation === this.generation) listener(cloneSharedServerMetadataResource(resource));
if (!disposed) listener(cloneSharedServerMetadataResource(resource));
};
const runtimeConfig = new QueryObserver(this.client, { ...this.runtimeConfigQueryOptions(), enabled: false });
@ -318,15 +348,15 @@ export class AppServerQueryCache {
async refreshAppServerMetadata(): Promise<void> {
this.assertUsable();
if (!appServerQueryContextIsComplete(this.context)) return;
const runtimeResult = this.fetchRuntimeConfig(true).then(
const runtimeResult = this.fetchRuntimeConfig().then(
() => ({ ok: true as const }),
(error: unknown) => ({ ok: false as const, error }),
);
const [, runtime] = await Promise.all([
Promise.allSettled([
this.fetchMetadataResource("skills", true),
this.fetchMetadataResource("permissionProfiles", true),
this.fetchMetadataResource("rateLimits", true),
this.fetchMetadataResource("skills"),
this.fetchMetadataResource("permissionProfiles"),
this.fetchMetadataResource("rateLimits"),
this.fetchModels({ force: true }),
]),
runtimeResult,
@ -338,14 +368,14 @@ export class AppServerQueryCache {
async refreshSkills(): Promise<void> {
this.assertUsable();
if (!appServerQueryContextIsComplete(this.context)) return;
await this.refreshMetadataResourceNotification("skills");
await this.refreshNotifiedMetadataResource("skills");
this.assertUsable();
}
async refreshRateLimits(): Promise<void> {
this.assertUsable();
if (!appServerQueryContextIsComplete(this.context)) return;
await this.refreshMetadataResourceNotification("rateLimits");
await this.refreshNotifiedMetadataResource("rateLimits");
this.assertUsable();
}
@ -366,7 +396,7 @@ export class AppServerQueryCache {
if (!appServerQueryContextIsComplete(this.context)) return [];
const key = appServerModelsQueryKey(this.context);
if (options.force) {
await this.client.invalidateQueries({ queryKey: key });
await this.client.invalidateQueries({ queryKey: key, refetchType: "none" });
this.assertUsable();
}
const models = await this.client.fetchQuery(this.modelsQueryOptions());
@ -378,33 +408,49 @@ export class AppServerQueryCache {
return this.fetchModels({ force: true });
}
private threadListQueryOptions(kind: ThreadListKind): AppServerQueryOptions<readonly Thread[]> {
private activeThreadListQueryOptions(): InfiniteQueryObserverOptions<
ThreadPage,
Error,
ActiveThreadListData,
ActiveThreadsQueryKey,
ActiveThreadPageParam
> {
return {
queryKey: this.threadListQueryKey(kind),
queryFn: async (): Promise<readonly Thread[]> => {
if (kind === "active") {
for (;;) {
const epoch = this.threadListEpochs.active;
const revision = this.activeThreadRevision;
const page = await this.runWithClient((client) => readThreadPage(client, this.context.vaultPath, { archived: false }));
if (this.threadListEpochs.active !== epoch) continue;
if (this.activeThreadRevision !== revision) return this.activeThreadsSnapshot() ?? [];
this.rememberActiveThreadCursor(page.nextCursor);
return cloneThreads(page.threads);
}
}
for (;;) {
const epoch = this.threadListEpochs.archived;
const threads = await this.runWithClient((client) => listThreads(client, this.context.vaultPath, { archived: true }));
if (this.threadListEpochs.archived === epoch) return cloneThreads(threads);
}
},
staleTime: THREAD_LIST_STALE_TIME_MS,
queryKey: activeThreadsQueryKey(this.context),
queryFn: ({ pageParam }) =>
this.runWithClient((client) => readThreadPage(client, this.context.vaultPath, { cursor: pageParam, archived: false })),
initialPageParam: null,
getNextPageParam: (lastPage) => lastPage.nextCursor ?? undefined,
staleTime: Number.POSITIVE_INFINITY,
};
}
private threadListQueryKey(kind: ThreadListKind): ReturnType<typeof activeThreadsQueryKey> | ReturnType<typeof archivedThreadsQueryKey> {
return kind === "archived" ? archivedThreadsQueryKey(this.context) : activeThreadsQueryKey(this.context);
private activeThreadListObserver(): InfiniteQueryObserver<
ThreadPage,
Error,
ActiveThreadListData,
ActiveThreadsQueryKey,
ActiveThreadPageParam
> {
return new InfiniteQueryObserver(this.client, {
...this.activeThreadListQueryOptions(),
enabled: false,
});
}
private activeThreadListIsFetchingNextPage(): boolean {
const observer = this.activeThreadListObserver();
const isFetchingNextPage = observer.getCurrentResult().isFetchingNextPage;
observer.destroy();
return isFetchingNextPage;
}
private archivedThreadListQueryOptions(): AppServerQueryOptions<readonly Thread[]> {
return {
queryKey: archivedThreadsQueryKey(this.context),
queryFn: () => this.runWithClient((client) => listThreads(client, this.context.vaultPath, { archived: true })).then(cloneThreads),
staleTime: Number.POSITIVE_INFINITY,
};
}
private runtimeConfigQueryOptions(): AppServerQueryOptions<RuntimeConfigSnapshot> {
@ -414,18 +460,14 @@ export class AppServerQueryCache {
this.runWithClient(async (client) =>
runtimeConfigSnapshotFromAppServerConfig(await readEffectiveConfig(client, this.context.vaultPath)),
),
staleTime: APP_SERVER_METADATA_STALE_TIME_MS,
};
}
private skillsQueryOptions(forceReload = false): AppServerQueryOptions<MetadataResourceSnapshot<readonly SkillMetadata[]>> {
private skillsQueryOptions(): AppServerQueryOptions<MetadataResourceSnapshot<readonly SkillMetadata[]>> {
return {
queryKey: appServerSkillsQueryKey(this.context),
queryFn: async () =>
this.runWithClient(async (client) =>
successfulMetadataResource(await readSkillMetadataProbe(client, this.context.vaultPath, forceReload)),
),
staleTime: APP_SERVER_METADATA_STALE_TIME_MS,
this.runWithClient(async (client) => successfulMetadataResource(await readSkillMetadataProbe(client, this.context.vaultPath))),
};
}
@ -436,7 +478,6 @@ export class AppServerQueryCache {
this.runWithClient(async (client) =>
successfulMetadataResource(await readPermissionProfileMetadataProbe(client, this.context.vaultPath)),
),
staleTime: APP_SERVER_METADATA_STALE_TIME_MS,
};
}
@ -444,92 +485,44 @@ export class AppServerQueryCache {
return {
queryKey: appServerRateLimitsQueryKey(this.context),
queryFn: async () => this.runWithClient(async (client) => successfulMetadataResource(await readRateLimitMetadataProbe(client))),
staleTime: APP_SERVER_METADATA_STALE_TIME_MS,
};
}
private async fetchRuntimeConfig(force: boolean): Promise<RuntimeConfigSnapshot> {
const key = appServerRuntimeConfigQueryKey(this.context);
if (force) {
await this.client.invalidateQueries({ queryKey: key });
this.assertUsable();
}
private async fetchRuntimeConfig(): Promise<RuntimeConfigSnapshot> {
const runtimeConfig = await this.client.fetchQuery(this.runtimeConfigQueryOptions());
this.assertUsable();
return cloneRuntimeConfigSnapshot(runtimeConfig);
}
private fetchMetadataResource(resource: MetadataResourceKind, force: boolean, reloadSkills = false): Promise<void> {
const refresh = (async (): Promise<void> => {
private fetchMetadataResource(resource: MetadataResourceKind): Promise<void> {
return (async (): Promise<void> => {
if (resource === "skills") {
const options = this.skillsQueryOptions(reloadSkills);
if (force) {
await this.client.invalidateQueries({ queryKey: options.queryKey });
this.assertUsable();
}
const options = this.skillsQueryOptions();
await this.client.fetchQuery(options);
this.assertUsable();
return;
}
if (resource === "permissionProfiles") {
const options = this.permissionProfilesQueryOptions();
if (force) {
await this.client.invalidateQueries({ queryKey: options.queryKey });
this.assertUsable();
}
await this.client.fetchQuery(options);
this.assertUsable();
return;
}
const options = this.rateLimitsQueryOptions();
if (force) {
await this.client.invalidateQueries({ queryKey: options.queryKey });
this.assertUsable();
}
await this.client.fetchQuery(options);
this.assertUsable();
})();
const active = this.metadataResourceFetches.get(resource) ?? new Set<Promise<void>>();
active.add(refresh);
this.metadataResourceFetches.set(resource, active);
const cleanup = (): void => {
active.delete(refresh);
if (active.size === 0 && this.metadataResourceFetches.get(resource) === active) this.metadataResourceFetches.delete(resource);
};
void refresh.then(cleanup, cleanup);
return refresh;
}
private refreshMetadataResourceNotification(resource: MetadataNotificationResourceKind): Promise<void> {
const generation = this.generation;
const current = this.metadataNotificationRefreshes.get(resource);
if (current?.generation === generation) {
current.dirty = true;
return current.promise;
private async refreshNotifiedMetadataResource(resource: "skills" | "rateLimits"): Promise<void> {
const queryKey = resource === "skills" ? appServerSkillsQueryKey(this.context) : appServerRateLimitsQueryKey(this.context);
await this.client.cancelQueries({ queryKey, exact: true });
this.assertUsable();
try {
await this.fetchMetadataResource(resource);
} catch (error) {
if (!(error instanceof CancelledError)) throw error;
}
const activeFetches = [...(this.metadataResourceFetches.get(resource) ?? [])];
const refresh: MetadataNotificationRefresh = {
dirty: false,
generation,
promise: Promise.resolve(),
};
refresh.promise = (async (): Promise<void> => {
if (activeFetches.length > 0) await Promise.allSettled(activeFetches);
if (generation !== this.generation) return;
for (;;) {
refresh.dirty = false;
await this.fetchMetadataResource(resource, true, resource === "skills").catch(() => undefined);
if (generation !== this.generation) return;
if (!metadataNotificationRefreshIsDirty(refresh)) return;
}
})();
this.metadataNotificationRefreshes.set(resource, refresh);
const cleanup = (): void => {
if (this.metadataNotificationRefreshes.get(resource) === refresh) this.metadataNotificationRefreshes.delete(resource);
};
void refresh.promise.then(cleanup, cleanup);
return refresh.promise;
}
private metadataResourceState(resource: "skills"): { value: readonly SkillMetadata[] | null; probe: DiagnosticProbeResult };
@ -581,27 +574,30 @@ export class AppServerQueryCache {
};
}
private observeQueryResult<T>(
queryOptions: AppServerQueryOptions<T>,
clone: (value: T) => T,
listener: ObservedResultListener<T>,
private observeQueryResult<TQuery, TValue>(
queryOptions: AppServerQueryOptions<TQuery>,
project: (value: TQuery) => TValue,
listener: ObservedResultListener<TValue>,
options: { emitCurrent?: boolean },
): () => void {
const observer = new QueryObserver<T>(this.client, {
const observer = new QueryObserver<TQuery>(this.client, {
...queryOptions,
enabled: false,
});
const emit = (result: QueryObserverResult<T>): void => {
listener(this.cloneObservedResult(result, clone));
const emit = (result: QueryObserverResult<TQuery>): void => {
listener(this.projectObservedResult(result, project));
};
const unsubscribe = observer.subscribe(emit);
if (options.emitCurrent ?? true) emit(observer.getCurrentResult());
return unsubscribe;
}
private cloneObservedResult<T>(result: QueryObserverResult<T>, clone: (value: T) => T): ObservedResult<T> {
private projectObservedResult<TQuery, TValue>(
result: QueryObserverResult<TQuery>,
project: (value: TQuery) => TValue,
): ObservedResult<TValue> {
return {
value: result.data === undefined ? null : clone(result.data),
value: result.data === undefined ? null : project(result.data),
error: result.error instanceof Error ? result.error : null,
isFetching: result.isFetching,
};
@ -609,40 +605,18 @@ export class AppServerQueryCache {
private runWithClient<T>(operation: (client: AppServerClient) => Promise<T>, options: AppServerClientAccessOptions = {}): Promise<T> {
this.assertUsable();
const generation = this.generation;
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).then(
(result) => {
if (this.disposed || generation !== this.generation) throw new DisposedAppServerQueryCacheError();
return result;
},
(error: unknown) => {
if (this.disposed || generation !== this.generation) throw new DisposedAppServerQueryCacheError();
throw error;
},
);
return runner.runWithClient(this.context, operation, options);
}
private assertUsable(): void {
if (this.disposed) throw new DisposedAppServerQueryCacheError();
if (this.disposed) throw new Error("Codex app-server query cache was disposed.");
}
private rememberActiveThreadCursor(cursor: string | null): void {
this.activeThreadCursorKnown = true;
this.activeThreadCursor = cursor;
this.bumpActiveThreadRevision();
}
private bumpActiveThreadRevision(): void {
this.activeThreadRevision += 1;
}
}
class DisposedAppServerQueryCacheError extends Error {
constructor() {
super("Codex app-server query cache was disposed.");
this.name = "DisposedAppServerQueryCacheError";
private activeThreadListData(): ActiveThreadListData | null {
if (!appServerQueryContextIsComplete(this.context)) return null;
return this.client.getQueryData<ActiveThreadListData>(activeThreadsQueryKey(this.context)) ?? null;
}
}
@ -662,15 +636,14 @@ function diagnosticProbeFromError(error: unknown): DiagnosticProbeResult | null
return error instanceof MetadataResourceQueryError ? error.probe : null;
}
function metadataNotificationRefreshIsDirty(refresh: MetadataNotificationRefresh): boolean {
return refresh.dirty;
function flattenActiveThreadList(data: ActiveThreadListData): readonly Thread[] {
return cloneThreads(data.pages.flatMap((page) => page.threads));
}
function createAppServerQueryClient(): QueryClient {
return new QueryClient({
defaultOptions: {
queries: {
gcTime: APP_SERVER_QUERY_GC_TIME_MS,
retry: false,
refetchOnReconnect: false,
refetchOnWindowFocus: false,

27
src/app-server/query/metadata-probes.ts
View file

@ -16,16 +16,9 @@ export type SkillMetadataProbeResult = MetadataProbeResult<SkillMetadata[], "ski
export type PermissionProfileMetadataProbeResult = MetadataProbeResult<RuntimePermissionProfileSummary[], "permissionProfiles">;
export type RateLimitMetadataProbeResult = MetadataProbeResult<RateLimitSnapshot | null, "rateLimits">;
export async function readSkillMetadataProbe(
client: AppServerRequestClient | null,
vaultPath: string,
forceReload = false,
): Promise<SkillMetadataProbeResult> {
if (!client) {
return { value: [], probe: diagnosticProbeError("skills", new Error("Codex app-server is not connected."), Date.now()) };
}
export async function readSkillMetadataProbe(client: AppServerRequestClient, vaultPath: string): Promise<SkillMetadataProbeResult> {
try {
const catalog = await listSkillCatalog(client, vaultPath, { forceReload });
const catalog = await listSkillCatalog(client, vaultPath);
return { value: catalog.skills, probe: diagnosticProbeOk("skills", `${String(catalog.totalCount)} skills`, Date.now()) };
} catch (error) {
return { value: [], probe: diagnosticProbeError("skills", error, Date.now()) };
@ -33,15 +26,9 @@ export async function readSkillMetadataProbe(
}
export async function readPermissionProfileMetadataProbe(
client: AppServerRequestClient | null,
client: AppServerRequestClient,
vaultPath: string,
): Promise<PermissionProfileMetadataProbeResult> {
if (!client) {
return {
value: [],
probe: diagnosticProbeError("permissionProfiles", new Error("Codex app-server is not connected."), Date.now()),
};
}
try {
const profiles = await listPermissionProfiles(client, vaultPath);
return { value: profiles, probe: diagnosticProbeOk("permissionProfiles", `${String(profiles.length)} profiles`, Date.now()) };
@ -50,13 +37,7 @@ export async function readPermissionProfileMetadataProbe(
}
}
export async function readRateLimitMetadataProbe(client: AppServerRequestClient | null): Promise<RateLimitMetadataProbeResult> {
if (!client) {
return {
value: null,
probe: diagnosticProbeError("rateLimits", new Error("Codex app-server is not connected."), Date.now()),
};
}
export async function readRateLimitMetadataProbe(client: AppServerRequestClient): Promise<RateLimitMetadataProbeResult> {
try {
const response = await readAccountRateLimits(client);
return {

7
src/app-server/query/thread-list-mutation.ts
View file

@ -9,14 +9,11 @@ export type ThreadListMutation =
readonly kind: "update";
readonly list: ThreadListKind;
readonly threadId: string;
readonly changes: Partial<Pick<Thread, "name" | "recencyAt">>;
}
| { readonly kind: "refresh"; readonly list: ThreadListKind };
readonly changes: Pick<Thread, "name">;
};
export function applyThreadListMutation(snapshot: readonly Thread[] | null, mutation: ThreadListMutation): readonly Thread[] | null {
switch (mutation.kind) {
case "refresh":
return snapshot;
case "upsert": {
if (!snapshot) return [mutation.thread];
const index = snapshot.findIndex((thread) => thread.id === mutation.thread.id);

4
src/features/chat/app-server/inbound/handler.ts
View file

@ -32,7 +32,6 @@ import { classifyAppServerLog } from "./app-server-logs";
import { type ChatNotificationEffect, planChatNotification } from "./notification-plan";
export interface ChatInboundHandlerActions {
refreshActiveThreads: () => void;
refreshServerDiagnostics: (options?: { forceResourceProbes?: boolean }) => void;
applyAppServerResourceEvent: (event: AppServerResourceEvent) => void;
maybeNameThread: (threadId: string, turnId: string, completedTurnTranscriptSummary: TurnTranscriptSummary | null) => void;
@ -249,9 +248,6 @@ function runNotificationEffect(
sourceContext: AppServerQueryContextIdentity,
): void {
switch (effect.type) {
case "refresh-threads":
context.actions.refreshActiveThreads();
return;
case "refresh-server-diagnostics":
context.actions.refreshServerDiagnostics({ forceResourceProbes: effect.forceResourceProbes === true });
return;

32
src/features/chat/app-server/inbound/notification-plan.ts
View file

@ -14,7 +14,6 @@ import { type DiagnosticStatusNotification, routeServerNotification, type Thread
import { turnRuntimeEventsFromNotification } from "./runtime-events";
export type ChatNotificationEffect =
| { type: "refresh-threads" }
| {
type: "maybe-name-thread";
threadId: string;
@ -25,7 +24,7 @@ export type ChatNotificationEffect =
| { type: "apply-app-server-resource-event"; event: AppServerResourceEvent }
| { type: "apply-thread-catalog-event"; event: ThreadCatalogEvent };
type TurnRuntimeCompletedTurnTranscriptSummary = Extract<TurnRuntimeOutcome, { type: "turn-completed" }>["completedTurnTranscriptSummary"];
type TurnRuntimeCompletedTurnTranscriptSummary = TurnRuntimeOutcome["completedTurnTranscriptSummary"];
export interface ChatNotificationPlan {
actions: readonly ChatAction[];
@ -242,25 +241,14 @@ function subagentTrackingActionsFromParentEvents(state: ChatState, events: reado
function chatNotificationEffectsFromTurnRuntimeOutcome(state: ChatState, outcome: TurnRuntimeOutcome): readonly ChatNotificationEffect[] {
if (activeThreadState(state)?.lifetime?.kind === "ephemeral") return [];
switch (outcome.type) {
case "turn-started":
return [
{
type: "apply-thread-catalog-event",
event: { type: "thread-touched", threadId: outcome.threadId, recencyAt: outcome.recencyAt },
},
];
case "turn-completed":
return [
{
type: "maybe-name-thread",
threadId: outcome.threadId,
turnId: outcome.turnId,
completedTurnTranscriptSummary: outcome.completedTurnTranscriptSummary,
},
{ type: "refresh-threads" },
];
}
return [
{
type: "maybe-name-thread",
threadId: outcome.threadId,
turnId: outcome.turnId,
completedTurnTranscriptSummary: outcome.completedTurnTranscriptSummary,
},
];
}
function planDiagnosticStatus(notification: DiagnosticStatusNotification): ChatNotificationPlan {
@ -346,7 +334,7 @@ function threadStartedPlan(
: [
{
type: "apply-thread-catalog-event",
event: { type: "thread-started", thread },
event: { type: "thread-upserted", thread },
},
];
if (activeThreadId(state) === notification.params.thread.id) {

1
src/features/chat/app-server/inbound/runtime-events.ts
View file

@ -126,7 +126,6 @@ export function turnRuntimeEventsFromNotification(
type: "turnStarted",
threadId: notification.params.threadId,
turnId: notification.params.turn.id,
recencyAt: notification.params.turn.startedAt,
},
];
case "turn/completed":

3
src/features/chat/application/threads/resume-actions.ts
View file

@ -1,4 +1,5 @@
import type { ThreadTokenUsage } from "../../../../domain/runtime/metrics";
import type { Thread } from "../../../../domain/threads/model";
import { effectCompletedInCurrentContext } from "../effect-outcome";
import { resumedThreadAction } from "../state/actions";
import { capturePanelTargetLease, type PanelTargetLease, panelTargetLeaseIsCurrent } from "../state/panel-target";
@ -18,6 +19,7 @@ export interface ResumeActionsHost {
closing: () => boolean;
resetThreadTurnPresence: (hadTurns: boolean) => void;
notifyActiveThreadIdentityChanged: () => void;
recordResumedThread: (thread: Thread) => void;
addSystemMessage: (text: string) => void;
refreshLiveState: () => void;
syncThreadGoal: (threadId: string) => Promise<void>;
@ -63,6 +65,7 @@ async function resumeThread(
const adoptedPanelTarget = applyResumedThread(host, effect.value, initialPanelTarget.revision);
if (!adoptedPanelTarget) return false;
currentPanelTarget = adoptedPanelTarget;
host.recordResumedThread(effect.value.activation.thread);
options?.onAdopted?.();
recoverResumedThreadTokenUsage(host, effect.value.activation.thread.id, effect.value.rolloutPath, resume, adoptedPanelTarget);
if (effect.value.initialHistoryPage) {

17
src/features/chat/application/threads/thread-management-actions.ts
View file

@ -25,8 +25,7 @@ export interface ThreadManagementActionsHost {
openThreadInNewView: (threadId: string) => Promise<void>;
openThreadInCurrentPanel: (threadId: string) => Promise<void>;
notifyActiveThreadIdentityChanged: () => void;
refreshAfterThreadMutation: () => Promise<void>;
recordForkedThread: (thread: Thread) => void;
recordThread: (thread: Thread) => void;
}
interface ThreadManagementOperations {
@ -138,7 +137,7 @@ async function forkThreadFromTurn(
if (!effectCompleted(effect)) return;
const forkedThread = effect.value;
const forkedThreadId = forkedThread.id;
host.recordForkedThread(forkedThread);
host.recordThread(forkedThread);
if (!effectCompletedInCurrentContext(effect)) return;
if (!threadManagementScopeStillTargetsOriginalPanel(host, scope)) return;
if (sourceName) {
@ -207,15 +206,9 @@ async function rollbackThread(host: ThreadManagementActionsHost, threadId: strin
if (!threadManagementScopeStillTargetsPanel(host, scope)) return;
const effect = await host.threadTransport.rollbackThread(threadId);
if (!effectCompleted(effect)) return;
if (!effectCompletedInCurrentContext(effect)) {
await host.refreshAfterThreadMutation();
return;
}
if (!effectCompletedInCurrentContext(effect)) return;
const snapshot = effect.value;
if (!threadManagementScopeStillTargetsPanel(host, scope)) {
await host.refreshAfterThreadMutation();
return;
}
if (!threadManagementScopeStillTargetsPanel(host, scope)) return;
threadManagementDispatch(
host,
resumedThreadActionFromActiveRuntime({
@ -236,7 +229,7 @@ async function rollbackThread(host: ThreadManagementActionsHost, threadId: strin
host.addSystemMessage("Rolled back the latest turn. Local file changes were not reverted.");
host.setStatus(STATUS_ROLLBACK_COMPLETE);
host.notifyActiveThreadIdentityChanged();
await host.refreshAfterThreadMutation();
host.recordThread(snapshot.thread);
} catch (error) {
if (!threadManagementScopeStillTargetsPanel(host, scope)) return;
host.addSystemMessage(error instanceof Error ? error.message : String(error));

23
src/features/chat/application/turns/runtime-event-plan.ts
View file

@ -6,14 +6,12 @@ import { threadStreamItems } from "../state/thread-stream";
import type { TurnRuntimeEvent } from "./runtime-events";
import { activeTurnId, pendingTurnStart as pendingTurnStartForState } from "./turn-state";
export type TurnRuntimeOutcome =
| { type: "turn-started"; threadId: string; turnId: string; recencyAt: number | null }
| {
type: "turn-completed";
threadId: string;
turnId: string;
completedTurnTranscriptSummary: TurnRuntimeEventCompletedTurnTranscriptSummary;
};
export interface TurnRuntimeOutcome {
type: "turn-completed";
threadId: string;
turnId: string;
completedTurnTranscriptSummary: TurnRuntimeEventCompletedTurnTranscriptSummary;
}
type TurnRuntimeEventCompletedTurnTranscriptSummary = Extract<
TurnRuntimeEvent,
@ -116,14 +114,7 @@ function turnStartedPlan(state: ChatState, event: Extract<TurnRuntimeEvent, { ty
items: threadStreamItemsWithPendingPromptSubmitHooks(state, event.turnId),
},
],
outcomes: [
{
type: "turn-started",
threadId: event.threadId,
turnId: event.turnId,
recencyAt: event.recencyAt,
},
],
outcomes: [],
};
}

1
src/features/chat/application/turns/runtime-events.ts
View file

@ -59,7 +59,6 @@ export type TurnRuntimeEvent =
type: "turnStarted";
threadId: string;
turnId: string;
recencyAt: number | null;
}
| {
type: "turnCompleted";

9
src/features/chat/host/bundles/connection-bundle.ts
View file

@ -140,18 +140,9 @@ export function createConnectionBundle(
const threads = await environment.plugin.threadCatalog.refreshActive();
stateStore.dispatch({ type: "thread-list/applied", threads });
};
const refreshSharedThreadsQuietly = (): void => {
void refreshSharedThreads().catch((error: unknown) => {
if (isStaleAppServerResourceContextError(error)) return;
status.addSystemMessage(error instanceof Error ? error.message : String(error));
});
};
const inboundHandler = createChatInboundHandler(
stateStore,
{
refreshActiveThreads: () => {
refreshSharedThreadsQuietly();
},
refreshServerDiagnostics: (options) => {
void serverDiagnostics.refreshServerDiagnostics(options).catch((error: unknown) => {
status.addSystemMessage(error instanceof Error ? error.message : String(error));

13
src/features/chat/host/bundles/thread-bundle.ts
View file

@ -104,7 +104,6 @@ interface ChatPanelThreadActionInput {
composerController: ChatComposerController;
foundation: ChatPanelThreadFoundation;
lifecycle: ChatPanelThreadLifecycleBundle;
refreshActiveThreads: () => Promise<void>;
notifyActiveThreadIdentityChanged: () => void;
navigation: PersistentNavigationLifecycle;
}
@ -279,7 +278,7 @@ export function createThreadLifecycleBundle(
}
export function createThreadActionBundle(host: ChatPanelThreadHost, input: ChatPanelThreadActionInput): ChatPanelThreadActionBundle {
const { appServer, status, composerController, foundation, lifecycle, refreshActiveThreads, notifyActiveThreadIdentityChanged } = input;
const { appServer, status, composerController, foundation, lifecycle, notifyActiveThreadIdentityChanged } = input;
const { environment, stateStore } = host;
const threadManagementHost: ThreadManagementActionsHost = {
stateStore,
@ -301,11 +300,8 @@ export function createThreadActionBundle(host: ChatPanelThreadHost, input: ChatP
await lifecycle.resume.resumeThread(threadId);
},
notifyActiveThreadIdentityChanged,
refreshAfterThreadMutation: async () => {
await refreshActiveThreads();
},
recordForkedThread: (thread) => {
environment.plugin.threadCatalog.apply({ type: "thread-forked", thread });
recordThread: (thread) => {
environment.plugin.threadCatalog.apply({ type: "thread-upserted", thread });
},
};
const actions = createThreadManagementActions(threadManagementHost);
@ -368,6 +364,9 @@ function createSessionThreadLifecycle(
closing: host.getClosing,
resetThreadTurnPresence,
notifyActiveThreadIdentityChanged,
recordResumedThread: (thread) => {
host.environment.plugin.threadCatalog.apply({ type: "thread-upserted", thread });
},
addSystemMessage: status.addSystemMessage,
refreshLiveState,
syncThreadGoal: (threadId) => goals.syncThreadGoal(threadId),

4
src/features/chat/host/session-runtime.ts
View file

@ -184,7 +184,6 @@ function composeChatPanelSessionRuntime(host: ChatPanelSessionRuntimeHost): Chat
return currentClient();
},
});
const refreshActiveThreads = () => connectionActions.refreshActiveThreads();
const runtime = createRuntimeBundle(host, {
connection,
appServer,
@ -195,7 +194,7 @@ function composeChatPanelSessionRuntime(host: ChatPanelSessionRuntimeHost): Chat
threadStartTransport: appServer.threadStart,
runtimeSnapshotForState: runtimeSnapshotForChatState,
recordStartedThread: (thread) => {
environment.plugin.threadCatalog.apply({ type: "thread-started", thread });
environment.plugin.threadCatalog.apply({ type: "thread-upserted", thread });
},
syncThreadGoal: (threadId) => {
void threadFoundation.goalSync.syncThreadGoal(threadId);
@ -237,7 +236,6 @@ function composeChatPanelSessionRuntime(host: ChatPanelSessionRuntimeHost): Chat
composerController,
foundation: threadFoundation,
lifecycle: threadLifecycle,
refreshActiveThreads,
notifyActiveThreadIdentityChanged,
navigation,
});

7
src/features/thread-picker/modal.obsidian.ts
View file

@ -22,12 +22,13 @@ const THREAD_PICKER_MODIFIER_ENTER_LISTENER_OPTIONS = { capture: true } as const
export async function openThreadPicker(host: ThreadPickerHost): Promise<void> {
try {
const threads = await host.threadCatalog.refreshActive();
const snapshot = host.threadCatalog.activeSnapshot();
const threads = snapshot ?? (await host.threadCatalog.loadActive());
if (threads.length === 0) {
new Notice("No Codex threads found.");
return;
}
new ThreadPickerModal(host, threads).open();
new ThreadPickerModal(host, threads, snapshot === null).open();
} catch (error) {
new Notice(error instanceof Error ? error.message : String(error));
}
@ -48,11 +49,11 @@ function threadOpenModeFromEvent(evt: MouseEvent | KeyboardEvent): ThreadOpenMod
class ThreadPickerModal extends SuggestModal<ThreadSuggestion> {
private threads: readonly Thread[];
private completeThreadsPromise: Promise<readonly Thread[]> | null = null;
private hasCompleteThreadList = false;
constructor(
private readonly host: ThreadPickerHost,
threads: readonly Thread[],
private hasCompleteThreadList: boolean,
) {
super(host.app);
this.threads = threads;

24
src/features/threads/catalog/thread-catalog.ts
View file

@ -28,9 +28,7 @@ export interface ThreadCatalogOptions {
}
export type ThreadCatalogEvent =
| { type: "thread-started"; thread: Thread }
| { type: "thread-forked"; thread: Thread }
| { type: "thread-touched"; threadId: string; recencyAt?: number | null }
| { type: "thread-upserted"; thread: Thread }
| { type: "thread-renamed"; threadId: string; name: string | null }
| { type: "thread-archived"; threadId: string }
| { type: "thread-deleted"; threadId: string }
@ -96,18 +94,8 @@ export function createThreadCatalog(options: ThreadCatalogOptions): ThreadCatalo
function threadListMutationsForEvent(store: ThreadCatalogStore, event: ThreadCatalogEvent): ThreadListMutation[] {
switch (event.type) {
case "thread-started":
case "thread-forked":
case "thread-upserted":
return [{ kind: "upsert", list: "active", thread: { ...event.thread, archived: false } }];
case "thread-touched":
return [
{
kind: "update",
list: "active",
threadId: event.threadId,
changes: event.recencyAt === undefined ? {} : { recencyAt: event.recencyAt },
},
];
case "thread-renamed":
return [
{ kind: "update", list: "active", threadId: event.threadId, changes: { name: event.name } },
@ -117,9 +105,7 @@ function threadListMutationsForEvent(store: ThreadCatalogStore, event: ThreadCat
const thread = threadById(store.activeThreadsSnapshot(), event.threadId);
return [
{ kind: "remove", list: "active", threadId: event.threadId },
...(thread
? [{ kind: "upsert", list: "archived", thread: { ...thread, archived: true } } satisfies ThreadListMutation]
: [{ kind: "refresh", list: "archived" } satisfies ThreadListMutation]),
...(thread ? [{ kind: "upsert", list: "archived", thread: { ...thread, archived: true } } satisfies ThreadListMutation] : []),
];
}
case "thread-deleted":
@ -136,9 +122,7 @@ function threadListMutationsForEvent(store: ThreadCatalogStore, event: ThreadCat
const thread = threadById(store.archivedThreadsSnapshot(), event.threadId);
return [
{ kind: "remove", list: "archived", threadId: event.threadId },
...(thread
? [{ kind: "upsert", list: "active", thread: { ...thread, archived: false } } satisfies ThreadListMutation]
: [{ kind: "refresh", list: "active" } satisfies ThreadListMutation]),
...(thread ? [{ kind: "upsert", list: "active", thread: { ...thread, archived: false } } satisfies ThreadListMutation] : []),
];
}
}

269
tests/app-server/query-cache.test.ts
View file

@ -7,25 +7,11 @@ import type { RuntimePermissionProfileSummary } from "../../src/domain/runtime/p
import type { Thread } from "../../src/domain/threads/model";
describe("AppServerQueryCache", () => {
it("garbage-collects inactive query records", async () => {
vi.useFakeTimers();
try {
const cache = cacheWithThreads(() => Promise.resolve([thread("temporary")]));
await cache.fetchActiveThreads();
await vi.advanceTimersByTimeAsync(300_001);
expect(cache.activeThreadsSnapshot()).toBeNull();
} finally {
vi.useRealTimers();
}
});
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.fetchActiveThreads()).resolves.toEqual([]);
await expect(cache.refreshActiveThreads()).resolves.toEqual([]);
expect(cache.activeThreadsSnapshot()).toBeNull();
expect(cache.appServerMetadataSnapshot()).toBeNull();
expect(cache.modelsSnapshot()).toBeNull();
@ -105,6 +91,27 @@ describe("AppServerQueryCache", () => {
});
});
it("shares concurrent Load more requests through the InfiniteQuery", async () => {
const nextPage = deferred<{ data: ReturnType<typeof thread>[]; nextCursor: null }>();
const listThreads = vi
.fn()
.mockResolvedValueOnce({ data: [thread("first")], nextCursor: "page-2" })
.mockImplementationOnce(() => nextPage.promise);
const cache = cacheWithRequestHandlers({ "thread/list": listThreads });
await cache.refreshActiveThreads();
const first = cache.loadMoreActiveThreads();
const second = cache.loadMoreActiveThreads();
await flushMicrotasks();
expect(listThreads).toHaveBeenCalledTimes(2);
nextPage.resolve({ data: [thread("second")], nextCursor: null });
await expect(Promise.all([first, second])).resolves.toEqual([
[thread("first"), thread("second")],
[thread("first"), thread("second")],
]);
});
it("does not append a stale load-more page after a newer first-page refresh", async () => {
const oldPage = deferred<{ data: ReturnType<typeof thread>[]; nextCursor: string | null }>();
const listThreads = vi
@ -116,23 +123,98 @@ describe("AppServerQueryCache", () => {
await cache.refreshActiveThreads();
const loadMore = cache.loadMoreActiveThreads();
await Promise.resolve();
await flushMicrotasks();
await cache.refreshActiveThreads();
oldPage.resolve({ data: [thread("old-second")], nextCursor: null });
await expect(loadMore).resolves.toEqual([thread("new-first")]);
await expect(loadMore).resolves.toEqual([thread("old-first")]);
expect(cache.activeThreadsSnapshot()).toEqual([thread("new-first")]);
expect(cache.hasMoreActiveThreads()).toBe(true);
});
it("rejects only a thread read started before an event and accepts the following server value", async () => {
it("does not append a load-more page invalidated by an exact event", async () => {
const oldPage = deferred<{ data: ReturnType<typeof thread>[]; nextCursor: null }>();
const listThreads = vi
.fn()
.mockResolvedValueOnce({ data: [thread("first")], nextCursor: "page-2" })
.mockImplementationOnce(() => oldPage.promise)
.mockResolvedValueOnce({ data: [thread("first")], nextCursor: "page-2" });
const cache = cacheWithRequestHandlers({ "thread/list": listThreads });
await cache.refreshActiveThreads();
const loadMore = cache.loadMoreActiveThreads();
await flushMicrotasks();
cache.applyThreadListMutations([{ kind: "remove", list: "active", threadId: "deleted-on-page-2" }]);
oldPage.resolve({ data: [thread("deleted-on-page-2")], nextCursor: null });
await expect(loadMore).resolves.toEqual([thread("first")]);
expect(cache.activeThreadsSnapshot()).toEqual([thread("first")]);
expect(cache.hasMoreActiveThreads()).toBe(true);
});
it("projects an exact event without preserving a synthetic page boundary", async () => {
const listThreads = vi
.fn()
.mockResolvedValueOnce({ data: [thread("old-first"), thread("old-second")], nextCursor: "page-2" })
.mockResolvedValueOnce({ data: [thread("new-first"), thread("old-first")], nextCursor: "page-2" });
const cache = cacheWithRequestHandlers({ "thread/list": listThreads });
await cache.refreshActiveThreads();
cache.applyThreadListMutations([{ kind: "upsert", list: "active", thread: thread("new-first") }]);
expect(cache.activeThreadsSnapshot()?.map((item) => item.id)).toEqual(["new-first", "old-first", "old-second"]);
expect(listThreads).toHaveBeenCalledOnce();
await cache.refreshActiveThreads();
expect(cache.activeThreadsSnapshot()?.map((item) => item.id)).toEqual(["new-first", "old-first"]);
expect(cache.hasMoreActiveThreads()).toBe(true);
});
it("continues the existing cursor chain after an exact event", async () => {
const listThreads = vi
.fn()
.mockResolvedValueOnce({ data: [thread("old-first")], nextCursor: "old-page-2" })
.mockResolvedValueOnce({ data: [thread("old-second")], nextCursor: null });
const cache = cacheWithRequestHandlers({ "thread/list": listThreads });
await cache.refreshActiveThreads();
cache.applyThreadListMutations([{ kind: "upsert", list: "active", thread: thread("event-thread") }]);
await expect(cache.loadMoreActiveThreads()).resolves.toEqual([thread("event-thread"), thread("old-first"), thread("old-second")]);
expect(listThreads).toHaveBeenNthCalledWith(2, {
cwd: "/vault",
cursor: "old-page-2",
archived: false,
limit: 100,
sortKey: "recency_at",
sortDirection: "desc",
});
});
it("updates a loaded thread in place instead of inventing a new rank", async () => {
const listThreads = vi
.fn()
.mockResolvedValueOnce({ data: [thread("first")], nextCursor: "page-2" })
.mockResolvedValueOnce({ data: [thread("second")], nextCursor: null });
const cache = cacheWithRequestHandlers({ "thread/list": listThreads });
await cache.refreshActiveThreads();
await cache.loadMoreActiveThreads();
cache.applyThreadListMutations([
{ kind: "upsert", list: "active", thread: { ...thread("second"), name: "Updated without re-ranking" } },
]);
expect(cache.activeThreadsSnapshot()).toEqual([thread("first"), { ...thread("second"), name: "Updated without re-ranking" }]);
expect(listThreads).toHaveBeenCalledTimes(2);
});
it("cancels a stale thread read before applying an exact event", async () => {
const staleRead = deferred<{ data: ReturnType<typeof thread>[]; nextCursor: null }>();
const postEventRead = deferred<{ data: ReturnType<typeof thread>[]; nextCursor: null }>();
const listThreads = vi
.fn()
.mockResolvedValueOnce({ data: [{ ...thread("target"), name: "initial" }], nextCursor: null })
.mockImplementationOnce(() => staleRead.promise)
.mockImplementationOnce(() => postEventRead.promise);
.mockResolvedValueOnce({ data: [{ ...thread("target"), name: "authoritative" }], nextCursor: null });
const cache = cacheWithRequestHandlers({ "thread/list": listThreads });
await cache.refreshActiveThreads();
@ -142,23 +224,36 @@ describe("AppServerQueryCache", () => {
expect(cache.activeThreadsSnapshot()?.[0]?.name).toBe("from-event");
staleRead.resolve({ data: [{ ...thread("target"), name: "stale" }], nextCursor: null });
await expect(refresh).resolves.toEqual([{ ...thread("target"), name: "from-event" }]);
await vi.waitFor(() => expect(listThreads).toHaveBeenCalledTimes(3));
expect(cache.activeThreadsSnapshot()?.[0]?.name).toBe("from-event");
postEventRead.resolve({ data: [{ ...thread("target"), name: "authoritative" }], nextCursor: null });
await refresh;
expect(cache.activeThreadsSnapshot()?.[0]?.name).toBe("authoritative");
expect(listThreads).toHaveBeenCalledTimes(3);
});
it("retries a full active-thread inventory when a first-page refresh wins the race", async () => {
it("restarts an initial thread read when an exact event arrives before any snapshot", async () => {
const staleRead = deferred<{ data: ReturnType<typeof thread>[]; nextCursor: null }>();
const listThreads = vi
.fn()
.mockImplementationOnce(() => staleRead.promise)
.mockResolvedValueOnce({ data: [{ ...thread("target"), name: "authoritative" }], nextCursor: null });
const cache = cacheWithRequestHandlers({ "thread/list": listThreads });
const initial = cache.refreshActiveThreads();
await flushMicrotasks();
cache.applyThreadListMutations([{ kind: "update", list: "active", threadId: "target", changes: { name: "from-event" } }]);
staleRead.resolve({ data: [{ ...thread("target"), name: "stale" }], nextCursor: null });
await expect(initial).resolves.toEqual([{ ...thread("target"), name: "authoritative" }]);
expect(cache.activeThreadsSnapshot()?.[0]?.name).toBe("authoritative");
expect(listThreads).toHaveBeenCalledTimes(2);
});
it("keeps a thread-picker inventory read out of the shared recent list", async () => {
const oldInventory = deferred<{ data: ReturnType<typeof thread>[]; nextCursor: string | null }>();
const listThreads = vi
.fn()
.mockImplementationOnce(() => oldInventory.promise)
.mockResolvedValueOnce({ data: [thread("new-first")], nextCursor: "new-page-2" })
.mockResolvedValueOnce({ data: [thread("new-first"), thread("new-second")], nextCursor: null });
.mockResolvedValueOnce({ data: [thread("new-first")], nextCursor: "new-page-2" });
const cache = cacheWithRequestHandlers({ "thread/list": listThreads });
const inventory = cache.fetchAllActiveThreads();
@ -166,35 +261,20 @@ describe("AppServerQueryCache", () => {
await cache.refreshActiveThreads();
oldInventory.resolve({ data: [thread("old")], nextCursor: null });
await expect(inventory).resolves.toEqual([thread("new-first"), thread("new-second")]);
expect(cache.activeThreadsSnapshot()).toEqual([thread("new-first"), thread("new-second")]);
await expect(inventory).resolves.toEqual([thread("old")]);
expect(cache.activeThreadsSnapshot()).toEqual([thread("new-first")]);
expect(cache.hasMoreActiveThreads()).toBe(true);
});
it("does not revive a disposed cache when an active-thread inventory resolves", async () => {
const inventory = deferred<{ data: ReturnType<typeof thread>[]; nextCursor: string | null }>();
const listThreads = vi.fn(() => inventory.promise);
const cache = cacheWithRequestHandlers({ "thread/list": listThreads });
const fetch = cache.fetchAllActiveThreads();
await flushMicrotasks();
cache.dispose();
inventory.resolve({ data: [thread("stale")], nextCursor: null });
await expect(fetch).rejects.toThrow("query cache was disposed");
expect(cache.activeThreadsSnapshot()).toBeNull();
expect(listThreads).toHaveBeenCalledOnce();
});
it("rejects a cached result when disposal wins before its continuation", async () => {
it("clears snapshots and rejects new reads after disposal", async () => {
const listModels = vi.fn().mockResolvedValue({ data: [catalogModel("cached")] });
const cache = cacheWithRequestHandlers({ "model/list": listModels });
await cache.fetchModels();
const cached = cache.fetchModels();
cache.dispose();
await expect(cached).rejects.toThrow("query cache was disposed");
expect(cache.modelsSnapshot()).toBeNull();
await expect(cache.fetchModels()).rejects.toThrow("query cache was disposed");
expect(listModels).toHaveBeenCalledOnce();
});
@ -259,7 +339,7 @@ describe("AppServerQueryCache", () => {
unsubscribe();
});
it("publishes metadata resources only after a retry settles", async () => {
it("publishes metadata resources only after a successful retry settles", async () => {
const retry = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>();
const nextRetry = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>();
const listSkills = vi
@ -268,7 +348,7 @@ describe("AppServerQueryCache", () => {
.mockImplementationOnce(() => retry.promise)
.mockImplementationOnce(() => nextRetry.promise);
const cache = cacheWithRequestHandlers({ "skills/list": listSkills });
await cache.refreshSkills();
await expect(cache.refreshSkills()).rejects.toThrow("skills offline");
const listener = vi.fn();
const unsubscribe = cache.observeAppServerMetadataResources(listener, { emitCurrent: false });
@ -301,47 +381,6 @@ describe("AppServerQueryCache", () => {
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 refreshes = Array.from({ length: 10 }, () => cache.refreshSkills());
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<typeof appServerRateLimit>; rateLimitsByLimitId: null }>();
const second = deferred<{ rateLimits: ReturnType<typeof appServerRateLimit>; rateLimitsByLimitId: null }>();
const readRateLimits = vi
.fn()
.mockImplementationOnce(() => first.promise)
.mockImplementationOnce(() => second.promise);
const cache = cacheWithRequestHandlers({ "account/rateLimits/read": readRateLimits });
const refreshes = Array.from({ length: 10 }, () => cache.refreshRateLimits());
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<Record<string, never>>();
const models = deferred<{ data: CatalogModel[] }>();
@ -370,7 +409,7 @@ describe("AppServerQueryCache", () => {
await Promise.all([first, second]);
});
it("runs a forced skills request after a notification overlaps a full refresh", async () => {
it("uses the regular skills query when a notification overlaps a full refresh", async () => {
const first = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>();
const second = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>();
const listSkills = vi
@ -390,51 +429,35 @@ describe("AppServerQueryCache", () => {
const notificationRefresh = cache.refreshSkills();
first.resolve({ data: [{ skills: [catalogSkill("old")] }] });
await vi.waitFor(() => expect(listSkills).toHaveBeenCalledTimes(2));
expect(listSkills).toHaveBeenNthCalledWith(2, { cwds: ["/vault"], forceReload: true });
expect(listSkills).toHaveBeenNthCalledWith(2, { cwds: ["/vault"], forceReload: false });
second.resolve({ data: [{ skills: [catalogSkill("new")] }] });
await Promise.all([fullRefresh, notificationRefresh]);
expect(cache.appServerMetadataSnapshot()?.availableSkills.map((skill) => skill.name)).toEqual(["new"]);
});
it("schedules one more skills read when a notification arrives during the trailing refresh", async () => {
it("does not surface cancellation when a newer skills notification replaces an older 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);
.mockResolvedValueOnce({ data: [{ skills: [catalogSkill("latest")] }] });
const cache = cacheWithRequestHandlers({ "skills/list": listSkills });
const listener = vi.fn();
const unsubscribe = cache.observeAppServerMetadataResources(listener, { emitCurrent: false });
const leading = cache.refreshSkills();
const trailing = cache.refreshSkills();
first.resolve({ data: [{ skills: [catalogSkill("first")] }] });
await vi.waitFor(() => expect(listSkills).toHaveBeenCalledTimes(2));
const afterTrailing = cache.refreshSkills();
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 disposed", async () => {
const first = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>();
const listSkills = vi.fn(() => first.promise);
const cache = cacheWithRequestHandlers({ "skills/list": listSkills });
const leading = cache.refreshSkills();
const trailing = cache.refreshSkills();
const older = cache.refreshSkills();
await flushMicrotasks();
cache.dispose();
const newer = cache.refreshSkills();
first.resolve({ data: [{ skills: [catalogSkill("stale")] }] });
await expect(Promise.all([leading, trailing])).rejects.toThrow("query cache was disposed");
await flushMicrotasks();
expect(listSkills).toHaveBeenCalledOnce();
expect(cache.appServerMetadataSnapshot()).toBeNull();
await expect(Promise.all([older, newer])).resolves.toEqual([undefined, undefined]);
expect(listener).toHaveBeenLastCalledWith({
id: "skills",
value: [expect.objectContaining({ name: "latest" })],
probe: expect.objectContaining({ status: "ok" }),
});
expect(listSkills).toHaveBeenCalledTimes(2);
unsubscribe();
});
it("rejects an initial metadata refresh when runtime config fails after optional resources settle", async () => {

30
tests/features/chat/app-server/inbound/handler.test.ts
View file

@ -34,7 +34,6 @@ function handlerForState(state = chatStateFixture(), actions: Partial<ChatInboun
const handler = createChatInboundHandler(
store,
{
refreshActiveThreads: vi.fn(),
refreshServerDiagnostics: vi.fn(),
applyAppServerResourceEvent: vi.fn(),
maybeNameThread: vi.fn(),
@ -368,9 +367,8 @@ describe("ChatInboundHandler", () => {
status: "completed",
},
]);
const refreshActiveThreads = vi.fn();
const applyThreadCatalogEvent = vi.fn();
const handler = handlerForState(state, { refreshActiveThreads, applyThreadCatalogEvent });
const handler = handlerForState(state, { applyThreadCatalogEvent });
handler.handleNotification({
method: "turn/started",
@ -392,15 +390,7 @@ describe("ChatInboundHandler", () => {
expect(chatStateThreadStreamItems(handler.currentState()).map((item) => item.id)).toEqual(["local-user-1", "hook-hook-1-1"]);
expect(chatStateThreadStreamItems(handler.currentState())[1]).toMatchObject({ id: "hook-hook-1-1", turnId: "turn-active" });
expect(pendingTurnStart(handler.currentState())).toBeNull();
expect(applyThreadCatalogEvent).toHaveBeenCalledWith(
{
type: "thread-touched",
threadId: "thread-active",
recencyAt: 1,
} satisfies ThreadCatalogEvent,
TEST_APP_SERVER_CONTEXT,
);
expect(refreshActiveThreads).not.toHaveBeenCalled();
expect(applyThreadCatalogEvent).not.toHaveBeenCalled();
});
it("captures only prompt-submit hooks observed during the pending turn start", () => {
@ -538,8 +528,7 @@ describe("ChatInboundHandler", () => {
},
]);
const maybeNameThread = vi.fn();
const refreshActiveThreads = vi.fn();
const handler = handlerForState(state, { maybeNameThread, refreshActiveThreads });
const handler = handlerForState(state, { maybeNameThread });
handler.handleNotification({
method: "turn/completed",
@ -564,7 +553,6 @@ describe("ChatInboundHandler", () => {
});
expect(chatStateThreadStreamItems(handler.currentState()).map((item) => item.id)).toEqual(["local-user-1", "hook-hook-1-1"]);
expect(maybeNameThread).not.toHaveBeenCalled();
expect(refreshActiveThreads).not.toHaveBeenCalled();
});
it("routes sparse account rate limit updates through the app-server resource event boundary", () => {
@ -1305,8 +1293,7 @@ describe("ChatInboundHandler", () => {
it("records deleted thread notifications in the active catalog", () => {
const applyThreadCatalogEvent = vi.fn();
const refreshActiveThreads = vi.fn();
const handler = handlerForState(chatStateFixture(), { applyThreadCatalogEvent, refreshActiveThreads });
const handler = handlerForState(chatStateFixture(), { applyThreadCatalogEvent });
handler.handleNotification({
method: "thread/deleted",
@ -1320,13 +1307,11 @@ describe("ChatInboundHandler", () => {
} satisfies ThreadCatalogEvent,
TEST_APP_SERVER_CONTEXT,
);
expect(refreshActiveThreads).not.toHaveBeenCalled();
});
it("routes unarchived thread notifications through the catalog instead of refreshing in the handler", () => {
const applyThreadCatalogEvent = vi.fn();
const refreshActiveThreads = vi.fn();
const handler = handlerForState(chatStateFixture(), { applyThreadCatalogEvent, refreshActiveThreads });
const handler = handlerForState(chatStateFixture(), { applyThreadCatalogEvent });
handler.handleNotification({
method: "thread/unarchived",
@ -1340,7 +1325,6 @@ describe("ChatInboundHandler", () => {
} satisfies ThreadCatalogEvent,
TEST_APP_SERVER_CONTEXT,
);
expect(refreshActiveThreads).not.toHaveBeenCalled();
});
it("records unrelated thread-started notifications without replacing the active cwd", () => {
@ -1358,7 +1342,7 @@ describe("ChatInboundHandler", () => {
expect(activeThreadState(handler.currentState())?.cwd).toBe("/workspace/active");
expect(applyThreadCatalogEvent).toHaveBeenCalledWith(
{
type: "thread-started",
type: "thread-upserted",
thread: expect.objectContaining({ id: "thread-other" }),
},
TEST_APP_SERVER_CONTEXT,
@ -1379,7 +1363,7 @@ describe("ChatInboundHandler", () => {
expect(activeThreadState(handler.currentState())?.cwd).toBe("/workspace/active");
expect(applyThreadCatalogEvent).toHaveBeenCalledWith(
{
type: "thread-started",
type: "thread-upserted",
thread: expect.objectContaining({ id: "thread-active" }),
},
TEST_APP_SERVER_CONTEXT,

10
tests/features/chat/app-server/inbound/routing.test.ts
View file

@ -70,19 +70,14 @@ describe("chat inbound routing", () => {
expectNotificationRouteKind(notification, "threadLifecycle", { activeThreadId: "thread-other", activeTurnId: "turn-active" });
});
it("translates turn-started runtime outcomes to thread catalog events at the inbound boundary", () => {
it("does not turn live turn-start state into thread catalog work", () => {
let state = chatStateFixture();
state = chatStateWith(state, { activeThread: { id: "thread-active" } });
state = chatStateWith(state, { turn: { lifecycle: { kind: "running", turnId: "turn-active" } } });
const plan = planChatNotification(state, turnStartedNotification(), (prefix) => `${prefix}-1`);
expect(plan.effects).toEqual([
{
type: "apply-thread-catalog-event",
event: { type: "thread-touched", threadId: "thread-active", recencyAt: null },
},
]);
expect(plan.effects).toEqual([]);
});
it("translates turn-completed runtime outcomes to thread follow-up effects at the inbound boundary", () => {
@ -99,7 +94,6 @@ describe("chat inbound routing", () => {
turnId: "turn-active",
completedTurnTranscriptSummary: { userText: "hello", assistantText: "done" },
},
{ type: "refresh-threads" },
]);
});

2
tests/features/chat/application/threads/resume-actions.test.ts
View file

@ -52,6 +52,7 @@ function createActions(response: ThreadResumeSnapshot | null = activation("threa
systemItem: (text: string) => ({ id: "system", kind: "system" as const, role: "system" as const, text }),
resetThreadTurnPresence: vi.fn(),
notifyActiveThreadIdentityChanged: vi.fn(),
recordResumedThread: vi.fn(),
addSystemMessage: vi.fn(),
refreshLiveState: vi.fn(),
syncThreadGoal: vi.fn().mockResolvedValue(undefined),
@ -81,6 +82,7 @@ describe("ResumeActions", () => {
expect(loadLatest).toHaveBeenCalledWith("thread");
expect(host.resetThreadTurnPresence).toHaveBeenCalledWith(false);
expect(host.notifyActiveThreadIdentityChanged).toHaveBeenCalledOnce();
expect(host.recordResumedThread).toHaveBeenCalledWith(panelThread("thread"));
});
it("hydrates resumed threads from the initial turns page when app-server returns one", async () => {

27
tests/features/chat/application/threads/thread-management-actions.test.ts
View file

@ -37,8 +37,7 @@ type ThreadManagementActionsHostMock = Omit<
| "openThreadInCurrentPanel"
| "openThreadInNewView"
| "operations"
| "recordForkedThread"
| "refreshAfterThreadMutation"
| "recordThread"
| "setComposerText"
| "setStatus"
| "threadTransport"
@ -51,8 +50,7 @@ type ThreadManagementActionsHostMock = Omit<
openThreadInNewView: Mock<ThreadManagementActionsHost["openThreadInNewView"]>;
openThreadInCurrentPanel: Mock<ThreadManagementActionsHost["openThreadInCurrentPanel"]>;
notifyActiveThreadIdentityChanged: Mock<ThreadManagementActionsHost["notifyActiveThreadIdentityChanged"]>;
refreshAfterThreadMutation: Mock<ThreadManagementActionsHost["refreshAfterThreadMutation"]>;
recordForkedThread: Mock<ThreadManagementActionsHost["recordForkedThread"]>;
recordThread: Mock<ThreadManagementActionsHost["recordThread"]>;
};
describe("thread management actions", () => {
@ -300,7 +298,7 @@ describe("thread management actions", () => {
await controller.forkThreadFromTurn("source", "turn-1", false);
expect(host.threadTransport.forkThread).toHaveBeenCalledWith("source", "turn-1");
expect(host.recordForkedThread).toHaveBeenCalledWith(
expect(host.recordThread).toHaveBeenCalledWith(
expect.objectContaining({
id: "forked",
}),
@ -497,7 +495,7 @@ describe("thread management actions", () => {
await controller.forkThreadFromTurn("source", null, false);
expect(host.recordForkedThread).not.toHaveBeenCalled();
expect(host.recordThread).not.toHaveBeenCalled();
expect(host.openThreadInNewView).not.toHaveBeenCalled();
expect(host.operations.archiveThread).not.toHaveBeenCalled();
});
@ -515,7 +513,7 @@ describe("thread management actions", () => {
await controller.forkThreadFromTurn("source", null, true);
expect(host.recordForkedThread).toHaveBeenCalledWith(panelThread("forked"));
expect(host.recordThread).toHaveBeenCalledWith(panelThread("forked"));
expect(host.operations.renameThread).not.toHaveBeenCalled();
expect(host.operations.archiveThread).not.toHaveBeenCalled();
expect(host.openThreadInCurrentPanel).not.toHaveBeenCalled();
@ -577,7 +575,7 @@ describe("thread management actions", () => {
{ kind: "dialogue", role: "assistant", text: "kept answer", turnId: "kept-turn" },
]);
expect(callOrder(host.setComposerText)).toBeLessThan(callOrder(host.addSystemMessage));
expect(callOrder(host.addSystemMessage)).toBeLessThan(callOrder(host.refreshAfterThreadMutation));
expect(host.recordThread).toHaveBeenCalledWith(rollbackSnapshot().thread);
});
it("does not roll back an ephemeral side chat", async () => {
@ -648,7 +646,7 @@ describe("thread management actions", () => {
expect(activeThreadId(host.stateStore.getState())).toBe("other");
expect(host.setComposerText).not.toHaveBeenCalled();
expect(host.notifyActiveThreadIdentityChanged).not.toHaveBeenCalled();
expect(host.refreshAfterThreadMutation).toHaveBeenCalledOnce();
expect(host.recordThread).not.toHaveBeenCalled();
});
it("ignores rollback responses after a new turn starts in the same thread", async () => {
@ -681,10 +679,10 @@ describe("thread management actions", () => {
expect(host.stateStore.getState().turn.lifecycle).toEqual({ kind: "running", turnId: "new-turn" });
expect(host.setComposerText).not.toHaveBeenCalled();
expect(host.refreshAfterThreadMutation).toHaveBeenCalledOnce();
expect(host.recordThread).not.toHaveBeenCalled();
});
it("refreshes shared thread facts when rollback completes in a stale app-server context", async () => {
it("does not publish rollback facts from a stale app-server context", async () => {
const host = hostMock({
items: turnItems(),
activeThread: { id: "source" },
@ -698,7 +696,7 @@ describe("thread management actions", () => {
await controller.rollbackThread("source");
expect(host.refreshAfterThreadMutation).toHaveBeenCalledOnce();
expect(host.recordThread).not.toHaveBeenCalled();
expect(host.setComposerText).not.toHaveBeenCalled();
expect(host.notifyActiveThreadIdentityChanged).not.toHaveBeenCalled();
});
@ -737,7 +735,7 @@ describe("thread management actions", () => {
expect(host.setComposerText).not.toHaveBeenCalled();
expect(host.notifyActiveThreadIdentityChanged).not.toHaveBeenCalled();
expect(host.refreshAfterThreadMutation).not.toHaveBeenCalled();
expect(host.recordThread).not.toHaveBeenCalled();
});
});
@ -831,8 +829,7 @@ function hostMock({
openThreadInNewView: vi.fn<ThreadManagementActionsHost["openThreadInNewView"]>().mockResolvedValue(undefined),
openThreadInCurrentPanel: vi.fn<ThreadManagementActionsHost["openThreadInCurrentPanel"]>().mockResolvedValue(undefined),
notifyActiveThreadIdentityChanged: vi.fn<ThreadManagementActionsHost["notifyActiveThreadIdentityChanged"]>(),
refreshAfterThreadMutation: vi.fn<ThreadManagementActionsHost["refreshAfterThreadMutation"]>().mockResolvedValue(undefined),
recordForkedThread: vi.fn<ThreadManagementActionsHost["recordForkedThread"]>(),
recordThread: vi.fn<ThreadManagementActionsHost["recordThread"]>(),
};
}

4
tests/features/chat/application/turns/runtime-event-plan.test.ts
View file

@ -20,9 +20,9 @@ describe("TurnRuntimeEvent planner", () => {
it("reports turn starts as turn-runtime outcomes", () => {
const state = chatStateWith(chatStateFixture(), { activeThread: { id: "thread-active" } });
const plan = planTurnRuntimeEvents(state, [{ type: "turnStarted", threadId: "thread-active", turnId: "turn-active", recencyAt: 123 }]);
const plan = planTurnRuntimeEvents(state, [{ type: "turnStarted", threadId: "thread-active", turnId: "turn-active" }]);
expect(plan.outcomes).toEqual([{ type: "turn-started", threadId: "thread-active", turnId: "turn-active", recencyAt: 123 }]);
expect(plan.outcomes).toEqual([]);
});
it("reconciles completed turn snapshots with optimistic local user dialogues", () => {

10
tests/features/chat/host/view-connection.test.ts
View file

@ -1580,18 +1580,10 @@ function chatHost(overrides: ChatHostFixtureOverrides = {}): TestCodexChatHost {
activeThreads = activeThreads?.map((thread) => (thread.id === event.threadId ? { ...thread, name: event.name } : thread)) ?? null;
emitActiveThreads();
return;
case "thread-started":
case "thread-forked":
case "thread-upserted":
case "thread-restored":
upsertActiveThread(event.thread);
return;
case "thread-touched":
activeThreads =
activeThreads?.map((thread) =>
thread.id === event.threadId && event.recencyAt !== undefined ? { ...thread, recencyAt: event.recencyAt } : thread,
) ?? activeThreads;
emitActiveThreads();
return;
case "thread-unarchived":
return;
}

26
tests/features/thread-picker/modal.test.ts
View file

@ -43,11 +43,22 @@ describe("threadPickerSuggestions", () => {
expect((await modal.getSuggestions("")).map((item) => item.thread.id)).toEqual(["recent"]);
expect(host.completeHistoryLoads).toBe(0);
expect(host.sharedRefreshes).toBe(0);
const [needleSuggestions] = await Promise.all([modal.getSuggestions("needle"), modal.getSuggestions("target")]);
expect(needleSuggestions.map((item) => item.thread.id)).toEqual(["older-target"]);
expect(host.completeHistoryLoads).toBe(1);
});
it("uses a modal-local inventory when the shared recent list is cold", async () => {
const host = threadPickerHost([], [thread({ id: "thread" })], false);
const modal = await openedThreadPicker(host);
expect((await modal.getSuggestions("")).map((item) => item.thread.id)).toEqual(["thread"]);
expect((await modal.getSuggestions("thread")).map((item) => item.thread.id)).toEqual(["thread"]);
expect(host.completeHistoryLoads).toBe(1);
expect(host.sharedRefreshes).toBe(0);
});
});
describe("threadOpenModeFromEvent", () => {
@ -114,9 +125,14 @@ interface TestThreadPickerHost extends ThreadPickerHost {
openedCurrent: string[];
openedAvailable: string[];
completeHistoryLoads: number;
sharedRefreshes: number;
}
function threadPickerHost(firstPage: readonly Thread[], completeHistory: readonly Thread[] = firstPage): TestThreadPickerHost {
function threadPickerHost(
firstPage: readonly Thread[],
completeHistory: readonly Thread[] = firstPage,
hasSharedSnapshot = true,
): TestThreadPickerHost {
const openedCurrent: string[] = [];
const openedAvailable: string[] = [];
const host: TestThreadPickerHost = {
@ -124,13 +140,17 @@ function threadPickerHost(firstPage: readonly Thread[], completeHistory: readonl
openedCurrent,
openedAvailable,
completeHistoryLoads: 0,
sharedRefreshes: 0,
threadCatalog: {
activeSnapshot: () => null,
activeSnapshot: () => (hasSharedSnapshot ? firstPage : null),
loadActive: async () => {
host.completeHistoryLoads += 1;
return completeHistory;
},
refreshActive: async () => firstPage,
refreshActive: async () => {
host.sharedRefreshes += 1;
return firstPage;
},
observeActive: () => () => undefined,
},
openThreadInCurrentView: async (threadId) => {

16
tests/features/threads/catalog/thread-catalog.test.ts
View file

@ -23,16 +23,13 @@ describe("ThreadCatalog", () => {
expect(onEventApplied).toHaveBeenCalledTimes(2);
});
it("requests an authoritative archived read when an archive event lacks a source snapshot", () => {
it("does not invent an archived record when an archive event lacks a source snapshot", () => {
const store = catalogStore();
const catalog = createThreadCatalog({ store });
catalog.apply({ type: "thread-archived", threadId: "unknown" });
expect(store.appliedMutations).toEqual([
{ kind: "remove", list: "active", threadId: "unknown" },
{ kind: "refresh", list: "archived" },
]);
expect(store.appliedMutations).toEqual([{ kind: "remove", list: "active", threadId: "unknown" }]);
});
it("moves known restored and unarchived threads between lists", () => {
@ -49,18 +46,17 @@ describe("ThreadCatalog", () => {
expect(catalog.archivedSnapshot()).toEqual([]);
});
it("patches touch and delete facts without inventing unknown thread records", () => {
it("patches exact rename and delete facts without inventing unknown thread records", () => {
const store = catalogStore({
active: [thread("active", false, { recencyAt: 1 }), thread("kept")],
active: [thread("active"), thread("kept")],
archived: [thread("deleted", true)],
});
const catalog = createThreadCatalog({ store });
catalog.apply({ type: "thread-touched", threadId: "active", recencyAt: 20 });
catalog.apply({ type: "thread-renamed", threadId: "missing", name: "Not invented" });
catalog.apply({ type: "thread-deleted", threadId: "deleted" });
expect(catalog.activeSnapshot()).toEqual([thread("active", false, { recencyAt: 20 }), thread("kept")]);
expect(catalog.activeSnapshot()).toEqual([thread("active"), thread("kept")]);
expect(catalog.archivedSnapshot()).toEqual([]);
});
@ -69,7 +65,7 @@ describe("ThreadCatalog", () => {
const catalog = createThreadCatalog({ store });
const staleContext = { codexPath: "codex", vaultPath: "/vault", generation: 0 };
catalog.applyConnectionEvent(staleContext, { type: "thread-started", thread: thread("stale") });
catalog.applyConnectionEvent(staleContext, { type: "thread-upserted", thread: thread("stale") });
expect(store.appliedMutations).toEqual([]);
expect(catalog.activeSnapshot()).toBeNull();