mirror of
https://github.com/murashit/codex-panel.git
synced 2026-07-22 06:57:10 +00:00
refactor(threads): bound lifecycle event arbitration
This commit is contained in:
murashit
parent
ffb11490ce
commit
dfeeb0e4b2
7 changed files with 322 additions and 887 deletions
|
|
@ -55,7 +55,6 @@ interface AppServerQueryOptions<T> {
|
|||
readonly staleTime: number;
|
||||
}
|
||||
|
||||
type ThreadListKind = "active" | "archived";
|
||||
interface MetadataResourceSnapshot<T> {
|
||||
readonly value: T;
|
||||
readonly probe: DiagnosticProbeResult;
|
||||
|
|
@ -78,6 +77,7 @@ export class AppServerQueryCache {
|
|||
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;
|
||||
|
|
@ -96,6 +96,8 @@ export class AppServerQueryCache {
|
|||
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();
|
||||
|
|
@ -179,6 +181,30 @@ export class AppServerQueryCache {
|
|||
return this.fetchArchivedThreads({ force: true });
|
||||
}
|
||||
|
||||
applyThreadListMutations(mutations: readonly ThreadListMutation[]): void {
|
||||
this.assertUsable();
|
||||
if (!appServerQueryContextIsComplete(this.context) || mutations.length === 0) return;
|
||||
for (const kind of ["active", "archived"] as const) {
|
||||
const relevant = mutations.filter((mutation) => mutation.list === kind);
|
||||
if (relevant.length === 0) continue;
|
||||
|
||||
const key = this.threadListQueryKey(kind);
|
||||
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(() => {
|
||||
// Query observers retain refresh failures while the event projection remains last-known-good state.
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private threadListSnapshot(kind: ThreadListKind): readonly Thread[] | null {
|
||||
if (!appServerQueryContextIsComplete(this.context)) return null;
|
||||
const threads = this.client.getQueryData<readonly Thread[]>(this.threadListQueryKey(kind));
|
||||
|
|
@ -357,13 +383,21 @@ export class AppServerQueryCache {
|
|||
queryKey: this.threadListQueryKey(kind),
|
||||
queryFn: async (): Promise<readonly Thread[]> => {
|
||||
if (kind === "active") {
|
||||
const revision = this.activeThreadRevision;
|
||||
const page = await this.runWithClient((client) => readThreadPage(client, this.context.vaultPath, { archived: false }));
|
||||
if (this.activeThreadRevision !== revision) return this.activeThreadsSnapshot() ?? [];
|
||||
this.rememberActiveThreadCursor(page.nextCursor);
|
||||
return cloneThreads(page.threads);
|
||||
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);
|
||||
}
|
||||
return cloneThreads(await this.runWithClient((client) => listThreads(client, this.context.vaultPath, { archived: true })));
|
||||
},
|
||||
staleTime: THREAD_LIST_STALE_TIME_MS,
|
||||
};
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ import {
|
|||
createAppServerContextLease,
|
||||
} from "./keys";
|
||||
import type { ObservedResultListener } from "./observed-result";
|
||||
import type { ThreadListMutation } from "./thread-list-mutation";
|
||||
|
||||
export interface AppServerResourceStoreOptions {
|
||||
cacheFactory?: (context: AppServerQueryContextIdentity) => AppServerQueryCache;
|
||||
|
|
@ -122,6 +123,10 @@ export class AppServerResourceStore {
|
|||
return this.runForCurrentContext((cache) => cache.refreshArchivedThreads());
|
||||
}
|
||||
|
||||
applyThreadListMutations(mutations: readonly ThreadListMutation[]): void {
|
||||
this.currentCache().applyThreadListMutations(mutations);
|
||||
}
|
||||
|
||||
observeActiveThreadsResult(listener: ObservedResultListener<readonly Thread[]>, options?: { emitCurrent?: boolean }): () => void {
|
||||
return this.observeCurrentContext(
|
||||
(cache, contextListener, observeOptions) => cache.observeActiveThreadsResult(contextListener, observeOptions),
|
||||
|
|
|
|||
31
src/app-server/query/thread-list-mutation.ts
Normal file
31
src/app-server/query/thread-list-mutation.ts
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
import type { Thread } from "../../domain/threads/model";
|
||||
|
||||
export type ThreadListKind = "active" | "archived";
|
||||
|
||||
export type ThreadListMutation =
|
||||
| { readonly kind: "upsert"; readonly list: ThreadListKind; readonly thread: Thread }
|
||||
| { readonly kind: "remove"; readonly list: ThreadListKind; readonly threadId: string }
|
||||
| {
|
||||
readonly kind: "update";
|
||||
readonly list: ThreadListKind;
|
||||
readonly threadId: string;
|
||||
readonly changes: Partial<Pick<Thread, "name" | "recencyAt">>;
|
||||
}
|
||||
| { readonly kind: "refresh"; readonly list: ThreadListKind };
|
||||
|
||||
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);
|
||||
if (index < 0) return [mutation.thread, ...snapshot];
|
||||
return snapshot.map((thread) => (thread.id === mutation.thread.id ? mutation.thread : thread));
|
||||
}
|
||||
case "remove":
|
||||
return snapshot?.filter((thread) => thread.id !== mutation.threadId) ?? null;
|
||||
case "update":
|
||||
return snapshot?.map((thread) => (thread.id === mutation.threadId ? { ...thread, ...mutation.changes } : thread)) ?? null;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,5 +1,6 @@
|
|||
import type { AppServerQueryContextIdentity } from "../../../app-server/query/keys";
|
||||
import type { ObservedResult, ObservedResultListener } from "../../../app-server/query/observed-result";
|
||||
import type { ObservedResultListener } from "../../../app-server/query/observed-result";
|
||||
import type { ThreadListMutation } from "../../../app-server/query/thread-list-mutation";
|
||||
import type { Thread } from "../../../domain/threads/model";
|
||||
|
||||
type ThreadListObserver = ObservedResultListener<readonly Thread[]>;
|
||||
|
|
@ -14,35 +15,11 @@ interface ThreadCatalogStore {
|
|||
loadMoreActiveThreads(): Promise<readonly Thread[]>;
|
||||
refreshActiveThreads(): Promise<readonly Thread[]>;
|
||||
refreshArchivedThreads(): Promise<readonly Thread[]>;
|
||||
applyThreadListMutations(mutations: readonly ThreadListMutation[]): void;
|
||||
observeActiveThreadsResult(observer: ThreadListObserver, options?: { emitCurrent?: boolean }): () => void;
|
||||
observeArchivedThreadsResult(observer: ThreadListObserver, options?: { emitCurrent?: boolean }): () => void;
|
||||
}
|
||||
|
||||
type ThreadCatalogEventStore = Pick<
|
||||
ThreadCatalogStore,
|
||||
"activeThreadsSnapshot" | "archivedThreadsSnapshot" | "refreshActiveThreads" | "refreshArchivedThreads"
|
||||
>;
|
||||
|
||||
interface PendingThreadUpsert {
|
||||
readonly thread: Thread;
|
||||
readonly acknowledgedBy: (thread: Thread) => boolean;
|
||||
}
|
||||
|
||||
type PendingThreadUpserts = Map<string, PendingThreadUpsert>;
|
||||
type PendingThreadRemovals = Set<string>;
|
||||
|
||||
interface PendingThreadListFacts {
|
||||
readonly upserts: PendingThreadUpserts;
|
||||
readonly removals: PendingThreadRemovals;
|
||||
}
|
||||
|
||||
interface ThreadCatalogFacts {
|
||||
readonly active: PendingThreadListFacts;
|
||||
readonly archived: PendingThreadListFacts;
|
||||
activeObservedResult: ObservedResult<readonly Thread[]> | null;
|
||||
archivedObservedResult: ObservedResult<readonly Thread[]> | null;
|
||||
}
|
||||
|
||||
type ThreadCatalogEventObserver = (event: ThreadCatalogEvent) => void;
|
||||
|
||||
export interface ThreadCatalogOptions {
|
||||
|
|
@ -90,313 +67,83 @@ export interface ThreadCatalog
|
|||
extends ThreadCatalogPaginatedActiveReader,
|
||||
ThreadCatalogArchivedReader,
|
||||
ThreadCatalogEventSink,
|
||||
ThreadCatalogConnectionEventSink {
|
||||
clear(): void;
|
||||
}
|
||||
ThreadCatalogConnectionEventSink {}
|
||||
|
||||
export function createThreadCatalog(options: ThreadCatalogOptions): ThreadCatalog {
|
||||
let factsState: { contextKey: string; facts: ThreadCatalogFacts } | null = null;
|
||||
const activeObservers = new Set<ThreadListObserver>();
|
||||
const archivedObservers = new Set<ThreadListObserver>();
|
||||
const { store } = options;
|
||||
const currentFacts = (): ThreadCatalogFacts => {
|
||||
const contextKey = store.contextKey();
|
||||
if (factsState?.contextKey === contextKey) return factsState.facts;
|
||||
const facts = pendingThreadCatalogFacts();
|
||||
factsState = { contextKey, facts };
|
||||
return facts;
|
||||
};
|
||||
const activeRawSnapshot = (): readonly Thread[] | null => store.activeThreadsSnapshot();
|
||||
const archivedRawSnapshot = (): readonly Thread[] | null => store.archivedThreadsSnapshot();
|
||||
const publish = (kind: ThreadListKind): void => {
|
||||
const facts = currentFacts();
|
||||
const raw = kind === "active" ? activeRawSnapshot() : archivedRawSnapshot();
|
||||
const value = threadListProjection(raw, kind === "active" ? facts.active : facts.archived);
|
||||
const observedResult = kind === "active" ? facts.activeObservedResult : facts.archivedObservedResult;
|
||||
const result: ObservedResult<readonly Thread[]> = {
|
||||
...(observedResult ?? { error: null, isFetching: false }),
|
||||
value,
|
||||
};
|
||||
for (const observer of kind === "active" ? activeObservers : archivedObservers) observer(result);
|
||||
};
|
||||
const applyToFacts = (
|
||||
eventStore: ThreadCatalogEventStore,
|
||||
facts: ThreadCatalogFacts,
|
||||
event: ThreadCatalogEvent,
|
||||
publishChanges: boolean,
|
||||
): void => {
|
||||
const changed = applyThreadCatalogEvent(eventStore, facts, event);
|
||||
if (publishChanges) {
|
||||
if (changed.active) publish("active");
|
||||
if (changed.archived) publish("archived");
|
||||
options.onEventApplied?.(event);
|
||||
}
|
||||
};
|
||||
const apply = (event: ThreadCatalogEvent): void => {
|
||||
applyToFacts(store, currentFacts(), event, true);
|
||||
store.applyThreadListMutations(threadListMutationsForEvent(store, event));
|
||||
options.onEventApplied?.(event);
|
||||
};
|
||||
|
||||
return {
|
||||
apply,
|
||||
applyConnectionEvent: (context, event) => {
|
||||
const capturedContextKey = store.contextKeyFor(context);
|
||||
if (capturedContextKey !== store.contextKey()) return;
|
||||
applyToFacts(store, currentFacts(), event, true);
|
||||
if (store.contextKeyFor(context) !== store.contextKey()) return;
|
||||
apply(event);
|
||||
},
|
||||
clear: () => {
|
||||
factsState = null;
|
||||
},
|
||||
activeSnapshot: () => threadListProjection(activeRawSnapshot(), currentFacts().active),
|
||||
loadActive: () => loadThreadList(store.fetchAllActiveThreads(), currentFacts().active),
|
||||
refreshActive: () => loadThreadList(store.refreshActiveThreads(), currentFacts().active),
|
||||
activeSnapshot: () => store.activeThreadsSnapshot(),
|
||||
loadActive: () => store.fetchAllActiveThreads(),
|
||||
refreshActive: () => store.refreshActiveThreads(),
|
||||
hasMoreActive: () => store.hasMoreActiveThreads(),
|
||||
loadMoreActive: () => loadThreadList(store.loadMoreActiveThreads(), currentFacts().active),
|
||||
observeActive: (observer, observeOptions) => {
|
||||
activeObservers.add(observer);
|
||||
const unsubscribe = store.observeActiveThreadsResult((result) => {
|
||||
const facts = currentFacts();
|
||||
facts.activeObservedResult = result;
|
||||
if (result.value) acknowledgeThreadListSnapshot(facts.active, result.value);
|
||||
observer({
|
||||
...result,
|
||||
value: threadListProjection(result.value, facts.active),
|
||||
});
|
||||
}, observeOptions);
|
||||
return () => {
|
||||
activeObservers.delete(observer);
|
||||
unsubscribe();
|
||||
};
|
||||
},
|
||||
archivedSnapshot: () => threadListProjection(archivedRawSnapshot(), currentFacts().archived),
|
||||
refreshArchived: () => loadThreadList(store.refreshArchivedThreads(), currentFacts().archived),
|
||||
observeArchived: (observer, observeOptions) => {
|
||||
archivedObservers.add(observer);
|
||||
const unsubscribe = store.observeArchivedThreadsResult((result) => {
|
||||
const facts = currentFacts();
|
||||
facts.archivedObservedResult = result;
|
||||
if (result.value) acknowledgeThreadListSnapshot(facts.archived, result.value);
|
||||
observer({
|
||||
...result,
|
||||
value: threadListProjection(result.value, facts.archived),
|
||||
});
|
||||
}, observeOptions);
|
||||
return () => {
|
||||
archivedObservers.delete(observer);
|
||||
unsubscribe();
|
||||
};
|
||||
},
|
||||
loadMoreActive: () => store.loadMoreActiveThreads(),
|
||||
observeActive: (observer, observeOptions) => store.observeActiveThreadsResult(observer, observeOptions),
|
||||
archivedSnapshot: () => store.archivedThreadsSnapshot(),
|
||||
refreshArchived: () => store.refreshArchivedThreads(),
|
||||
observeArchived: (observer, observeOptions) => store.observeArchivedThreadsResult(observer, observeOptions),
|
||||
};
|
||||
}
|
||||
|
||||
type ThreadListKind = "active" | "archived";
|
||||
|
||||
function pendingThreadCatalogFacts(): ThreadCatalogFacts {
|
||||
return {
|
||||
active: pendingThreadListFacts(),
|
||||
archived: pendingThreadListFacts(),
|
||||
activeObservedResult: null,
|
||||
archivedObservedResult: null,
|
||||
};
|
||||
}
|
||||
|
||||
function applyThreadCatalogEvent(
|
||||
store: ThreadCatalogEventStore,
|
||||
facts: ThreadCatalogFacts,
|
||||
event: ThreadCatalogEvent,
|
||||
): { active: boolean; archived: boolean } {
|
||||
const { active: activeFacts, archived: archivedFacts } = facts;
|
||||
function threadListMutationsForEvent(store: ThreadCatalogStore, event: ThreadCatalogEvent): ThreadListMutation[] {
|
||||
switch (event.type) {
|
||||
case "thread-started":
|
||||
upsertActiveThread(activeFacts, event.thread, acknowledgeByThreadId);
|
||||
return { active: true, archived: false };
|
||||
case "thread-forked":
|
||||
upsertActiveThread(activeFacts, event.thread, acknowledgedByThreadVersion(event.thread));
|
||||
return { active: true, archived: false };
|
||||
return [{ kind: "upsert", list: "active", thread: { ...event.thread, archived: false } }];
|
||||
case "thread-touched":
|
||||
applyThreadTouchedEvent(store, facts, event.threadId, event.recencyAt);
|
||||
return { active: true, archived: false };
|
||||
return [
|
||||
{
|
||||
kind: "update",
|
||||
list: "active",
|
||||
threadId: event.threadId,
|
||||
changes: event.recencyAt === undefined ? {} : { recencyAt: event.recencyAt },
|
||||
},
|
||||
];
|
||||
case "thread-renamed":
|
||||
applyThreadRenamedEvent(store, facts, event.threadId, event.name);
|
||||
return { active: true, archived: true };
|
||||
case "thread-archived":
|
||||
applyThreadArchivedEvent(store, facts, event.threadId);
|
||||
return { active: true, archived: true };
|
||||
return [
|
||||
{ kind: "update", list: "active", threadId: event.threadId, changes: { name: event.name } },
|
||||
{ kind: "update", list: "archived", threadId: event.threadId, changes: { name: event.name } },
|
||||
];
|
||||
case "thread-archived": {
|
||||
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]),
|
||||
];
|
||||
}
|
||||
case "thread-deleted":
|
||||
applyThreadDeletedEvent(activeFacts, archivedFacts, event.threadId);
|
||||
return { active: true, archived: true };
|
||||
return [
|
||||
{ kind: "remove", list: "active", threadId: event.threadId },
|
||||
{ kind: "remove", list: "archived", threadId: event.threadId },
|
||||
];
|
||||
case "thread-restored":
|
||||
applyThreadRestoredEvent(activeFacts, archivedFacts, event.thread);
|
||||
return { active: true, archived: true };
|
||||
case "thread-unarchived":
|
||||
applyThreadUnarchivedEvent(store, facts, event.threadId);
|
||||
return { active: true, archived: true };
|
||||
}
|
||||
}
|
||||
|
||||
async function loadThreadList(threadsPromise: Promise<readonly Thread[]>, facts: PendingThreadListFacts): Promise<readonly Thread[]> {
|
||||
const threads = await threadsPromise;
|
||||
acknowledgeThreadListSnapshot(facts, threads);
|
||||
return threadListProjection(threads, facts) ?? [];
|
||||
}
|
||||
|
||||
function upsertActiveThread(activeFacts: PendingThreadListFacts, thread: Thread, acknowledgedBy: (thread: Thread) => boolean): void {
|
||||
rememberPendingThreadUpsert(activeFacts, thread, acknowledgedBy);
|
||||
}
|
||||
|
||||
function applyThreadTouchedEvent(
|
||||
store: ThreadCatalogEventStore,
|
||||
facts: ThreadCatalogFacts,
|
||||
threadId: string,
|
||||
recencyAt: number | null | undefined,
|
||||
): void {
|
||||
const activeFacts = facts.active;
|
||||
const existingFact = activeFacts.upserts.get(threadId)?.thread ?? null;
|
||||
let touchedThread = existingFact ? touchedActiveThread(existingFact, recencyAt) : null;
|
||||
const currentThread =
|
||||
threadListProjection(store.activeThreadsSnapshot(), activeFacts)?.find((thread) => thread.id === threadId) ?? touchedThread;
|
||||
if (currentThread) touchedThread = touchedActiveThread(currentThread, recencyAt);
|
||||
if (!touchedThread) return;
|
||||
const promotedThread = touchedThread;
|
||||
rememberPendingThreadUpsert(
|
||||
activeFacts,
|
||||
promotedThread,
|
||||
recencyAt === undefined ? acknowledgeByThreadId : acknowledgedByRecency(recencyAt),
|
||||
);
|
||||
}
|
||||
|
||||
function applyThreadRenamedEvent(store: ThreadCatalogEventStore, facts: ThreadCatalogFacts, threadId: string, name: string | null): void {
|
||||
const { active: activeFacts, archived: archivedFacts } = facts;
|
||||
const activeThread = threadListProjection(store.activeThreadsSnapshot(), activeFacts)?.find((thread) => thread.id === threadId) ?? null;
|
||||
if (activeThread) rememberPendingThreadUpsert(activeFacts, { ...activeThread, name }, acknowledgedByName(name));
|
||||
const archivedThread =
|
||||
threadListProjection(store.archivedThreadsSnapshot(), archivedFacts)?.find((thread) => thread.id === threadId) ?? null;
|
||||
if (archivedThread) rememberPendingThreadUpsert(archivedFacts, { ...archivedThread, name }, acknowledgedByName(name));
|
||||
}
|
||||
|
||||
function applyThreadArchivedEvent(store: ThreadCatalogEventStore, facts: ThreadCatalogFacts, threadId: string): void {
|
||||
const { active: activeFacts, archived: archivedFacts } = facts;
|
||||
const archivedThread = threadListProjection(store.activeThreadsSnapshot(), activeFacts)?.find((thread) => thread.id === threadId);
|
||||
rememberPendingThreadRemoval(activeFacts, threadId);
|
||||
if (archivedThread) {
|
||||
const pendingArchivedThread = { ...archivedThread, archived: true };
|
||||
rememberPendingThreadUpsert(archivedFacts, pendingArchivedThread, acknowledgeByThreadId);
|
||||
} else {
|
||||
refreshArchivedThreadsAfterUnknownArchive(store, archivedFacts);
|
||||
}
|
||||
}
|
||||
|
||||
function applyThreadDeletedEvent(activeFacts: PendingThreadListFacts, archivedFacts: PendingThreadListFacts, threadId: string): void {
|
||||
rememberPendingThreadRemoval(activeFacts, threadId);
|
||||
rememberPendingThreadRemoval(archivedFacts, threadId);
|
||||
}
|
||||
|
||||
function applyThreadRestoredEvent(activeFacts: PendingThreadListFacts, archivedFacts: PendingThreadListFacts, thread: Thread): void {
|
||||
upsertActiveThread(activeFacts, { ...thread, archived: false }, acknowledgeByThreadId);
|
||||
rememberPendingThreadRemoval(archivedFacts, thread.id);
|
||||
}
|
||||
|
||||
function applyThreadUnarchivedEvent(store: ThreadCatalogEventStore, facts: ThreadCatalogFacts, threadId: string): void {
|
||||
const { active: activeFacts, archived: archivedFacts } = facts;
|
||||
const restoredThread =
|
||||
threadListProjection(store.archivedThreadsSnapshot(), archivedFacts)?.find((thread) => thread.id === threadId) ?? null;
|
||||
rememberPendingThreadRemoval(archivedFacts, threadId);
|
||||
if (restoredThread) {
|
||||
upsertActiveThread(activeFacts, { ...restoredThread, archived: false }, acknowledgeByThreadId);
|
||||
return;
|
||||
}
|
||||
refreshThreadListsAfterUnknownUnarchive(store, activeFacts, archivedFacts);
|
||||
}
|
||||
|
||||
function threadListProjection(snapshot: readonly Thread[] | null, facts: PendingThreadListFacts): readonly Thread[] | null {
|
||||
if (!snapshot && facts.upserts.size === 0 && facts.removals.size === 0) return null;
|
||||
const threads = (snapshot ?? [])
|
||||
.filter((thread) => !facts.removals.has(thread.id))
|
||||
.map((thread) => facts.upserts.get(thread.id)?.thread ?? thread);
|
||||
const snapshotThreadIds = new Set(threads.map((thread) => thread.id));
|
||||
const pendingThreads = Array.from(facts.upserts.values())
|
||||
.map((entry) => entry.thread)
|
||||
.filter((thread) => !facts.removals.has(thread.id) && !snapshotThreadIds.has(thread.id));
|
||||
return pendingThreads.length === 0 ? threads : [...pendingThreads.reverse(), ...threads];
|
||||
}
|
||||
|
||||
function pendingThreadListFacts(): PendingThreadListFacts {
|
||||
return {
|
||||
upserts: new Map(),
|
||||
removals: new Set(),
|
||||
};
|
||||
}
|
||||
|
||||
function acknowledgeThreadListSnapshot(facts: PendingThreadListFacts, threads: readonly Thread[]): void {
|
||||
const threadIds = new Set(threads.map((thread) => thread.id));
|
||||
for (const thread of threads) {
|
||||
const upsert = facts.upserts.get(thread.id);
|
||||
if (upsert?.acknowledgedBy(thread)) facts.upserts.delete(thread.id);
|
||||
}
|
||||
for (const threadId of [...facts.removals]) {
|
||||
if (!threadIds.has(threadId)) facts.removals.delete(threadId);
|
||||
}
|
||||
}
|
||||
|
||||
function rememberPendingThreadUpsert(facts: PendingThreadListFacts, thread: Thread, acknowledgedBy: (thread: Thread) => boolean): void {
|
||||
facts.removals.delete(thread.id);
|
||||
facts.upserts.delete(thread.id);
|
||||
facts.upserts.set(thread.id, { thread, acknowledgedBy });
|
||||
}
|
||||
|
||||
function rememberPendingThreadRemoval(facts: PendingThreadListFacts, threadId: string): void {
|
||||
facts.upserts.delete(threadId);
|
||||
facts.removals.add(threadId);
|
||||
}
|
||||
|
||||
function touchedActiveThread(thread: Thread, recencyAt: number | null | undefined): Thread {
|
||||
return recencyAt === undefined ? thread : { ...thread, recencyAt };
|
||||
}
|
||||
|
||||
function acknowledgeByThreadId(): boolean {
|
||||
return true;
|
||||
}
|
||||
|
||||
function acknowledgedByName(name: string | null): (thread: Thread) => boolean {
|
||||
return (thread) => thread.name === name;
|
||||
}
|
||||
|
||||
function acknowledgedByThreadVersion(reference: Thread): (thread: Thread) => boolean {
|
||||
return (thread) =>
|
||||
thread.updatedAt > reference.updatedAt ||
|
||||
(thread.updatedAt === reference.updatedAt &&
|
||||
thread.preview === reference.preview &&
|
||||
thread.name === reference.name &&
|
||||
thread.recencyAt === reference.recencyAt);
|
||||
}
|
||||
|
||||
function acknowledgedByRecency(recencyAt: number | null): (thread: Thread) => boolean {
|
||||
return (thread) => thread.recencyAt === recencyAt;
|
||||
}
|
||||
|
||||
function refreshArchivedThreadsAfterUnknownArchive(store: ThreadCatalogEventStore, archivedFacts: PendingThreadListFacts): void {
|
||||
// A force refresh can join an older in-flight archived request. Run one more
|
||||
// refresh afterward so an archive recorded during that request is not lost.
|
||||
void refreshArchivedThreadsTwice(store, archivedFacts);
|
||||
}
|
||||
|
||||
async function refreshArchivedThreadsTwice(store: ThreadCatalogEventStore, archivedFacts: PendingThreadListFacts): Promise<void> {
|
||||
for (let attempt = 0; attempt < 2; attempt += 1) {
|
||||
try {
|
||||
acknowledgeThreadListSnapshot(archivedFacts, await store.refreshArchivedThreads());
|
||||
} catch {
|
||||
// The query observer retains the failure for diagnostics; continue so a joined stale request gets one fresh retry.
|
||||
return [
|
||||
{ kind: "upsert", list: "active", thread: { ...event.thread, archived: false } },
|
||||
{ kind: "remove", list: "archived", threadId: event.thread.id },
|
||||
];
|
||||
case "thread-unarchived": {
|
||||
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]),
|
||||
];
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function refreshThreadListsAfterUnknownUnarchive(
|
||||
store: ThreadCatalogEventStore,
|
||||
activeFacts: PendingThreadListFacts,
|
||||
archivedFacts: PendingThreadListFacts,
|
||||
): void {
|
||||
// Unknown unarchives need both lists: active gains the thread, archived loses it.
|
||||
void Promise.allSettled([store.refreshActiveThreads(), store.refreshArchivedThreads()]).then(([active, archived]) => {
|
||||
if (active.status === "fulfilled") acknowledgeThreadListSnapshot(activeFacts, active.value);
|
||||
if (archived.status === "fulfilled") acknowledgeThreadListSnapshot(archivedFacts, archived.value);
|
||||
});
|
||||
function threadById(threads: readonly Thread[] | null, threadId: string): Thread | null {
|
||||
return threads?.find((thread) => thread.id === threadId) ?? null;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -97,7 +97,6 @@ export class CodexPanelRuntime implements AppServerClientAccess {
|
|||
this.selectionRewriteController = null;
|
||||
this.panels.reset();
|
||||
this.appServerResourceStore.reset();
|
||||
this.threadCatalog.clear();
|
||||
}
|
||||
|
||||
activeWorkspaceLeafChanged(leaf: Parameters<WorkspacePanelCoordinator["activeLeafChanged"]>[0]): void {
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import { AppServerQueryCache } from "../../src/app-server/query/cache";
|
|||
import type { AppServerQueryContext, AppServerQueryContextIdentity } from "../../src/app-server/query/keys";
|
||||
import type { RateLimitSnapshot } from "../../src/domain/runtime/metrics";
|
||||
import type { RuntimePermissionProfileSummary } from "../../src/domain/runtime/permissions";
|
||||
import type { Thread } from "../../src/domain/threads/model";
|
||||
|
||||
describe("AppServerQueryCache", () => {
|
||||
it("garbage-collects inactive query records", async () => {
|
||||
|
|
@ -124,6 +125,33 @@ describe("AppServerQueryCache", () => {
|
|||
expect(cache.hasMoreActiveThreads()).toBe(true);
|
||||
});
|
||||
|
||||
it("rejects only a thread read started before an event and accepts the following server value", 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);
|
||||
const cache = cacheWithRequestHandlers({ "thread/list": listThreads });
|
||||
await cache.refreshActiveThreads();
|
||||
|
||||
const refresh = cache.refreshActiveThreads();
|
||||
await flushMicrotasks();
|
||||
cache.applyThreadListMutations([{ kind: "update", list: "active", threadId: "target", changes: { name: "from-event" } }]);
|
||||
|
||||
expect(cache.activeThreadsSnapshot()?.[0]?.name).toBe("from-event");
|
||||
staleRead.resolve({ data: [{ ...thread("target"), name: "stale" }], nextCursor: null });
|
||||
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 () => {
|
||||
const oldInventory = deferred<{ data: ReturnType<typeof thread>[]; nextCursor: string | null }>();
|
||||
const listThreads = vi
|
||||
|
|
@ -231,6 +259,48 @@ describe("AppServerQueryCache", () => {
|
|||
unsubscribe();
|
||||
});
|
||||
|
||||
it("publishes metadata resources only after a retry settles", async () => {
|
||||
const retry = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>();
|
||||
const nextRetry = deferred<{ data: { skills: CatalogSkillMetadata[] }[] }>();
|
||||
const listSkills = vi
|
||||
.fn()
|
||||
.mockRejectedValueOnce(new Error("skills offline"))
|
||||
.mockImplementationOnce(() => retry.promise)
|
||||
.mockImplementationOnce(() => nextRetry.promise);
|
||||
const cache = cacheWithRequestHandlers({ "skills/list": listSkills });
|
||||
await cache.refreshSkills();
|
||||
const listener = vi.fn();
|
||||
const unsubscribe = cache.observeAppServerMetadataResources(listener, { emitCurrent: false });
|
||||
|
||||
const refresh = cache.refreshSkills();
|
||||
await flushMicrotasks();
|
||||
|
||||
expect(listener).not.toHaveBeenCalled();
|
||||
retry.resolve({ data: [{ skills: [catalogSkill("writer")] }] });
|
||||
await refresh;
|
||||
|
||||
expect(listener).toHaveBeenCalledOnce();
|
||||
expect(listener).toHaveBeenCalledWith({
|
||||
id: "skills",
|
||||
value: [expect.objectContaining({ name: "writer" })],
|
||||
probe: expect.objectContaining({ id: "skills", status: "ok" }),
|
||||
});
|
||||
|
||||
const nextRefresh = cache.refreshSkills();
|
||||
await flushMicrotasks();
|
||||
expect(listener).toHaveBeenCalledOnce();
|
||||
nextRetry.resolve({ data: [{ skills: [catalogSkill("editor")] }] });
|
||||
await nextRefresh;
|
||||
|
||||
expect(listener).toHaveBeenCalledTimes(2);
|
||||
expect(listener).toHaveBeenLastCalledWith({
|
||||
id: "skills",
|
||||
value: [expect.objectContaining({ name: "editor" })],
|
||||
probe: expect.objectContaining({ id: "skills", status: "ok" }),
|
||||
});
|
||||
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[] }[] }>();
|
||||
|
|
@ -601,7 +671,7 @@ function appServerRateLimit(usedPercent: number): RateLimitSnapshot {
|
|||
};
|
||||
}
|
||||
|
||||
function thread(id: string, archived = false) {
|
||||
function thread(id: string, archived = false): Thread {
|
||||
return {
|
||||
id,
|
||||
preview: "",
|
||||
|
|
|
|||
|
|
@ -1,585 +1,154 @@
|
|||
import { describe, expect, it, type Mock, vi } from "vitest";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import type { AppServerQueryClientRunner } from "../../../../src/app-server/query/cache";
|
||||
import type { AppServerQueryContext, AppServerQueryContextIdentity } from "../../../../src/app-server/query/keys";
|
||||
import type { ObservedResult, ObservedResultListener } from "../../../../src/app-server/query/observed-result";
|
||||
import { AppServerResourceStore } from "../../../../src/app-server/query/resource-store";
|
||||
import type { AppServerQueryContextIdentity } from "../../../../src/app-server/query/keys";
|
||||
import type { ObservedResultListener } from "../../../../src/app-server/query/observed-result";
|
||||
import { applyThreadListMutation, type ThreadListMutation } from "../../../../src/app-server/query/thread-list-mutation";
|
||||
import type { Thread } from "../../../../src/domain/threads/model";
|
||||
import { createThreadCatalog, type ThreadCatalog } from "../../../../src/features/threads/catalog/thread-catalog";
|
||||
import { createThreadCatalog } from "../../../../src/features/threads/catalog/thread-catalog";
|
||||
|
||||
describe("ThreadCatalog", () => {
|
||||
it("applies rename mutations after updating the catalog cache", () => {
|
||||
const { catalog } = catalogFixture();
|
||||
receiveActive(catalog, [thread("thread"), thread("other")]);
|
||||
|
||||
catalog.apply({ type: "thread-renamed", threadId: "thread", name: "Renamed" });
|
||||
|
||||
expect(catalog.activeSnapshot()).toEqual([{ ...thread("thread"), name: "Renamed" }, thread("other")]);
|
||||
});
|
||||
|
||||
it("applies archive mutations after updating catalog membership", () => {
|
||||
const { catalog } = catalogFixture();
|
||||
const listener = vi.fn();
|
||||
const archivedListener = vi.fn();
|
||||
catalog.observeActive(listener);
|
||||
catalog.observeArchived(archivedListener);
|
||||
receiveActive(catalog, [thread("thread"), thread("other")]);
|
||||
receiveArchived(catalog, [thread("archived", true)]);
|
||||
|
||||
catalog.apply({ type: "thread-archived", threadId: "thread" });
|
||||
|
||||
expect(catalog.activeSnapshot()).toEqual([thread("other")]);
|
||||
expect(catalog.archivedSnapshot()).toEqual([{ ...thread("thread"), archived: true }, thread("archived", true)]);
|
||||
expect(listener).toHaveBeenLastCalledWith(expect.objectContaining({ value: [thread("other")] }));
|
||||
expect(archivedListener).toHaveBeenLastCalledWith(
|
||||
expect.objectContaining({ value: [{ ...thread("thread"), archived: true }, thread("archived", true)] }),
|
||||
);
|
||||
});
|
||||
|
||||
it("refreshes archived snapshots after unknown archive mutations even when an older archived refresh is in flight", async () => {
|
||||
const staleArchivedRefresh = deferred<readonly Thread[]>();
|
||||
const secondArchivedRefreshStarted = deferred<undefined>();
|
||||
let archivedRefreshCount = 0;
|
||||
const fetchThreads = vi.fn((_context: { codexPath: string; vaultPath: string }, archived: boolean) => {
|
||||
if (!archived) return Promise.resolve([]);
|
||||
archivedRefreshCount += 1;
|
||||
if (archivedRefreshCount === 1) return staleArchivedRefresh.promise;
|
||||
secondArchivedRefreshStarted.resolve(undefined);
|
||||
return Promise.resolve([thread("thread", true)]);
|
||||
it("projects rename and archive events through the shared query owner", () => {
|
||||
const store = catalogStore({
|
||||
active: [thread("active"), thread("other")],
|
||||
archived: [thread("archived", true)],
|
||||
});
|
||||
const { catalog } = catalogFixture({ fetchThreads });
|
||||
|
||||
const staleRefresh = catalog.refreshArchived();
|
||||
catalog.apply({ type: "thread-archived", threadId: "thread" });
|
||||
|
||||
staleArchivedRefresh.resolve([thread("old", true)]);
|
||||
await staleRefresh;
|
||||
await secondArchivedRefreshStarted.promise;
|
||||
|
||||
await vi.waitFor(() => {
|
||||
expect(catalog.archivedSnapshot()).toEqual([thread("thread", true)]);
|
||||
});
|
||||
expect(fetchThreads).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("applies known delete mutations to cache", () => {
|
||||
const { catalog } = catalogFixture();
|
||||
receiveActive(catalog, [thread("thread"), thread("other")]);
|
||||
receiveArchived(catalog, [thread("thread", true), thread("archived", true)]);
|
||||
|
||||
catalog.apply({ type: "thread-deleted", threadId: "thread" });
|
||||
|
||||
expect(catalog.activeSnapshot()).toEqual([thread("other")]);
|
||||
expect(catalog.archivedSnapshot()).toEqual([thread("archived", true)]);
|
||||
});
|
||||
|
||||
it("records started, forked, and restored thread membership", () => {
|
||||
const { catalog } = catalogFixture();
|
||||
receiveActive(catalog, [thread("existing")]);
|
||||
receiveArchived(catalog, [thread("restored", true), thread("archived", true)]);
|
||||
|
||||
catalog.apply({ type: "thread-started", thread: thread("started") });
|
||||
catalog.apply({ type: "thread-forked", thread: thread("forked") });
|
||||
catalog.apply({ type: "thread-restored", thread: thread("restored") });
|
||||
|
||||
expect(catalog.activeSnapshot()?.map((item) => item.id)).toEqual(["restored", "forked", "started", "existing"]);
|
||||
expect(catalog.archivedSnapshot()?.map((item) => item.id)).toEqual(["archived"]);
|
||||
});
|
||||
|
||||
it("keeps app-server lifecycle threads visible until the server list acknowledges them", async () => {
|
||||
const fetchThreads = vi
|
||||
.fn()
|
||||
.mockResolvedValueOnce([thread("other")])
|
||||
.mockResolvedValueOnce([thread("started"), thread("other")])
|
||||
.mockResolvedValueOnce([thread("other")]);
|
||||
const { catalog } = catalogFixture({ fetchThreads });
|
||||
const listener = vi.fn();
|
||||
catalog.observeActive(listener);
|
||||
|
||||
catalog.apply({ type: "thread-started", thread: thread("started") });
|
||||
|
||||
await expect(catalog.refreshActive()).resolves.toEqual([thread("started"), thread("other")]);
|
||||
expect(catalog.activeSnapshot()).toEqual([thread("started"), thread("other")]);
|
||||
expect(observedActiveThreadIds(listener)).not.toContainEqual(["other"]);
|
||||
|
||||
await expect(catalog.refreshActive()).resolves.toEqual([thread("started"), thread("other")]);
|
||||
await expect(catalog.refreshActive()).resolves.toEqual([thread("other")]);
|
||||
expect(catalog.activeSnapshot()).toEqual([thread("other")]);
|
||||
});
|
||||
|
||||
it("starts a fresh lifecycle overlay for every resource lease", () => {
|
||||
const { catalog, replaceContext } = catalogFixture({ context: () => ({ codexPath: "codex-a", vaultPath: "/vault" }) });
|
||||
catalog.apply({ type: "thread-started", thread: thread("started-a1") });
|
||||
expect(catalog.activeSnapshot()).toEqual([thread("started-a1")]);
|
||||
|
||||
replaceContext({ codexPath: "codex-b", vaultPath: "/vault" });
|
||||
expect(catalog.activeSnapshot()).toBeNull();
|
||||
replaceContext({ codexPath: "codex-a", vaultPath: "/vault" });
|
||||
|
||||
expect(catalog.activeSnapshot()).toBeNull();
|
||||
});
|
||||
|
||||
it("does not publish connection events captured by an earlier lease", () => {
|
||||
const { catalog, contextIdentity, replaceContext } = catalogFixture({
|
||||
context: () => ({ codexPath: "codex-a", vaultPath: "/vault" }),
|
||||
});
|
||||
const sourceA1 = contextIdentity();
|
||||
|
||||
replaceContext({ codexPath: "codex-b", vaultPath: "/vault" });
|
||||
catalog.applyConnectionEvent(sourceA1, { type: "thread-started", thread: thread("stale-a1") });
|
||||
replaceContext({ codexPath: "codex-a", vaultPath: "/vault" });
|
||||
|
||||
expect(catalog.activeSnapshot()).toBeNull();
|
||||
});
|
||||
|
||||
it("preserves raw query status when lifecycle overlays publish", async () => {
|
||||
const refresh = deferred<readonly Thread[]>();
|
||||
const { catalog } = catalogFixture({ fetchThreads: () => refresh.promise });
|
||||
const listener = vi.fn();
|
||||
catalog.observeActive(listener);
|
||||
const refreshing = catalog.refreshActive();
|
||||
await flushMicrotasks();
|
||||
|
||||
catalog.apply({ type: "thread-started", thread: thread("started") });
|
||||
|
||||
expect(listener).toHaveBeenLastCalledWith({ value: [thread("started")], error: null, isFetching: true });
|
||||
refresh.resolve([]);
|
||||
await refreshing;
|
||||
});
|
||||
|
||||
it("keeps rollback fork metadata until active snapshots catch up to the rollback version", () => {
|
||||
const { catalog } = catalogFixture();
|
||||
const forkBeforeRollback = thread("forked", false, { name: "Before", preview: "Before rollback", updatedAt: 20 });
|
||||
const forkAfterRollback = thread("forked", false, { name: "After", preview: "After rollback", updatedAt: 20 });
|
||||
const forkAfterFutureUpdate = thread("forked", false, { name: "Future", preview: "Future update", updatedAt: 21 });
|
||||
receiveActive(catalog, [thread("existing")]);
|
||||
|
||||
catalog.apply({ type: "thread-forked", thread: forkAfterRollback });
|
||||
receiveActive(catalog, [forkBeforeRollback, thread("existing")]);
|
||||
|
||||
expect(catalog.activeSnapshot()).toEqual([forkAfterRollback, thread("existing")]);
|
||||
|
||||
receiveActive(catalog, [forkAfterFutureUpdate, thread("existing")]);
|
||||
|
||||
expect(catalog.activeSnapshot()).toEqual([forkAfterFutureUpdate, thread("existing")]);
|
||||
});
|
||||
|
||||
it("keeps app-server rename facts when an older active list resolves later", async () => {
|
||||
const staleRefresh = deferred<readonly Thread[]>();
|
||||
const fetchThreads = vi
|
||||
.fn()
|
||||
.mockReturnValueOnce(staleRefresh.promise)
|
||||
.mockResolvedValueOnce([{ ...thread("thread"), name: "Renamed" }, thread("other")]);
|
||||
const { catalog } = catalogFixture({ fetchThreads });
|
||||
receiveActive(catalog, [thread("thread"), thread("other")]);
|
||||
|
||||
const refresh = catalog.refreshActive();
|
||||
await flushMicrotasks();
|
||||
catalog.apply({ type: "thread-renamed", threadId: "thread", name: "Renamed" });
|
||||
staleRefresh.resolve([thread("thread"), thread("other")]);
|
||||
|
||||
await expect(refresh).resolves.toEqual([{ ...thread("thread"), name: "Renamed" }, thread("other")]);
|
||||
expect(catalog.activeSnapshot()).toEqual([{ ...thread("thread"), name: "Renamed" }, thread("other")]);
|
||||
|
||||
await expect(catalog.refreshActive()).resolves.toEqual([{ ...thread("thread"), name: "Renamed" }, thread("other")]);
|
||||
expect(catalog.activeSnapshot()).toEqual([{ ...thread("thread"), name: "Renamed" }, thread("other")]);
|
||||
});
|
||||
|
||||
it("keeps app-server removal facts when older active and archived lists resolve later", async () => {
|
||||
const staleActiveRefresh = deferred<readonly Thread[]>();
|
||||
const staleArchivedRefresh = deferred<readonly Thread[]>();
|
||||
const fetchThreads = vi.fn((_context: { codexPath: string; vaultPath: string }, archived: boolean) =>
|
||||
archived ? staleArchivedRefresh.promise : staleActiveRefresh.promise,
|
||||
);
|
||||
const { catalog } = catalogFixture({ fetchThreads });
|
||||
receiveActive(catalog, [thread("active"), thread("other")]);
|
||||
receiveArchived(catalog, [thread("archived", true), thread("kept", true)]);
|
||||
|
||||
const activeRefresh = catalog.refreshActive();
|
||||
const archivedRefresh = catalog.refreshArchived();
|
||||
await flushMicrotasks();
|
||||
const onEventApplied = vi.fn();
|
||||
const catalog = createThreadCatalog({ store, onEventApplied });
|
||||
|
||||
catalog.apply({ type: "thread-renamed", threadId: "active", name: "Renamed" });
|
||||
catalog.apply({ type: "thread-archived", threadId: "active" });
|
||||
catalog.apply({ type: "thread-deleted", threadId: "archived" });
|
||||
staleActiveRefresh.resolve([thread("active"), thread("other")]);
|
||||
staleArchivedRefresh.resolve([thread("archived", true), thread("kept", true)]);
|
||||
|
||||
await expect(activeRefresh).resolves.toEqual([thread("other")]);
|
||||
await expect(archivedRefresh).resolves.toEqual([thread("active", true), thread("kept", true)]);
|
||||
expect(catalog.activeSnapshot()).toEqual([thread("other")]);
|
||||
expect(catalog.archivedSnapshot()).toEqual([thread("active", true), thread("kept", true)]);
|
||||
expect(catalog.archivedSnapshot()).toEqual([{ ...thread("active"), name: "Renamed", archived: true }, thread("archived", true)]);
|
||||
expect(onEventApplied).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("archives unacknowledged active lifecycle threads without waiting for list acknowledgement", async () => {
|
||||
const fetchThreads = vi.fn().mockResolvedValue([thread("other")]);
|
||||
const { catalog } = catalogFixture({ fetchThreads });
|
||||
const listener = vi.fn();
|
||||
const archivedListener = vi.fn();
|
||||
catalog.observeActive(listener);
|
||||
catalog.observeArchived(archivedListener);
|
||||
it("requests an authoritative archived read when an archive event lacks a source snapshot", () => {
|
||||
const store = catalogStore();
|
||||
const catalog = createThreadCatalog({ store });
|
||||
|
||||
catalog.apply({ type: "thread-started", thread: thread("started") });
|
||||
catalog.apply({ type: "thread-archived", threadId: "started" });
|
||||
catalog.apply({ type: "thread-archived", threadId: "unknown" });
|
||||
|
||||
expect(catalog.activeSnapshot()).toEqual([]);
|
||||
expect(catalog.archivedSnapshot()).toEqual([thread("started", true)]);
|
||||
expect(archivedListener).toHaveBeenLastCalledWith(expect.objectContaining({ value: [thread("started", true)] }));
|
||||
|
||||
await expect(catalog.refreshActive()).resolves.toEqual([thread("other")]);
|
||||
expect(catalog.activeSnapshot()).toEqual([thread("other")]);
|
||||
expect(observedActiveThreadIds(listener)).not.toContainEqual(["started", "other"]);
|
||||
});
|
||||
|
||||
it("deletes unacknowledged active lifecycle threads", async () => {
|
||||
const fetchThreads = vi.fn().mockResolvedValue([thread("other")]);
|
||||
const { catalog } = catalogFixture({ fetchThreads });
|
||||
catalog.apply({ type: "thread-started", thread: thread("started") });
|
||||
|
||||
catalog.apply({ type: "thread-deleted", threadId: "started" });
|
||||
|
||||
expect(catalog.activeSnapshot()).toEqual([]);
|
||||
expect(catalog.archivedSnapshot()).toEqual([]);
|
||||
await expect(catalog.refreshActive()).resolves.toEqual([thread("other")]);
|
||||
expect(catalog.activeSnapshot()).toEqual([thread("other")]);
|
||||
});
|
||||
|
||||
it("records active thread touches as catalog ordering facts", () => {
|
||||
const { catalog } = catalogFixture();
|
||||
receiveActive(catalog, [
|
||||
thread("active", false, { updatedAt: 1, recencyAt: 1 }),
|
||||
thread("other", false, { updatedAt: 10, recencyAt: 10 }),
|
||||
expect(store.appliedMutations).toEqual([
|
||||
{ kind: "remove", list: "active", threadId: "unknown" },
|
||||
{ kind: "refresh", list: "archived" },
|
||||
]);
|
||||
});
|
||||
|
||||
it("moves known restored and unarchived threads between lists", () => {
|
||||
const store = catalogStore({
|
||||
active: [thread("active")],
|
||||
archived: [thread("restored", true), thread("unarchived", true)],
|
||||
});
|
||||
const catalog = createThreadCatalog({ store });
|
||||
|
||||
catalog.apply({ type: "thread-restored", thread: thread("restored", true) });
|
||||
catalog.apply({ type: "thread-unarchived", threadId: "unarchived" });
|
||||
|
||||
expect(catalog.activeSnapshot()).toEqual([thread("unarchived"), thread("restored"), thread("active")]);
|
||||
expect(catalog.archivedSnapshot()).toEqual([]);
|
||||
});
|
||||
|
||||
it("patches touch and delete facts without inventing unknown thread records", () => {
|
||||
const store = catalogStore({
|
||||
active: [thread("active", false, { recencyAt: 1 }), 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, { updatedAt: 1, recencyAt: 20 }),
|
||||
thread("other", false, { updatedAt: 10, recencyAt: 10 }),
|
||||
]);
|
||||
});
|
||||
|
||||
it("model-checks stale snapshots around unacknowledged rename and archive facts", () => {
|
||||
const maxEventDepth = 3;
|
||||
|
||||
for (const sequence of eventSequences(renameOrderingEvents(), maxEventDepth)) {
|
||||
const { catalog } = catalogFixture();
|
||||
receiveActive(catalog, [thread("target"), thread("other")]);
|
||||
receiveArchived(catalog, []);
|
||||
catalog.apply({ type: "thread-renamed", threadId: "target", name: "Renamed" });
|
||||
|
||||
let renameAcknowledged = false;
|
||||
for (const event of sequence) {
|
||||
event.apply(catalog);
|
||||
renameAcknowledged = renameAcknowledged || event.acknowledgesRename;
|
||||
}
|
||||
|
||||
if (!renameAcknowledged) {
|
||||
expectTargetName(catalog.activeSnapshot(), "Renamed", sequenceDescription(sequence));
|
||||
expectTargetName(catalog.archivedSnapshot(), "Renamed", sequenceDescription(sequence));
|
||||
}
|
||||
}
|
||||
|
||||
for (const sequence of eventSequences(archiveOrderingEvents(), maxEventDepth)) {
|
||||
const { catalog } = catalogFixture();
|
||||
receiveActive(catalog, [thread("target"), thread("other")]);
|
||||
receiveArchived(catalog, [thread("archived", true)]);
|
||||
catalog.apply({ type: "thread-archived", threadId: "target" });
|
||||
|
||||
let activeRemovalAcknowledged = false;
|
||||
let archivedUpsertAcknowledged = false;
|
||||
for (const event of sequence) {
|
||||
event.apply(catalog);
|
||||
activeRemovalAcknowledged = activeRemovalAcknowledged || event.acknowledgesActiveRemoval;
|
||||
archivedUpsertAcknowledged = archivedUpsertAcknowledged || event.acknowledgesArchivedUpsert;
|
||||
}
|
||||
|
||||
const sequenceName = sequenceDescription(sequence);
|
||||
if (!activeRemovalAcknowledged) expectNoTarget(catalog.activeSnapshot(), sequenceName);
|
||||
if (!archivedUpsertAcknowledged) expectHasArchivedTarget(catalog.archivedSnapshot(), sequenceName);
|
||||
}
|
||||
});
|
||||
|
||||
it("moves known unarchived threads through the catalog and refreshes unknown unarchives", async () => {
|
||||
const unknownActiveRefreshStarted = deferred<undefined>();
|
||||
const unknownArchivedRefreshStarted = deferred<undefined>();
|
||||
const fetchThreads = vi.fn((_context: { codexPath: string; vaultPath: string }, archived: boolean) => {
|
||||
if (archived) {
|
||||
unknownArchivedRefreshStarted.resolve(undefined);
|
||||
return Promise.resolve([]);
|
||||
}
|
||||
unknownActiveRefreshStarted.resolve(undefined);
|
||||
return Promise.resolve([thread("unknown")]);
|
||||
});
|
||||
const { catalog } = catalogFixture({ fetchThreads });
|
||||
receiveActive(catalog, [thread("active")]);
|
||||
receiveArchived(catalog, [thread("known", true)]);
|
||||
|
||||
catalog.apply({ type: "thread-unarchived", threadId: "known" });
|
||||
|
||||
expect(catalog.activeSnapshot()).toEqual([thread("known"), thread("active")]);
|
||||
expect(catalog.activeSnapshot()).toEqual([thread("active", false, { recencyAt: 20 }), thread("kept")]);
|
||||
expect(catalog.archivedSnapshot()).toEqual([]);
|
||||
});
|
||||
|
||||
catalog.apply({ type: "thread-unarchived", threadId: "unknown" });
|
||||
it("ignores connection events from an earlier resource lease", () => {
|
||||
const store = catalogStore();
|
||||
const catalog = createThreadCatalog({ store });
|
||||
const staleContext = { codexPath: "codex", vaultPath: "/vault", generation: 0 };
|
||||
|
||||
await unknownActiveRefreshStarted.promise;
|
||||
await unknownArchivedRefreshStarted.promise;
|
||||
await vi.waitFor(() => {
|
||||
expect(catalog.activeSnapshot()).toEqual([thread("known"), thread("unknown")]);
|
||||
});
|
||||
expect(fetchThreads).toHaveBeenCalledTimes(2);
|
||||
catalog.applyConnectionEvent(staleContext, { type: "thread-started", thread: thread("stale") });
|
||||
|
||||
expect(store.appliedMutations).toEqual([]);
|
||||
expect(catalog.activeSnapshot()).toBeNull();
|
||||
});
|
||||
|
||||
it("delegates reads, pagination, and observation without a second projection store", async () => {
|
||||
const store = catalogStore({ active: [thread("active")], archived: [thread("archived", true)] });
|
||||
const catalog = createThreadCatalog({ store });
|
||||
const activeObserver = vi.fn();
|
||||
const archivedObserver = vi.fn();
|
||||
|
||||
expect(await catalog.loadActive()).toEqual([thread("active")]);
|
||||
expect(await catalog.refreshActive()).toEqual([thread("active")]);
|
||||
expect(catalog.hasMoreActive()).toBe(false);
|
||||
expect(await catalog.loadMoreActive()).toEqual([thread("active")]);
|
||||
expect(await catalog.refreshArchived()).toEqual([thread("archived", true)]);
|
||||
|
||||
const unsubscribeActive = catalog.observeActive(activeObserver);
|
||||
const unsubscribeArchived = catalog.observeArchived(archivedObserver);
|
||||
expect(activeObserver).toHaveBeenCalledWith({ value: [thread("active")], error: null, isFetching: false });
|
||||
expect(archivedObserver).toHaveBeenCalledWith({ value: [thread("archived", true)], error: null, isFetching: false });
|
||||
unsubscribeActive();
|
||||
unsubscribeArchived();
|
||||
});
|
||||
});
|
||||
|
||||
interface ModeledCatalogEvent {
|
||||
readonly name: string;
|
||||
readonly acknowledgesRename: boolean;
|
||||
readonly acknowledgesActiveRemoval: boolean;
|
||||
readonly acknowledgesArchivedUpsert: boolean;
|
||||
apply(catalog: ThreadCatalog): void;
|
||||
interface CatalogStoreOptions {
|
||||
readonly active?: readonly Thread[] | null;
|
||||
readonly archived?: readonly Thread[] | null;
|
||||
}
|
||||
|
||||
function renameOrderingEvents(): readonly ModeledCatalogEvent[] {
|
||||
return [
|
||||
modeledCatalogEvent("stale active snapshot", (catalog) => {
|
||||
receiveActive(catalog, [thread("target"), thread("other")]);
|
||||
}),
|
||||
modeledCatalogEvent(
|
||||
"rename-ack active snapshot",
|
||||
(catalog) => {
|
||||
receiveActive(catalog, [{ ...thread("target"), name: "Renamed" }, thread("other")]);
|
||||
},
|
||||
{ acknowledgesRename: true },
|
||||
),
|
||||
modeledCatalogEvent("empty archived snapshot", (catalog) => {
|
||||
receiveArchived(catalog, []);
|
||||
}),
|
||||
];
|
||||
}
|
||||
|
||||
function archiveOrderingEvents(): readonly ModeledCatalogEvent[] {
|
||||
return [
|
||||
modeledCatalogEvent("stale active snapshot", (catalog) => {
|
||||
receiveActive(catalog, [thread("target"), thread("other")]);
|
||||
}),
|
||||
modeledCatalogEvent(
|
||||
"archive-ack active snapshot",
|
||||
(catalog) => {
|
||||
receiveActive(catalog, [thread("other")]);
|
||||
},
|
||||
{ acknowledgesActiveRemoval: true },
|
||||
),
|
||||
modeledCatalogEvent("stale archived snapshot", (catalog) => {
|
||||
receiveArchived(catalog, [thread("archived", true)]);
|
||||
}),
|
||||
modeledCatalogEvent(
|
||||
"archive-ack archived snapshot",
|
||||
(catalog) => {
|
||||
receiveArchived(catalog, [thread("target", true), thread("archived", true)]);
|
||||
},
|
||||
{ acknowledgesArchivedUpsert: true },
|
||||
),
|
||||
];
|
||||
}
|
||||
|
||||
function modeledCatalogEvent(
|
||||
name: string,
|
||||
apply: (catalog: ThreadCatalog) => void,
|
||||
acknowledgements: Partial<
|
||||
Pick<ModeledCatalogEvent, "acknowledgesRename" | "acknowledgesActiveRemoval" | "acknowledgesArchivedUpsert">
|
||||
> = {},
|
||||
): ModeledCatalogEvent {
|
||||
return {
|
||||
name,
|
||||
apply,
|
||||
acknowledgesRename: acknowledgements.acknowledgesRename ?? false,
|
||||
acknowledgesActiveRemoval: acknowledgements.acknowledgesActiveRemoval ?? false,
|
||||
acknowledgesArchivedUpsert: acknowledgements.acknowledgesArchivedUpsert ?? false,
|
||||
function catalogStore(options: CatalogStoreOptions = {}) {
|
||||
let active = options.active ?? null;
|
||||
let archived = options.archived ?? null;
|
||||
const activeObservers = new Set<ObservedResultListener<readonly Thread[]>>();
|
||||
const archivedObservers = new Set<ObservedResultListener<readonly Thread[]>>();
|
||||
const appliedMutations: ThreadListMutation[] = [];
|
||||
const currentContext: AppServerQueryContextIdentity = {
|
||||
codexPath: "codex",
|
||||
vaultPath: "/vault",
|
||||
generation: 1,
|
||||
};
|
||||
}
|
||||
|
||||
function eventSequences<T>(events: readonly T[], maxDepth: number): T[][] {
|
||||
const sequences: T[][] = [[]];
|
||||
for (let depth = 1; depth <= maxDepth; depth += 1) {
|
||||
for (const prefix of sequences.filter((sequence) => sequence.length === depth - 1)) {
|
||||
for (const event of events) {
|
||||
sequences.push([...prefix, event]);
|
||||
const applyMutations = (mutations: readonly ThreadListMutation[]): void => {
|
||||
appliedMutations.push(...mutations);
|
||||
for (const mutation of mutations) {
|
||||
if (mutation.list === "active") {
|
||||
active = applyThreadListMutation(active, mutation);
|
||||
} else {
|
||||
archived = applyThreadListMutation(archived, mutation);
|
||||
}
|
||||
}
|
||||
}
|
||||
return sequences;
|
||||
}
|
||||
|
||||
function expectTargetName(threads: readonly Thread[] | null, expectedName: string, sequence: string): void {
|
||||
const target = threads?.find((item) => item.id === "target") ?? null;
|
||||
if (!target) return;
|
||||
expect(target.name, sequence).toBe(expectedName);
|
||||
}
|
||||
|
||||
function expectNoTarget(threads: readonly Thread[] | null, sequence: string): void {
|
||||
expect(threads?.some((item) => item.id === "target") ?? false, sequence).toBe(false);
|
||||
}
|
||||
|
||||
function expectHasArchivedTarget(threads: readonly Thread[] | null, sequence: string): void {
|
||||
expect(threads?.some((item) => item.id === "target" && item.archived) ?? false, sequence).toBe(true);
|
||||
}
|
||||
|
||||
function sequenceDescription(sequence: readonly ModeledCatalogEvent[]): string {
|
||||
return sequence.length === 0 ? "no additional snapshots" : sequence.map((event) => event.name).join(" -> ");
|
||||
}
|
||||
|
||||
const catalogStores = new WeakMap<ThreadCatalog, TestThreadCatalogStore>();
|
||||
|
||||
function receiveActive(catalog: ThreadCatalog, threads: readonly Thread[]): void {
|
||||
catalogStores.get(catalog)?.receiveActive(threads);
|
||||
}
|
||||
|
||||
function receiveArchived(catalog: ThreadCatalog, threads: readonly Thread[]): void {
|
||||
catalogStores.get(catalog)?.receiveArchived(threads);
|
||||
}
|
||||
|
||||
function catalogFixture(
|
||||
options: {
|
||||
fetchThreads?: (context: { codexPath: string; vaultPath: string }, archived: boolean) => Promise<readonly Thread[]>;
|
||||
context?: () => { codexPath: string; vaultPath: string };
|
||||
} = {},
|
||||
) {
|
||||
const context = options.context ?? (() => ({ codexPath: "codex", vaultPath: "/vault" }));
|
||||
const queries = new AppServerResourceStore({
|
||||
clientRunner: clientRunnerWithThreads(options.fetchThreads ?? (() => Promise.resolve([]))),
|
||||
});
|
||||
queries.initialize(context());
|
||||
const store = new TestThreadCatalogStore(queries);
|
||||
const catalog = createThreadCatalog({ store });
|
||||
catalogStores.set(catalog, store);
|
||||
catalog.observeActive(() => undefined);
|
||||
catalog.observeArchived(() => undefined);
|
||||
return {
|
||||
catalog,
|
||||
contextIdentity: () => queries.contextIdentity(),
|
||||
replaceContext: (nextContext: AppServerQueryContext) => queries.replaceContext(nextContext),
|
||||
};
|
||||
}
|
||||
|
||||
class TestThreadCatalogStore {
|
||||
private readonly activeSnapshots = new Map<string, readonly Thread[]>();
|
||||
private readonly archivedSnapshots = new Map<string, readonly Thread[]>();
|
||||
private readonly activeObservers = new Set<ObservedResultListener<readonly Thread[]>>();
|
||||
private readonly archivedObservers = new Set<ObservedResultListener<readonly Thread[]>>();
|
||||
|
||||
constructor(private readonly queries: AppServerResourceStore) {}
|
||||
|
||||
contextKey(): string {
|
||||
return this.queries.contextKey();
|
||||
}
|
||||
|
||||
contextKeyFor(context: AppServerQueryContextIdentity): string {
|
||||
return this.queries.contextKeyFor(context);
|
||||
}
|
||||
|
||||
activeThreadsSnapshot(): readonly Thread[] | null {
|
||||
return this.activeSnapshots.get(this.contextKey()) ?? this.queries.activeThreadsSnapshot();
|
||||
}
|
||||
|
||||
archivedThreadsSnapshot(): readonly Thread[] | null {
|
||||
return this.archivedSnapshots.get(this.contextKey()) ?? this.queries.archivedThreadsSnapshot();
|
||||
}
|
||||
|
||||
async fetchAllActiveThreads(): Promise<readonly Thread[]> {
|
||||
return this.receiveLoadedActive(await this.queries.fetchAllActiveThreads());
|
||||
}
|
||||
|
||||
hasMoreActiveThreads(): boolean {
|
||||
return this.queries.hasMoreActiveThreads();
|
||||
}
|
||||
|
||||
async loadMoreActiveThreads(): Promise<readonly Thread[]> {
|
||||
return this.receiveLoadedActive(await this.queries.loadMoreActiveThreads());
|
||||
}
|
||||
|
||||
async refreshActiveThreads(): Promise<readonly Thread[]> {
|
||||
return this.receiveLoadedActive(await this.queries.refreshActiveThreads());
|
||||
}
|
||||
|
||||
async refreshArchivedThreads(): Promise<readonly Thread[]> {
|
||||
return this.receiveLoadedArchived(await this.queries.refreshArchivedThreads());
|
||||
}
|
||||
|
||||
observeActiveThreadsResult(observer: ObservedResultListener<readonly Thread[]>, options?: { emitCurrent?: boolean }): () => void {
|
||||
this.activeObservers.add(observer);
|
||||
const unsubscribe = this.queries.observeActiveThreadsResult((result) => {
|
||||
if (result.value && !result.isFetching) this.activeSnapshots.delete(this.contextKey());
|
||||
observer(result);
|
||||
}, options);
|
||||
return () => {
|
||||
this.activeObservers.delete(observer);
|
||||
unsubscribe();
|
||||
};
|
||||
}
|
||||
|
||||
observeArchivedThreadsResult(observer: ObservedResultListener<readonly Thread[]>, options?: { emitCurrent?: boolean }): () => void {
|
||||
this.archivedObservers.add(observer);
|
||||
const unsubscribe = this.queries.observeArchivedThreadsResult((result) => {
|
||||
if (result.value && !result.isFetching) this.archivedSnapshots.delete(this.contextKey());
|
||||
observer(result);
|
||||
}, options);
|
||||
return () => {
|
||||
this.archivedObservers.delete(observer);
|
||||
unsubscribe();
|
||||
};
|
||||
}
|
||||
|
||||
receiveActive(threads: readonly Thread[]): void {
|
||||
this.activeSnapshots.set(this.contextKey(), threads);
|
||||
const result = observedSnapshot(threads);
|
||||
for (const observer of this.activeObservers) observer(result);
|
||||
}
|
||||
|
||||
receiveArchived(threads: readonly Thread[]): void {
|
||||
this.archivedSnapshots.set(this.contextKey(), threads);
|
||||
const result = observedSnapshot(threads);
|
||||
for (const observer of this.archivedObservers) observer(result);
|
||||
}
|
||||
|
||||
private receiveLoadedActive(threads: readonly Thread[]): readonly Thread[] {
|
||||
this.activeSnapshots.delete(this.contextKey());
|
||||
return threads;
|
||||
}
|
||||
|
||||
private receiveLoadedArchived(threads: readonly Thread[]): readonly Thread[] {
|
||||
this.archivedSnapshots.delete(this.contextKey());
|
||||
return threads;
|
||||
}
|
||||
}
|
||||
|
||||
function observedSnapshot(threads: readonly Thread[]): ObservedResult<readonly Thread[]> {
|
||||
return { value: threads, error: null, isFetching: false };
|
||||
}
|
||||
|
||||
function clientRunnerWithThreads(
|
||||
fetchThreads: (context: { codexPath: string; vaultPath: string }, archived: boolean) => Promise<readonly Thread[]>,
|
||||
): AppServerQueryClientRunner {
|
||||
return {
|
||||
runWithClient: async (context, 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),
|
||||
nextCursor: null,
|
||||
};
|
||||
},
|
||||
} as never);
|
||||
appliedMutations,
|
||||
contextKey: () => contextKey(currentContext),
|
||||
contextKeyFor: contextKey,
|
||||
activeThreadsSnapshot: () => active,
|
||||
archivedThreadsSnapshot: () => archived,
|
||||
fetchAllActiveThreads: async () => active ?? [],
|
||||
hasMoreActiveThreads: () => false,
|
||||
loadMoreActiveThreads: async () => active ?? [],
|
||||
refreshActiveThreads: async () => active ?? [],
|
||||
refreshArchivedThreads: async () => archived ?? [],
|
||||
applyThreadListMutations: applyMutations,
|
||||
observeActiveThreadsResult: (observer: ObservedResultListener<readonly Thread[]>, observeOptions: { emitCurrent?: boolean } = {}) => {
|
||||
activeObservers.add(observer);
|
||||
if (observeOptions.emitCurrent ?? true) observer({ value: active, error: null, isFetching: false });
|
||||
return () => activeObservers.delete(observer);
|
||||
},
|
||||
observeArchivedThreadsResult: (observer: ObservedResultListener<readonly Thread[]>, observeOptions: { emitCurrent?: boolean } = {}) => {
|
||||
archivedObservers.add(observer);
|
||||
if (observeOptions.emitCurrent ?? true) observer({ value: archived, error: null, isFetching: false });
|
||||
return () => archivedObservers.delete(observer);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function observedActiveThreadIds(listener: Mock): string[][] {
|
||||
return listener.mock.calls
|
||||
.map((call) => {
|
||||
const result = call[0] as { value: readonly Thread[] | null };
|
||||
return result.value?.map((item) => item.id) ?? null;
|
||||
})
|
||||
.filter((ids): ids is string[] => ids !== null);
|
||||
function contextKey(context: AppServerQueryContextIdentity): string {
|
||||
return `${String(context.generation)}:${context.codexPath}:${context.vaultPath}`;
|
||||
}
|
||||
|
||||
function thread(id: string, archived = false, overrides: Partial<Thread> = {}): Thread {
|
||||
|
|
@ -594,23 +163,3 @@ function thread(id: string, archived = false, overrides: Partial<Thread> = {}):
|
|||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function deferred<T>(): {
|
||||
promise: Promise<T>;
|
||||
resolve: (value: T) => void;
|
||||
reject: (reason?: unknown) => void;
|
||||
} {
|
||||
let resolve!: (value: T) => void;
|
||||
let reject!: (reason?: unknown) => void;
|
||||
const promise = new Promise<T>((promiseResolve, promiseReject) => {
|
||||
resolve = promiseResolve;
|
||||
reject = promiseReject;
|
||||
});
|
||||
return { promise, resolve, reject };
|
||||
}
|
||||
|
||||
async function flushMicrotasks(): Promise<void> {
|
||||
for (let index = 0; index < 10; index += 1) {
|
||||
await Promise.resolve();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue