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(threads): coordinate fork publication atomically

Browse source
This commit is contained in:
murashit 2026-07-22 11:19:13 +09:00
parent bb673b2a75
commit 05ea87643a
37 changed files with 981 additions and 465 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
docs/design.md
View file

@ -56,6 +56,8 @@ Chat-visible state should have one authoritative owner. Components should consum
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.
Thread-list updates have three explicit stages. `ThreadOperationCoordinator` orders lifecycle facts across multi-step operations such as fork without knowing list mutation details. The pure read-model projector converts each committed fact sequence into one mutation batch, and the shared app-server query owner applies that batch while structurally providing narrow `ThreadCatalog` read ports to consumers. Ordinary app-server notifications and completed local operations use the same path, while RPC sequencing and fork-specific buffering remain outside both the projector and query owner.
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.

23
src/execution-runtime.ts
View file

@ -16,7 +16,13 @@ import { createAppServerSelectionRewriteTransport } from "./features/selection-r
import type { SelectionRewriteTransport } from "./features/selection-rewrite/transport";
import { openThreadPicker, type ThreadPickerController } from "./features/thread-picker/modal.obsidian";
import { createThreadOperationsTransport, createThreadTitleTransport } from "./features/threads/app-server/workflow-transports";
import { createThreadCatalog, type ThreadCatalog, type ThreadCatalogEvent } from "./features/threads/catalog/thread-catalog";
import type { ThreadCatalog } from "./features/threads/catalog/thread-catalog";
import {
createThreadOperationCoordinator,
type ThreadOperationCoordinator,
} from "./features/threads/workflows/thread-operation-coordinator";
import type { ThreadLifecycleEvent } from "./features/threads/workflows/thread-operation-event";
import { projectThreadListChanges } from "./features/threads/workflows/thread-read-model-projection";
import type { ThreadsViewHost, ThreadsViewSettingsAccess } from "./features/threads-view/session";
import type { ThreadsViewPanelActivity } from "./features/threads-view/state";
import type { ThreadsRuntimeView } from "./features/threads-view/view.obsidian";
@ -31,7 +37,7 @@ export interface CodexExecutionRuntimeOptions {
context: AppServerExecutionContext;
settings: () => CodexPanelSettings;
workspace: WorkspacePanels;
onThreadCatalogEvent(event: ThreadCatalogEvent): void;
onThreadLifecycleEvents(events: readonly ThreadLifecycleEvent[]): void;
openNewPanel(): Promise<unknown>;
openThreadInCurrentView(threadId: string): Promise<void>;
openThreadInAvailableView(threadId: string): Promise<void>;
@ -47,6 +53,7 @@ export class CodexExecutionRuntime implements AppServerClientAccess {
private readonly context: Readonly<AppServerExecutionContext>;
private readonly appServerQueries: AppServerQueryCache;
private readonly threadCatalog: ThreadCatalog;
private readonly threadOperationCoordinator: ThreadOperationCoordinator;
readonly settingsDynamicData: SettingsDynamicDataAccess;
private readonly threadNameMutations = createKeyedOperationQueue<string>();
private readonly threadGoalOperations = createThreadGoalOperationCoordinator();
@ -62,17 +69,17 @@ export class CodexExecutionRuntime implements AppServerClientAccess {
constructor(private readonly options: CodexExecutionRuntimeOptions) {
this.context = Object.freeze({ ...options.context });
this.appServerQueries = new AppServerQueryCache(this.context, this);
this.threadCatalog = createThreadCatalog({
store: this.appServerQueries,
onEventApplied: (event) => {
options.onThreadCatalogEvent(event);
},
this.threadCatalog = this.appServerQueries;
this.threadOperationCoordinator = createThreadOperationCoordinator((events) => {
this.threadCatalog.applyThreadListMutations(projectThreadListChanges(this.threadCatalog, events));
options.onThreadLifecycleEvents(events);
});
this.settingsDynamicData = createSettingsAppServerDynamicData({
vaultPath: this.context.vaultPath,
clientAccess: this,
appServerQueries: this.appServerQueries,
threadCatalog: this.threadCatalog,
threadEvents: this.threadOperationCoordinator,
});
}
@ -85,6 +92,7 @@ export class CodexExecutionRuntime implements AppServerClientAccess {
workspace: this.options.workspace,
appServerQueries: this.appServerQueries,
threadCatalog: this.threadCatalog,
threadOperationCoordinator: this.threadOperationCoordinator,
threadNameMutations: this.threadNameMutations,
threadTitleTransport: this.threadTitleTransport(),
threadGoalOperations: this.threadGoalOperations,
@ -98,6 +106,7 @@ export class CodexExecutionRuntime implements AppServerClientAccess {
settings: this.threadsSettings(),
vaultPath: this.context.vaultPath,
threadCatalog: this.threadCatalog,
threadEvents: this.threadOperationCoordinator,
threadNameMutations: this.threadNameMutations,
threadOperationsTransport: createThreadOperationsTransport(this),
threadTitleTransport: this.threadTitleTransport(),

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

@ -14,7 +14,7 @@ import {
type PendingUserInput,
} from "../../../../domain/pending-requests/model";
import type { TurnTranscriptSummary } from "../../../../domain/threads/transcript";
import type { ThreadCatalogEvent } from "../../../threads/catalog/thread-catalog";
import type { ThreadOperationEvent } from "../../../threads/workflows/thread-operation-event";
import type { AppServerResourceEvent } from "../../application/connection/server-metadata-actions";
import type { LocalIdSource } from "../../application/local-id-source";
import { activeThreadId, type ChatAction, type ChatState } from "../../application/state/root-reducer";
@ -34,7 +34,7 @@ export interface ChatInboundHandlerActions {
refreshServerDiagnostics: (options?: { forceResourceProbes?: boolean }) => void;
applyAppServerResourceEvent: (event: AppServerResourceEvent) => void;
maybeNameThread: (threadId: string, turnId: string, completedTurnTranscriptSummary: TurnTranscriptSummary | null) => void;
applyThreadCatalogEvent: (event: ThreadCatalogEvent) => void;
applyThreadOperationEvent: (event: ThreadOperationEvent) => void;
respondToServerRequest: (requestId: RequestId, result: unknown) => boolean;
rejectServerRequest: (requestId: RequestId, code: number, message: string) => boolean;
}
@ -248,8 +248,8 @@ function runNotificationEffect(context: ChatInboundHandlerContext, effect: ChatN
case "maybe-name-thread":
context.actions.maybeNameThread(effect.threadId, effect.turnId, effect.completedTurnTranscriptSummary);
return;
case "apply-thread-catalog-event":
context.actions.applyThreadCatalogEvent(effect.event);
case "apply-thread-operation-event":
context.actions.applyThreadOperationEvent(effect.event);
return;
}
}

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

@ -2,7 +2,7 @@ import type { ServerNotification } from "../../../../app-server/connection/rpc-m
import { threadFromAppServerRecord } from "../../../../app-server/services/threads";
import { threadTokenUsageFromRuntimeUsage } from "../../../../domain/runtime/metrics";
import { normalizeExplicitThreadName } from "../../../../domain/threads/model";
import type { ThreadCatalogEvent } from "../../../threads/catalog/thread-catalog";
import type { ThreadOperationEvent } from "../../../threads/workflows/thread-operation-event";
import type { AppServerResourceEvent } from "../../application/connection/server-metadata-actions";
import { activeThreadSettingsAppliedAction } from "../../application/state/actions";
import { activeThreadId, activeThreadState, type ChatAction, type ChatState } from "../../application/state/root-reducer";
@ -22,7 +22,7 @@ export type ChatNotificationEffect =
}
| { type: "refresh-server-diagnostics"; forceResourceProbes?: boolean }
| { type: "apply-app-server-resource-event"; event: AppServerResourceEvent }
| { type: "apply-thread-catalog-event"; event: ThreadCatalogEvent };
| { type: "apply-thread-operation-event"; event: ThreadOperationEvent };
type TurnRuntimeCompletedTurnTranscriptSummary = TurnRuntimeOutcome["completedTurnTranscriptSummary"];
@ -291,17 +291,23 @@ function planThreadLifecycle(
case "thread/started":
return threadStartedPlan(state, notification);
case "thread/archived":
return effectPlan({ type: "apply-thread-catalog-event", event: { type: "thread-archived", threadId: notification.params.threadId } });
return effectPlan({
type: "apply-thread-operation-event",
event: { type: "thread-archived", threadId: notification.params.threadId },
});
case "thread/deleted":
return effectPlan({ type: "apply-thread-catalog-event", event: { type: "thread-deleted", threadId: notification.params.threadId } });
return effectPlan({
type: "apply-thread-operation-event",
event: { type: "thread-deleted", threadId: notification.params.threadId },
});
case "thread/unarchived":
return effectPlan({
type: "apply-thread-catalog-event",
type: "apply-thread-operation-event",
event: { type: "thread-unarchived", threadId: notification.params.threadId },
});
case "thread/name/updated":
return effectPlan({
type: "apply-thread-catalog-event",
type: "apply-thread-operation-event",
event: {
type: "thread-renamed",
threadId: notification.params.threadId,
@ -333,8 +339,12 @@ function threadStartedPlan(
? []
: [
{
type: "apply-thread-catalog-event",
event: { type: "thread-upserted", thread },
type: "apply-thread-operation-event",
event: {
type: "thread-upserted",
thread,
forkedFromThreadId: notification.params.thread.forkedFromId,
},
},
];
return { actions: trackAction, effects };

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

@ -23,10 +23,15 @@ export interface ThreadManagementActionsHost {
setComposerText: (text: string) => void;
openThreadInNewView: (threadId: string) => Promise<void>;
openThreadInCurrentPanel: (threadId: string, onAdopted: () => void) => Promise<{ adopted: boolean }>;
recordThread: (thread: Thread) => void;
beginThreadForkPublication: (sourceThreadId: string) => ThreadForkPublication;
threadHasPendingOrRunningPanel: (threadId: string) => boolean;
}
interface ThreadForkPublication {
record(thread: Thread): void;
finish(options?: { sourceArchived?: boolean }): void;
}
interface ThreadManagementOperations {
renameThread(threadId: string, value: string): Promise<boolean>;
archiveThread(threadId: string, options?: { saveMarkdown?: boolean }): Promise<boolean>;
@ -131,18 +136,21 @@ async function forkThreadFromTurn(
host.addSystemMessage("Could not find the selected turn to fork.");
return;
}
let publication: ThreadForkPublication | null = null;
let publicationFinished = false;
try {
const sourceName = inheritedForkThreadName(threadId, threadManagementState(host).threadList.listedThreads);
if (!(await host.threadTransport.ensureConnected())) return;
if (!threadManagementScopeStillTargetsOriginalPanel(host, scope)) return;
publication = host.beginThreadForkPublication(threadId);
const effect = turnId
? await host.threadTransport.forkThread(threadId, { position: { kind: "through-turn", turnId } })
: await host.threadTransport.forkThread(threadId);
if (!effectCompleted(effect)) return;
const forkedThread = effect.value;
const forkedThreadId = forkedThread.id;
host.recordThread(forkedThread);
publication.record(forkedThread);
if (!effectCompletedInCurrentContext(effect)) return;
if (!threadManagementScopeStillTargetsOriginalPanel(host, scope)) return;
if (sourceName) {
@ -171,9 +179,11 @@ async function forkThreadFromTurn(
}
return;
}
await archiveReplacedSource(host, threadId, forkedThreadId, {
const sourceArchived = await archiveReplacedSource(host, threadId, forkedThreadId, {
failureMessage: "Forked the thread, but could not archive the previous version",
});
publication.finish({ sourceArchived });
publicationFinished = true;
return;
}
try {
@ -186,6 +196,8 @@ async function forkThreadFromTurn(
} catch (error) {
if (!threadManagementScopeStillTargetsOriginalPanel(host, scope)) return;
host.addSystemMessage(error instanceof Error ? error.message : String(error));
} finally {
if (!publicationFinished) publication?.finish();
}
}
@ -226,11 +238,14 @@ async function rollbackThread(host: ThreadManagementActionsHost, threadId: strin
? { sandboxPolicy: runtime.sandboxPolicy }
: {}),
};
let publication: ThreadForkPublication | null = null;
let publicationFinished = false;
try {
host.setStatus(STATUS_ROLLBACK_STARTING);
if (!(await host.threadTransport.ensureConnected())) return;
if (!threadManagementScopeStillTargetsPanel(host, scope)) return;
publication = host.beginThreadForkPublication(threadId);
const effect = await host.threadTransport.forkThread(threadId, {
position: { kind: "before-turn", turnId: candidate.turnId },
deferGoalContinuation: true,
@ -238,7 +253,7 @@ async function rollbackThread(host: ThreadManagementActionsHost, threadId: strin
});
if (!effectCompleted(effect)) return;
const forkedThread = effect.value;
if (effectCompletedInCurrentContext(effect)) host.recordThread(forkedThread);
if (effectCompletedInCurrentContext(effect)) publication.record(forkedThread);
else return;
if (!threadManagementScopeStillTargetsPanel(host, scope)) return;
const adoption = await host.openThreadInCurrentPanel(forkedThread.id, () => {
@ -255,14 +270,18 @@ 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);
}
await archiveReplacedSource(host, threadId, forkedThread.id, {
const sourceArchived = await archiveReplacedSource(host, threadId, forkedThread.id, {
saveMarkdown: false,
failureMessage: "Rolled back the latest turn, but could not archive the previous version",
});
publication.finish({ sourceArchived });
publicationFinished = true;
} catch (error) {
if (!threadManagementScopeStillTargetsPanel(host, scope)) return;
host.addSystemMessage(error instanceof Error ? error.message : String(error));
host.setStatus(STATUS_ROLLBACK_FAILED);
} finally {
if (!publicationFinished) publication?.finish();
}
}
@ -271,15 +290,16 @@ async function archiveReplacedSource(
sourceThreadId: string,
replacementThreadId: string,
options: { readonly saveMarkdown?: boolean; readonly failureMessage: string },
): Promise<void> {
): Promise<boolean> {
try {
const archiveOptions = options.saveMarkdown === undefined ? {} : { saveMarkdown: options.saveMarkdown };
if (await host.operations.archiveThread(sourceThreadId, archiveOptions)) return;
if (await host.operations.archiveThread(sourceThreadId, archiveOptions)) return true;
reportReplacementArchiveFailure(host, replacementThreadId, options.failureMessage, "archive was not completed");
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
reportReplacementArchiveFailure(host, replacementThreadId, options.failureMessage, message);
}
return false;
}
function reportReplacementArchiveFailure(

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

@ -135,7 +135,7 @@ export function createConnectionBundle(
appServerMetadataSnapshot: () => environment.plugin.appServerQueries.appServerMetadataSnapshot(),
});
const refreshSharedThreads = async (): Promise<void> => {
await environment.plugin.threadCatalog.refreshActive();
await environment.plugin.threadCatalog.refreshActiveThreads();
};
const inboundHandler = createChatInboundHandler(
stateStore,
@ -153,8 +153,8 @@ export function createConnectionBundle(
maybeNameThread: (threadId, turnId, completedTurnTranscriptSummary) => {
autoTitleCoordinator.maybeAutoTitleThread(threadId, turnId, completedTurnTranscriptSummary);
},
applyThreadCatalogEvent: (event) => {
environment.plugin.threadCatalog.apply(event);
applyThreadOperationEvent: (event) => {
environment.plugin.threadOperationCoordinator.apply(event);
},
respondToServerRequest: (requestId, result) => serverRequestResponders.respond(requestId, result),
rejectServerRequest: (requestId, code, message) => serverRequestResponders.reject(requestId, code, message),

2
src/features/chat/host/bundles/shell-bundle.ts
View file

@ -68,7 +68,7 @@ export function createShellBundle(host: ChatPanelShellBundleHost, input: ChatPan
toolbarPanel: toolbarPanelActions,
rename,
navigation,
loadMoreThreads: () => environment.plugin.threadCatalog.loadMoreActive(),
loadMoreThreads: () => environment.plugin.threadCatalog.loadMoreActiveThreads(),
openSideChat: () => {
const state = stateStore.getState();
if (activePanelOperationDecision(state, "start-side-chat").kind !== "allowed") return;

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

@ -127,7 +127,7 @@ export function createThreadFoundation(host: ChatPanelThreadHost, input: ChatPan
vaultConfigDir: environment.obsidian.app.vault.configDir,
},
archiveDestination: environment.obsidian.archiveDestination,
catalog: environment.plugin.threadCatalog,
operationEvents: environment.plugin.threadOperationCoordinator,
referenceThreads: () => stateStore.getState().threadList.listedThreads,
notice: (text) => {
new Notice(text);
@ -281,9 +281,7 @@ export function createThreadActionBundle(host: ChatPanelThreadHost, input: ChatP
});
return { adopted };
},
recordThread: (thread) => {
environment.plugin.threadCatalog.apply({ type: "thread-upserted", thread });
},
beginThreadForkPublication: (sourceThreadId) => environment.plugin.threadOperationCoordinator.beginForkPublication(sourceThreadId),
threadHasPendingOrRunningPanel: (threadId) => environment.plugin.workspace.threadHasPendingOrRunningPanel(threadId),
};
const actions = createThreadManagementActions(threadManagementHost);
@ -355,7 +353,7 @@ function createSessionThreadLifecycle(
resetThreadTurnPresence,
notifyActiveThreadIdentityChanged,
recordResumedThread: (thread) => {
host.environment.plugin.threadCatalog.apply({ type: "thread-upserted", thread });
host.environment.plugin.threadOperationCoordinator.apply({ type: "thread-upserted", thread });
},
addSystemMessage: status.addSystemMessage,
syncThreadGoal: (threadId) => goalSync.syncThreadGoal(threadId),

6
src/features/chat/host/contracts.ts
View file

@ -7,9 +7,10 @@ import type { ModelMetadata } from "../../../domain/catalog/metadata";
import type { SendShortcut } from "../../../domain/input/send-shortcut";
import type { SharedServerMetadata, SharedServerMetadataResource } from "../../../domain/server/metadata";
import type { KeyedOperationQueue } from "../../../shared/runtime/keyed-operation-queue";
import type { ThreadCatalogEventSink, ThreadCatalogPaginatedActiveReader } from "../../threads/catalog/thread-catalog";
import type { ThreadCatalogPaginatedActiveReader } from "../../threads/catalog/thread-catalog";
import type { ArchiveExportDestination, ArchiveExportSettings } from "../../threads/workflows/archive-export";
import type { ThreadTitleTransport } from "../../threads/workflows/ports";
import type { ThreadOperationCoordinator } from "../../threads/workflows/thread-operation-coordinator";
import type { TurnDiffViewState } from "../../turn-diff/model";
import type { ThreadGoalOperationCoordinator } from "../application/threads/goal-actions";
@ -20,6 +21,7 @@ export interface CodexChatHost {
readonly workspace: WorkspacePanels;
readonly appServerQueries: ChatAppServerQueries;
readonly threadCatalog: ChatThreadCatalog;
readonly threadOperationCoordinator: ThreadOperationCoordinator;
readonly threadNameMutations: KeyedOperationQueue<string>;
readonly threadTitleTransport: ThreadTitleTransport;
readonly threadGoalOperations: ThreadGoalOperationCoordinator;
@ -47,7 +49,7 @@ export interface WorkspacePanels {
openSideChat(sourceThreadId: string, sourceThreadTitle: string | null): Promise<void>;
}
type ChatThreadCatalog = ThreadCatalogPaginatedActiveReader & ThreadCatalogEventSink;
type ChatThreadCatalog = ThreadCatalogPaginatedActiveReader;
interface ChatAppServerQueries {
appServerMetadataSnapshot(): SharedServerMetadata | null;

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

@ -177,7 +177,7 @@ function composeChatPanelSessionRuntime(host: ChatPanelSessionRuntimeHost): Chat
threadStartTransport: appServer.threadStart,
runtimeSnapshotForState: runtimeSnapshotForChatState,
recordStartedThread: (thread) => {
environment.plugin.threadCatalog.apply({ type: "thread-upserted", thread });
environment.plugin.threadOperationCoordinator.apply({ type: "thread-upserted", thread });
},
syncThreadGoal: (threadId) => {
void threadFoundation.goalSync.syncThreadGoal(threadId);

12
src/features/chat/host/session/shared-state-binding.ts
View file

@ -6,9 +6,9 @@ import type { ChatStateStore } from "../../application/state/store";
type ThreadObserver = (result: ObservedPaginatedResult<readonly Thread[]>) => void;
interface SharedStateThreadCatalog {
activeSnapshot(): readonly Thread[] | null;
hasMoreActive(): boolean;
observeActive(observer: ThreadObserver, options?: { emitCurrent?: boolean }): () => void;
activeThreadsSnapshot(): readonly Thread[] | null;
hasMoreActiveThreads(): boolean;
observeActiveThreadsResult(observer: ThreadObserver, options?: { emitCurrent?: boolean }): () => void;
}
interface SharedStateAppServerQueries {
@ -56,12 +56,12 @@ export function createChatPanelSharedStateBinding(options: ChatPanelSharedStateB
}
};
const applyCached = (): void => {
const threads = threadCatalog.activeSnapshot();
const threads = threadCatalog.activeThreadsSnapshot();
if (threads) {
stateStore.dispatch({
type: "thread-list/applied",
threads,
hasMore: threadCatalog.hasMoreActive(),
hasMore: threadCatalog.hasMoreActiveThreads(),
isFetching: false,
isFetchingNextPage: false,
error: null,
@ -75,7 +75,7 @@ export function createChatPanelSharedStateBinding(options: ChatPanelSharedStateB
unsubscribe();
applyCached();
unsubscribers.push(
threadCatalog.observeActive(receiveThreadResult, { emitCurrent: false }),
threadCatalog.observeActiveThreadsResult(receiveThreadResult, { emitCurrent: false }),
appServerQueries.observeAppServerMetadataResources(applyAppServerMetadataResource),
);
},

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

@ -34,11 +34,11 @@ export function openThreadPicker(host: ThreadPickerHost, onClosed: () => void):
};
const loadAndOpen = async (): Promise<void> => {
try {
const recentSnapshot = host.threadCatalog.recentActiveSnapshot();
const loadedThreads = recentSnapshot ?? (await host.threadCatalog.loadActive());
const recentSnapshot = host.threadCatalog.recentActiveThreadsSnapshot();
const loadedThreads = recentSnapshot ?? (await host.threadCatalog.fetchActiveThreads());
if (state.closed) return;
const recentThreads = host.threadCatalog.recentActiveSnapshot() ?? loadedThreads;
if (recentThreads.length === 0 && !host.threadCatalog.hasMoreActive()) {
const recentThreads = host.threadCatalog.recentActiveThreadsSnapshot() ?? loadedThreads;
if (recentThreads.length === 0 && !host.threadCatalog.hasMoreActiveThreads()) {
new Notice("No Codex threads found.");
finish();
return;
@ -152,7 +152,7 @@ class ThreadPickerModal extends SuggestModal<ThreadSuggestion> {
private async loadCompleteThreadList(): Promise<readonly Thread[]> {
if (this.completeThreads) return this.completeThreads;
const pending = this.completeThreadsPromise ?? this.host.threadCatalog.searchActive();
const pending = this.completeThreadsPromise ?? this.host.threadCatalog.fetchActiveThreadSearchInventory();
this.completeThreadsPromise = pending;
try {
this.completeThreads = Object.freeze([...(await pending)]);

22
src/features/threads-view/session.ts
View file

@ -8,9 +8,10 @@ import { DeferredTask } from "../../shared/runtime/deferred-task";
import { isStaleExecutionRuntimeError } from "../../shared/runtime/execution-runtime-lifetime";
import type { KeyedOperationQueue } from "../../shared/runtime/keyed-operation-queue";
import { OwnerLifetime } from "../../shared/runtime/owner-lifetime";
import type { ThreadCatalogEventSink, ThreadCatalogPaginatedActiveReader } from "../threads/catalog/thread-catalog";
import type { ThreadCatalogPaginatedActiveReader } from "../threads/catalog/thread-catalog";
import type { ArchiveExportDestination, ArchiveExportSettings } from "../threads/workflows/archive-export";
import type { ThreadOperationsTransport, ThreadTitleTransport } from "../threads/workflows/ports";
import type { ThreadOperationEventSink } from "../threads/workflows/thread-operation-event";
import { createThreadOperations, type ThreadOperations } from "../threads/workflows/thread-operations";
import { createThreadTitleService, type ThreadTitleService } from "../threads/workflows/thread-title-service";
import { isThreadsArchiveConfirmPointer, renderThreadsViewShell, unmountThreadsViewShell } from "./shell.dom";
@ -19,6 +20,7 @@ export interface ThreadsViewHost {
readonly settings: ThreadsViewSettingsAccess;
readonly vaultPath: string;
readonly threadCatalog: ThreadsViewThreadCatalog;
readonly threadEvents: ThreadOperationEventSink;
readonly threadNameMutations: KeyedOperationQueue<string>;
readonly threadOperationsTransport: ThreadOperationsTransport;
readonly threadTitleTransport: ThreadTitleTransport;
@ -27,7 +29,7 @@ export interface ThreadsViewHost {
openPanelActivities(): readonly ThreadsViewPanelActivity[];
}
type ThreadsViewThreadCatalog = ThreadCatalogPaginatedActiveReader & ThreadCatalogEventSink;
type ThreadsViewThreadCatalog = ThreadCatalogPaginatedActiveReader;
export interface ThreadsViewSettingsAccess {
archiveExportEnabled(): boolean;
@ -83,7 +85,7 @@ export class ThreadsViewSession {
vaultConfigDir: this.environment.vaultConfigDir(),
},
archiveDestination: () => this.environment.archiveDestination(),
catalog: this.host.threadCatalog,
operationEvents: this.host.threadEvents,
referenceThreads: () => this.threads,
notice: (message) => {
new Notice(message);
@ -99,12 +101,12 @@ export class ThreadsViewSession {
this.environment.registerPointerDown((event) => {
this.cancelArchiveConfirmOnOutsidePointer(event);
});
const activeThreadsSnapshot = this.host.threadCatalog.activeSnapshot();
const activeThreadsSnapshot = this.host.threadCatalog.activeThreadsSnapshot();
if (activeThreadsSnapshot) {
this.threads = activeThreadsSnapshot;
this.threadsLoaded = true;
}
this.unsubscribeThreads = this.host.threadCatalog.observeActive((result) => {
this.unsubscribeThreads = this.host.threadCatalog.observeActiveThreadsResult((result) => {
this.receiveObservedThreadsResult(result);
});
this.render();
@ -126,11 +128,11 @@ export class ThreadsViewSession {
}
async refresh(): Promise<void> {
await this.requestThreads(() => this.host.threadCatalog.refreshActive());
await this.requestThreads(() => this.host.threadCatalog.refreshActiveThreads());
}
private async load(): Promise<void> {
await this.requestThreads(() => this.host.threadCatalog.loadActive());
await this.requestThreads(() => this.host.threadCatalog.fetchActiveThreads());
}
private async requestThreads(request: () => Promise<readonly Thread[]>): Promise<void> {
@ -151,9 +153,9 @@ export class ThreadsViewSession {
async loadMore(): Promise<void> {
const lifetime = this.lifetime.signal();
if (!this.lifetime.isCurrent(lifetime) || !this.host.threadCatalog.hasMoreActive() || this.observedFetching) return;
if (!this.lifetime.isCurrent(lifetime) || !this.host.threadCatalog.hasMoreActiveThreads() || this.observedFetching) return;
try {
await this.host.threadCatalog.loadMoreActive();
await this.host.threadCatalog.loadMoreActiveThreads();
} catch (error) {
if (!this.lifetime.isCurrent(lifetime) || isStaleExecutionRuntimeError(error)) return;
this.noticeError(error);
@ -209,7 +211,7 @@ export class ThreadsViewSession {
status: this.status.kind === "idle" ? null : this.status,
loading: this.observedFetchingNextPage,
fetching: this.observedFetching,
hasMore: this.host.threadCatalog.hasMoreActive(),
hasMore: this.host.threadCatalog.hasMoreActiveThreads(),
rows: threadRows(
this.threads,
this.host.openPanelActivities(),

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

@ -5,129 +5,32 @@ import type { Thread } from "../../../domain/threads/model";
type ActiveThreadListObserver = ObservedPaginatedResultListener<readonly Thread[]>;
type ArchivedThreadListObserver = ObservedResultListener<readonly Thread[]>;
interface ThreadCatalogStore {
export interface ThreadCatalogPaginatedActiveReader {
activeThreadsSnapshot(): readonly Thread[] | null;
recentActiveThreadsSnapshot(): readonly Thread[] | null;
archivedThreadsSnapshot(): readonly Thread[] | null;
fetchActiveThreadSearchInventory(): Promise<readonly Thread[]>;
fetchActiveThreads(): Promise<readonly Thread[]>;
refreshActiveThreads(): Promise<readonly Thread[]>;
observeActiveThreadsResult(observer: ActiveThreadListObserver, options?: { emitCurrent?: boolean }): () => void;
hasMoreActiveThreads(): boolean;
loadMoreActiveThreads(): Promise<readonly Thread[]>;
refreshActiveThreads(): Promise<readonly Thread[]>;
refreshArchivedThreads(): Promise<readonly Thread[]>;
applyThreadListMutations(mutations: readonly ThreadListMutation[]): void;
observeActiveThreadsResult(observer: ActiveThreadListObserver, options?: { emitCurrent?: boolean }): () => void;
observeArchivedThreadsResult(observer: ArchivedThreadListObserver, options?: { emitCurrent?: boolean }): () => void;
}
type ThreadCatalogEventObserver = (event: ThreadCatalogEvent) => void;
export interface ThreadCatalogOptions {
store: ThreadCatalogStore;
onEventApplied?: ThreadCatalogEventObserver;
}
export type ThreadCatalogEvent =
| { type: "thread-upserted"; thread: Thread }
| { type: "thread-renamed"; threadId: string; name: string | null }
| { type: "thread-archived"; threadId: string }
| { type: "thread-deleted"; threadId: string }
| { type: "thread-restored"; thread: Thread }
| { type: "thread-unarchived"; threadId: string };
export interface ThreadCatalogPaginatedActiveReader {
activeSnapshot(): readonly Thread[] | null;
recentActiveSnapshot(): readonly Thread[] | null;
loadActive(): Promise<readonly Thread[]>;
refreshActive(): Promise<readonly Thread[]>;
observeActive(observer: ActiveThreadListObserver, options?: { emitCurrent?: boolean }): () => void;
hasMoreActive(): boolean;
loadMoreActive(): Promise<readonly Thread[]>;
}
export interface ThreadCatalogSearchReader {
searchActive(): Promise<readonly Thread[]>;
fetchActiveThreadSearchInventory(): Promise<readonly Thread[]>;
}
export interface ThreadCatalogArchivedReader {
archivedSnapshot(): readonly Thread[] | null;
refreshArchived(): Promise<readonly Thread[]>;
observeArchived(observer: ArchivedThreadListObserver, options?: { emitCurrent?: boolean }): () => void;
archivedThreadsSnapshot(): readonly Thread[] | null;
refreshArchivedThreads(): Promise<readonly Thread[]>;
observeArchivedThreadsResult(observer: ArchivedThreadListObserver, options?: { emitCurrent?: boolean }): () => void;
}
export interface ThreadCatalogEventSink {
apply(event: ThreadCatalogEvent): void;
interface ThreadCatalogWriter {
applyThreadListMutations(changes: readonly ThreadListMutation[]): void;
}
export interface ThreadCatalog
extends ThreadCatalogPaginatedActiveReader,
ThreadCatalogSearchReader,
ThreadCatalogArchivedReader,
ThreadCatalogEventSink {}
export function createThreadCatalog(options: ThreadCatalogOptions): ThreadCatalog {
const { store } = options;
const apply = (event: ThreadCatalogEvent): void => {
store.applyThreadListMutations(threadListMutationsForEvent(store, event));
options.onEventApplied?.(event);
};
return {
apply,
activeSnapshot: () => store.activeThreadsSnapshot(),
recentActiveSnapshot: () => store.recentActiveThreadsSnapshot(),
loadActive: () => store.fetchActiveThreads(),
searchActive: () => store.fetchActiveThreadSearchInventory(),
refreshActive: () => store.refreshActiveThreads(),
hasMoreActive: () => store.hasMoreActiveThreads(),
loadMoreActive: () => store.loadMoreActiveThreads(),
observeActive: (observer, observeOptions) => store.observeActiveThreadsResult(observer, observeOptions),
archivedSnapshot: () => store.archivedThreadsSnapshot(),
refreshArchived: () => store.refreshArchivedThreads(),
observeArchived: (observer, observeOptions) => store.observeArchivedThreadsResult(observer, observeOptions),
};
}
function threadListMutationsForEvent(store: ThreadCatalogStore, event: ThreadCatalogEvent): ThreadListMutation[] {
switch (event.type) {
case "thread-upserted":
return [{ kind: "upsert", list: "active", thread: { ...event.thread, archived: false } }];
case "thread-renamed":
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":
return [
{ kind: "remove", list: "active", threadId: event.threadId },
{ kind: "remove", list: "archived", threadId: event.threadId },
];
case "thread-restored":
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 threadById(threads: readonly Thread[] | null, threadId: string): Thread | null {
return threads?.find((thread) => thread.id === threadId) ?? null;
}
ThreadCatalogWriter {}

184
src/features/threads/workflows/thread-operation-coordinator.ts Normal file
View file

@ -0,0 +1,184 @@
import type { Thread } from "../../../domain/threads/model";
import type {
ThreadLifecycleEvent,
ThreadOperationCommitter,
ThreadOperationEvent,
ThreadOperationEventSink,
} from "./thread-operation-event";
interface ThreadForkPublication {
record(thread: Thread): void;
finish(options?: { sourceArchived?: boolean }): void;
}
export interface ThreadOperationCoordinator extends ThreadOperationEventSink {
beginForkPublication(sourceThreadId: string): ThreadForkPublication;
}
type ForkedThreadState = "active" | "archived" | "deleted";
interface PendingForkPublication {
child: PendingForkChild | null;
finished: boolean;
}
interface PendingForkChild {
thread: Thread;
state: ForkedThreadState;
eventsBeforeClaim: ThreadOperationEvent[] | null;
}
interface ForkSourceGroup {
publications: Set<PendingForkPublication>;
unclaimedChildren: Map<string, PendingForkChild>;
sourceState: "active" | "archived" | null;
restoredSource: Thread | null;
}
export function createThreadOperationCoordinator(committer: ThreadOperationCommitter): ThreadOperationCoordinator {
const sourceGroups = new Map<string, ForkSourceGroup>();
const pendingChildrenByThread = new Map<string, PendingForkChild>();
const apply = (event: ThreadOperationEvent): void => {
const pendingChild = pendingChildrenByThread.get(threadIdForEvent(event));
if (pendingChild) {
pendingChild.eventsBeforeClaim?.push(event);
if (applyChildEvent(pendingChild, event)) return;
}
if (event.type === "thread-upserted" && event.forkedFromThreadId) {
const group = sourceGroups.get(event.forkedFromThreadId);
if (group) {
const child = { thread: event.thread, state: "active", eventsBeforeClaim: [] } satisfies PendingForkChild;
group.unclaimedChildren.set(event.thread.id, child);
pendingChildrenByThread.set(event.thread.id, child);
return;
}
}
const sourceGroup = sourceGroups.get(threadIdForEvent(event));
if (sourceGroup && applySourceEvent(sourceGroup, event)) return;
committer([lifecycleEvent(event)]);
};
const beginForkPublication = (sourceThreadId: string): ThreadForkPublication => {
const publication: PendingForkPublication = { child: null, finished: false };
const group = sourceGroups.get(sourceThreadId) ?? createForkSourceGroup();
group.publications.add(publication);
sourceGroups.set(sourceThreadId, group);
return {
record: (thread) => {
if (publication.finished) return;
const child = group.unclaimedChildren.get(thread.id) ?? { thread, state: "active", eventsBeforeClaim: null };
const eventsBeforeClaim = child.eventsBeforeClaim;
child.thread = thread;
child.state = "active";
child.eventsBeforeClaim = null;
for (const event of eventsBeforeClaim ?? []) applyChildEvent(child, event);
publication.child = child;
group.unclaimedChildren.delete(thread.id);
pendingChildrenByThread.set(thread.id, child);
},
finish: (options = {}) => {
if (publication.finished) return;
publication.finished = true;
if (publication.child) pendingChildrenByThread.delete(publication.child.thread.id);
group.publications.delete(publication);
if (group.sourceState === null && options.sourceArchived !== undefined) {
group.sourceState = options.sourceArchived ? "archived" : "active";
}
const events = completedForkChildEvents(publication);
if (group.publications.size === 0) events.push(...completedSourceEvents(group, sourceThreadId));
if (events.length > 0) committer(events);
if (group.publications.size > 0) return;
sourceGroups.delete(sourceThreadId);
for (const child of group.unclaimedChildren.values()) {
pendingChildrenByThread.delete(child.thread.id);
committer(completedChildEvents(child));
}
},
};
};
return { apply, beginForkPublication };
}
function createForkSourceGroup(): ForkSourceGroup {
return {
publications: new Set(),
unclaimedChildren: new Map(),
sourceState: null,
restoredSource: null,
};
}
function applyChildEvent(child: PendingForkChild, event: ThreadOperationEvent): boolean {
switch (event.type) {
case "thread-upserted":
child.thread = event.thread;
return true;
case "thread-renamed":
child.thread = { ...child.thread, name: event.name };
return true;
case "thread-archived":
child.state = "archived";
return true;
case "thread-deleted":
child.state = "deleted";
return true;
case "thread-restored":
child.thread = event.thread;
child.state = "active";
return true;
case "thread-unarchived":
child.state = "active";
return true;
}
}
function applySourceEvent(group: ForkSourceGroup, event: ThreadOperationEvent): boolean {
switch (event.type) {
case "thread-archived":
group.sourceState = "archived";
return true;
case "thread-unarchived":
if (group.sourceState !== "archived") return false;
group.sourceState = "active";
return true;
case "thread-restored":
if (group.sourceState) group.sourceState = "active";
group.restoredSource = event.thread;
return true;
default:
return false;
}
}
function completedForkChildEvents(publication: PendingForkPublication): ThreadLifecycleEvent[] {
return publication.child ? completedChildEvents(publication.child) : [];
}
function completedChildEvents(child: PendingForkChild): ThreadLifecycleEvent[] {
const events: ThreadLifecycleEvent[] = [{ type: "thread-upserted", thread: child.thread }];
if (child.state === "archived") events.push({ type: "thread-archived", threadId: child.thread.id });
if (child.state === "deleted") events.push({ type: "thread-deleted", threadId: child.thread.id });
return events;
}
function completedSourceEvents(group: ForkSourceGroup, sourceThreadId: string): ThreadLifecycleEvent[] {
if (group.sourceState === "archived") return [{ type: "thread-archived", threadId: sourceThreadId }];
if (group.restoredSource) return [{ type: "thread-restored", thread: group.restoredSource }];
return [];
}
function lifecycleEvent(event: ThreadOperationEvent): ThreadLifecycleEvent {
if (event.type === "thread-upserted") return { type: "thread-upserted", thread: event.thread };
return event;
}
function threadIdForEvent(event: ThreadOperationEvent): string {
return event.type === "thread-upserted" || event.type === "thread-restored" ? event.thread.id : event.threadId;
}

19
src/features/threads/workflows/thread-operation-event.ts Normal file
View file

@ -0,0 +1,19 @@
import type { Thread } from "../../../domain/threads/model";
export type ThreadLifecycleEvent =
| { type: "thread-upserted"; thread: Thread }
| { type: "thread-renamed"; threadId: string; name: string | null }
| { type: "thread-archived"; threadId: string }
| { type: "thread-deleted"; threadId: string }
| { type: "thread-restored"; thread: Thread }
| { type: "thread-unarchived"; threadId: string };
export type ThreadOperationEvent =
| Exclude<ThreadLifecycleEvent, { type: "thread-upserted" }>
| { type: "thread-upserted"; thread: Thread; forkedFromThreadId?: string | null };
export interface ThreadOperationEventSink {
apply(event: ThreadOperationEvent): void;
}
export type ThreadOperationCommitter = (events: readonly ThreadLifecycleEvent[]) => void;

8
src/features/threads/workflows/thread-operations.ts
View file

@ -1,9 +1,9 @@
import { normalizeExplicitThreadName, type Thread } from "../../../domain/threads/model";
import { threadDisplayTitle } from "../../../domain/threads/title";
import type { KeyedOperationQueue } from "../../../shared/runtime/keyed-operation-queue";
import type { ThreadCatalogEventSink } from "../catalog/thread-catalog";
import { type ArchiveExportDestination, type ArchiveExportSettings, exportArchivedThreadMarkdown } from "./archive-export";
import type { ThreadOperationsTransport } from "./ports";
import type { ThreadOperationEventSink } from "./thread-operation-event";
export interface ThreadOperationsHost {
transport: ThreadOperationsTransport;
@ -15,7 +15,7 @@ export interface ThreadOperationsHost {
vaultConfigDir: string;
};
archiveDestination(): ArchiveExportDestination;
catalog: ThreadCatalogEventSink;
operationEvents: ThreadOperationEventSink;
referenceThreads(): readonly Thread[];
notice(message: string): void;
}
@ -58,7 +58,7 @@ async function renameThread(
if (!(options.shouldStart?.() ?? true)) return false;
await host.transport.renameThread(threadId, name);
if (options.shouldPublish?.() ?? true) {
host.catalog.apply({ type: "thread-renamed", threadId, name });
host.operationEvents.apply({ type: "thread-renamed", threadId, name });
}
return true;
});
@ -101,6 +101,6 @@ async function archiveThread(
if (exportedPath) {
host.notice(`Saved archived thread to ${exportedPath}.`);
}
host.catalog.apply({ type: "thread-archived", threadId });
host.operationEvents.apply({ type: "thread-archived", threadId });
return { exportedPath };
}

73
src/features/threads/workflows/thread-read-model-projection.ts Normal file
View file

@ -0,0 +1,73 @@
import { applyThreadListMutation, type ThreadListMutation } from "../../../app-server/query/thread-list-mutation";
import type { Thread } from "../../../domain/threads/model";
import type { ThreadLifecycleEvent } from "./thread-operation-event";
interface ThreadReadModelSnapshots {
activeThreadsSnapshot(): readonly Thread[] | null;
archivedThreadsSnapshot(): readonly Thread[] | null;
}
export function projectThreadListChanges(
snapshots: ThreadReadModelSnapshots,
events: readonly ThreadLifecycleEvent[],
): ThreadListMutation[] {
let active = snapshots.activeThreadsSnapshot();
let archived = snapshots.archivedThreadsSnapshot();
const changes: ThreadListMutation[] = [];
for (const event of events) {
const eventChanges = threadListChangesForEvent({ active, archived }, event);
changes.push(...eventChanges);
for (const change of eventChanges) {
if (change.list === "active") active = applyThreadListMutation(active, change);
else archived = applyThreadListMutation(archived, change);
}
}
return changes;
}
function threadListChangesForEvent(
snapshots: { active: readonly Thread[] | null; archived: readonly Thread[] | null },
event: ThreadLifecycleEvent,
): ThreadListMutation[] {
switch (event.type) {
case "thread-upserted":
return [{ kind: "upsert", list: "active", thread: { ...event.thread, archived: false } }];
case "thread-renamed":
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(snapshots.active, 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":
return [
{ kind: "remove", list: "active", threadId: event.threadId },
{ kind: "remove", list: "archived", threadId: event.threadId },
];
case "thread-restored":
return [
{ kind: "upsert", list: "active", thread: { ...event.thread, archived: false } },
{ kind: "remove", list: "archived", threadId: event.thread.id },
];
case "thread-unarchived": {
const thread = threadById(snapshots.archived, 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 threadById(threads: readonly Thread[] | null, threadId: string): Thread | null {
return threads?.find((thread) => thread.id === threadId) ?? null;
}

12
src/plugin-runtime.ts
View file

@ -9,7 +9,7 @@ import type {
} from "./features/chat/host/contracts";
import type { SelectionRewriteCommandController } from "./features/selection-rewrite/command.obsidian";
import type { SelectionRewriteTransport } from "./features/selection-rewrite/transport";
import type { ThreadCatalogEvent } from "./features/threads/catalog/thread-catalog";
import type { ThreadLifecycleEvent } from "./features/threads/workflows/thread-operation-event";
import type { ThreadsViewPanelActivity } from "./features/threads-view/state";
import { CodexThreadsView, type ThreadsRuntimeView, type ThreadsViewRuntimeOwner } from "./features/threads-view/view.obsidian";
import { persistedTurnDiffViewState, type TurnDiffViewState } from "./features/turn-diff/model";
@ -187,7 +187,11 @@ export class CodexPanelRuntime implements ChatViewRuntimeOwner, ThreadsViewRunti
}
}
private applyThreadCatalogSurfaceEvent(event: ThreadCatalogEvent): void {
private applyThreadLifecycleSurfaceEvents(events: readonly ThreadLifecycleEvent[]): void {
for (const event of events) this.applyThreadLifecycleSurfaceEvent(event);
}
private applyThreadLifecycleSurfaceEvent(event: ThreadLifecycleEvent): void {
switch (event.type) {
case "thread-archived":
this.applyThreadArchived(event.threadId);
@ -251,8 +255,8 @@ export class CodexPanelRuntime implements ChatViewRuntimeOwner, ThreadsViewRunti
this.refreshThreadsViewLiveState();
},
},
onThreadCatalogEvent: (event) => {
this.applyThreadCatalogSurfaceEvent(event);
onThreadLifecycleEvents: (events) => {
this.applyThreadLifecycleSurfaceEvents(events);
},
openNewPanel: () => this.panels.openNewPanel(),
openThreadInCurrentView: (threadId) => this.panels.openThreadInCurrentView(threadId),

19
src/settings/app-server-dynamic-data.ts
View file

@ -4,7 +4,8 @@ import type { ObservedResultListener } from "../app-server/query/observed-result
import { listHookCatalog, setHookItemEnabled, trustHookItem } from "../app-server/services/catalog";
import { deleteThread, restoreArchivedThread as restoreArchivedThreadOnAppServer } from "../app-server/services/threads";
import type { HookItem, ModelMetadata } from "../domain/catalog/metadata";
import type { ThreadCatalogArchivedReader, ThreadCatalogEventSink } from "../features/threads/catalog/thread-catalog";
import type { ThreadCatalogArchivedReader } from "../features/threads/catalog/thread-catalog";
import type { ThreadOperationEventSink } from "../features/threads/workflows/thread-operation-event";
import { createKeyedOperationQueue } from "../shared/runtime/keyed-operation-queue";
import type { SettingsDynamicDataAccess, SettingsHookCatalog } from "./dynamic-data";
@ -15,13 +16,12 @@ interface SettingsAppServerQueries {
refreshModels(): Promise<readonly ModelMetadata[]>;
}
type SettingsArchivedThreadCatalog = ThreadCatalogArchivedReader & ThreadCatalogEventSink;
export interface SettingsAppServerDynamicDataOptions {
vaultPath: string;
clientAccess: AppServerClientAccess;
appServerQueries: SettingsAppServerQueries;
threadCatalog: SettingsArchivedThreadCatalog;
threadCatalog: ThreadCatalogArchivedReader;
threadEvents: ThreadOperationEventSink;
}
export function createSettingsAppServerDynamicData(options: SettingsAppServerDynamicDataOptions): SettingsDynamicDataAccess {
@ -45,22 +45,23 @@ export function createSettingsAppServerDynamicData(options: SettingsAppServerDyn
observeModelsResult: (listener, observeOptions) => options.appServerQueries.observeModelsResult(listener, observeOptions),
fetchModels: () => options.appServerQueries.fetchModels(),
refreshModels: () => options.appServerQueries.refreshModels(),
archivedThreadsSnapshot: () => options.threadCatalog.archivedSnapshot(),
observeArchivedThreadsResult: (listener, observeOptions) => options.threadCatalog.observeArchived(listener, observeOptions),
refreshArchivedThreads: () => options.threadCatalog.refreshArchived(),
archivedThreadsSnapshot: () => options.threadCatalog.archivedThreadsSnapshot(),
observeArchivedThreadsResult: (listener, observeOptions) =>
options.threadCatalog.observeArchivedThreadsResult(listener, observeOptions),
refreshArchivedThreads: () => options.threadCatalog.refreshArchivedThreads(),
refreshHooks: () => withSettingsConnection(loadHooks),
trustHook: (hook) => mutateHook(hook, trustHookItem),
setHookEnabled: (hook, enabled) => mutateHook(hook, (client, item) => setHookItemEnabled(client, item, enabled)),
restoreArchivedThread: (threadId) =>
runArchivedThreadMutation(threadId, async () => {
const thread = await withSettingsConnection((client) => restoreArchivedThreadOnAppServer(client, threadId));
options.threadCatalog.apply({ type: "thread-restored", thread });
options.threadEvents.apply({ type: "thread-restored", thread });
return thread;
}),
deleteArchivedThread: (threadId) =>
runArchivedThreadMutation(threadId, async () => {
await withSettingsConnection((client) => deleteThread(client, threadId));
options.threadCatalog.apply({ type: "thread-deleted", threadId });
options.threadEvents.apply({ type: "thread-deleted", threadId });
}),
};
}

2
tests/execution-runtime.test.ts
View file

@ -145,7 +145,7 @@ function executionRuntime(): CodexExecutionRuntime {
context: { codexPath: "codex", vaultPath: "/vault" },
settings: () => ({ ...DEFAULT_SETTINGS }),
workspace: {} as never,
onThreadCatalogEvent: vi.fn(),
onThreadLifecycleEvents: vi.fn(),
openNewPanel: vi.fn(),
openThreadInCurrentView: vi.fn(),
openThreadInAvailableView: vi.fn(),

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

@ -17,7 +17,7 @@ import {
} from "../../../../../src/features/chat/application/state/root-reducer";
import type { ChatStateStore } from "../../../../../src/features/chat/application/state/store";
import { pendingTurnStart } from "../../../../../src/features/chat/application/turns/turn-state";
import type { ThreadCatalogEvent } from "../../../../../src/features/threads/catalog/thread-catalog";
import type { ThreadOperationEvent as ThreadCatalogEvent } from "../../../../../src/features/threads/workflows/thread-operation-event";
import { chatStateFixture, chatStateWith } from "../../support/state";
import { chatStateThreadStreamItems, withChatStateThreadStreamItems } from "../../support/thread-stream";
@ -28,18 +28,24 @@ type TestChatInboundHandler = Omit<ChatInboundHandler, "handleNotification"> & {
currentState(): ChatState;
};
function handlerForState(state = chatStateFixture(), actions: Partial<ChatInboundHandlerActions> = {}): TestChatInboundHandler {
function handlerForState(
state = chatStateFixture(),
actions: Partial<ChatInboundHandlerActions> & {
applyThreadCatalogEvent?: ChatInboundHandlerActions["applyThreadOperationEvent"];
} = {},
): TestChatInboundHandler {
const store = testStoreForState(state);
const { applyThreadCatalogEvent, ...inboundActions } = actions;
const handler = createChatInboundHandler(
store,
{
refreshServerDiagnostics: vi.fn(),
applyAppServerResourceEvent: vi.fn(),
maybeNameThread: vi.fn(),
applyThreadCatalogEvent: vi.fn(),
applyThreadOperationEvent: applyThreadCatalogEvent ?? vi.fn(),
respondToServerRequest: vi.fn(() => true),
rejectServerRequest: vi.fn(() => true),
...actions,
...inboundActions,
},
createLocalIdSource({ nowMs: () => 1, seed: "test" }),
);
@ -1324,6 +1330,23 @@ describe("ChatInboundHandler", () => {
expect(applyThreadCatalogEvent).toHaveBeenCalledWith({
type: "thread-upserted",
thread: expect.objectContaining({ id: "thread-other" }),
forkedFromThreadId: null,
});
});
it("preserves fork ancestry for atomic catalog publication", () => {
const applyThreadCatalogEvent = vi.fn();
const handler = handlerForState(chatStateFixture(), { applyThreadCatalogEvent });
handler.handleNotification({
method: "thread/started",
params: { thread: { ...appServerThread("thread-forked", "/workspace"), forkedFromId: "thread-source" } },
} satisfies Extract<ServerNotification, { method: "thread/started" }>);
expect(applyThreadCatalogEvent).toHaveBeenCalledWith({
type: "thread-upserted",
thread: expect.objectContaining({ id: "thread-forked" }),
forkedFromThreadId: "thread-source",
});
});
@ -1342,6 +1365,7 @@ describe("ChatInboundHandler", () => {
expect(applyThreadCatalogEvent).toHaveBeenCalledWith({
type: "thread-upserted",
thread: expect.objectContaining({ id: "thread-active" }),
forkedFromThreadId: null,
});
});

2
tests/features/chat/application/threads/auto-title-coordinator.test.ts
View file

@ -206,7 +206,7 @@ function coordinatorFixture(
createFolder: vi.fn().mockResolvedValue(undefined),
createMarkdownFile: vi.fn().mockResolvedValue(undefined),
}),
catalog: {
operationEvents: {
apply: (event) => {
if (event.type === "thread-renamed") {
stateStore.dispatch({

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

@ -32,7 +32,7 @@ type ThreadManagementActionsHostMock = Omit<
| "openThreadInCurrentPanel"
| "openThreadInNewView"
| "operations"
| "recordThread"
| "beginThreadForkPublication"
| "setComposerText"
| "setStatus"
| "threadTransport"
@ -44,7 +44,11 @@ type ThreadManagementActionsHostMock = Omit<
setComposerText: Mock<ThreadManagementActionsHost["setComposerText"]>;
openThreadInNewView: Mock<ThreadManagementActionsHost["openThreadInNewView"]>;
openThreadInCurrentPanel: Mock<ThreadManagementActionsHost["openThreadInCurrentPanel"]>;
recordThread: Mock<ThreadManagementActionsHost["recordThread"]>;
forkPublication: {
record: Mock<ReturnType<ThreadManagementActionsHost["beginThreadForkPublication"]>["record"]>;
finish: Mock<ReturnType<ThreadManagementActionsHost["beginThreadForkPublication"]>["finish"]>;
};
beginThreadForkPublication: Mock<ThreadManagementActionsHost["beginThreadForkPublication"]>;
};
describe("thread management actions", () => {
@ -301,12 +305,14 @@ describe("thread management actions", () => {
expect(host.threadTransport.forkThread).toHaveBeenCalledWith("source", {
position: { kind: "through-turn", turnId: "turn-1" },
});
expect(host.recordThread).toHaveBeenCalledWith(
expect(host.forkPublication.record).toHaveBeenCalledWith(
expect.objectContaining({
id: "forked",
}),
);
expect(host.openThreadInNewView).toHaveBeenCalledWith("forked");
expect(host.forkPublication.finish).toHaveBeenCalledWith();
expect(callOrder(host.openThreadInNewView)).toBeLessThan(callOrder(host.forkPublication.finish));
expect(host.operations.archiveThread).not.toHaveBeenCalled();
expect(host.openThreadInCurrentPanel).not.toHaveBeenCalled();
});
@ -323,6 +329,8 @@ describe("thread management actions", () => {
expect(host.operations.archiveThread).toHaveBeenCalledWith("source", {});
expect(host.openThreadInCurrentPanel).toHaveBeenCalledWith("forked", expect.any(Function));
expect(callOrder(host.openThreadInCurrentPanel)).toBeLessThan(callOrder(host.operations.archiveThread));
expect(host.forkPublication.finish).toHaveBeenCalledWith({ sourceArchived: true });
expect(callOrder(host.operations.archiveThread)).toBeLessThan(callOrder(host.forkPublication.finish));
});
it("keeps the replacement panel when source archiving fails", async () => {
@ -339,6 +347,7 @@ describe("thread management actions", () => {
expect(host.operations.archiveThread).toHaveBeenCalledWith("source", {});
expect(host.openThreadInCurrentPanel).toHaveBeenCalledWith("forked", expect.any(Function));
expect(activeThreadId(host.stateStore.getState())).toBe("forked");
expect(host.forkPublication.finish).toHaveBeenCalledWith({ sourceArchived: false });
expect(host.addSystemMessage).toHaveBeenCalledWith("Forked the thread, but could not archive the previous version: archive failed");
});
@ -499,7 +508,8 @@ describe("thread management actions", () => {
await controller.forkThreadFromTurn("source", null, false);
expect(host.recordThread).not.toHaveBeenCalled();
expect(host.forkPublication.record).not.toHaveBeenCalled();
expect(host.forkPublication.finish).toHaveBeenCalledWith();
expect(host.openThreadInNewView).not.toHaveBeenCalled();
expect(host.operations.archiveThread).not.toHaveBeenCalled();
});
@ -517,7 +527,8 @@ describe("thread management actions", () => {
await controller.forkThreadFromTurn("source", null, true);
expect(host.recordThread).toHaveBeenCalledWith(panelThread("forked"));
expect(host.forkPublication.record).toHaveBeenCalledWith(panelThread("forked"));
expect(host.forkPublication.finish).toHaveBeenCalledWith();
expect(host.operations.renameThread).not.toHaveBeenCalled();
expect(host.operations.archiveThread).not.toHaveBeenCalled();
expect(host.openThreadInCurrentPanel).not.toHaveBeenCalled();
@ -582,7 +593,8 @@ describe("thread management actions", () => {
expect(host.operations.archiveThread).toHaveBeenCalledWith("source", { saveMarkdown: false });
expect(callOrder(host.openThreadInCurrentPanel)).toBeLessThan(callOrder(host.operations.archiveThread));
expect(callOrder(host.setComposerText)).toBeLessThan(callOrder(host.addSystemMessage));
expect(host.recordThread).toHaveBeenCalledWith(panelThread("forked"));
expect(host.forkPublication.record).toHaveBeenCalledWith(panelThread("forked"));
expect(host.forkPublication.finish).toHaveBeenCalledWith({ sourceArchived: true });
expect(activeThreadId(host.stateStore.getState())).toBe("forked");
});
@ -645,7 +657,7 @@ describe("thread management actions", () => {
await threadManagementActions(host).rollbackThread("source");
expect(host.recordThread).toHaveBeenCalledWith(panelThread("forked"));
expect(host.forkPublication.record).toHaveBeenCalledWith(panelThread("forked"));
expect(activeThreadId(host.stateStore.getState())).toBe("source");
expect(host.setComposerText).not.toHaveBeenCalled();
expect(host.operations.archiveThread).not.toHaveBeenCalled();
@ -756,7 +768,7 @@ describe("thread management actions", () => {
expect(activeThreadId(host.stateStore.getState())).toBe("other");
expect(host.setComposerText).not.toHaveBeenCalled();
expect(host.recordThread).toHaveBeenCalledWith(panelThread("forked"));
expect(host.forkPublication.record).toHaveBeenCalledWith(panelThread("forked"));
expect(host.operations.archiveThread).not.toHaveBeenCalled();
});
@ -789,7 +801,7 @@ describe("thread management actions", () => {
expect(host.stateStore.getState().turn.lifecycle).toEqual({ kind: "running", turnId: "new-turn" });
expect(host.setComposerText).not.toHaveBeenCalled();
expect(host.recordThread).toHaveBeenCalledWith(panelThread("forked"));
expect(host.forkPublication.record).toHaveBeenCalledWith(panelThread("forked"));
expect(host.operations.archiveThread).not.toHaveBeenCalled();
});
@ -807,7 +819,7 @@ describe("thread management actions", () => {
await controller.rollbackThread("source");
expect(host.recordThread).not.toHaveBeenCalled();
expect(host.forkPublication.record).not.toHaveBeenCalled();
expect(host.setComposerText).not.toHaveBeenCalled();
expect(host.operations.archiveThread).not.toHaveBeenCalled();
});
@ -844,7 +856,7 @@ describe("thread management actions", () => {
await controller.rollbackThread("source");
expect(host.setComposerText).not.toHaveBeenCalled();
expect(host.recordThread).not.toHaveBeenCalled();
expect(host.forkPublication.record).not.toHaveBeenCalled();
expect(host.operations.archiveThread).not.toHaveBeenCalled();
});
});
@ -913,6 +925,10 @@ function hostMock({
renameThread: vi.fn<ThreadManagementActionsHost["operations"]["renameThread"]>().mockResolvedValue(true),
...operationOverrides,
};
const forkPublication = {
record: vi.fn(),
finish: vi.fn(),
};
return {
stateStore,
threadTransport,
@ -928,7 +944,8 @@ function hostMock({
onAdopted();
return { adopted: true };
}),
recordThread: vi.fn<ThreadManagementActionsHost["recordThread"]>(),
forkPublication,
beginThreadForkPublication: vi.fn<ThreadManagementActionsHost["beginThreadForkPublication"]>().mockReturnValue(forkPublication),
threadHasPendingOrRunningPanel: vi.fn(() => false),
};
}

2
tests/features/chat/host/connection-bundle.test.ts
View file

@ -79,7 +79,7 @@ function connectionBundleFixture(overrides: { readServerDiagnostics?: ReturnType
refreshRateLimits: vi.fn().mockResolvedValue(undefined),
},
threadCatalog: {
refreshActive: vi.fn().mockResolvedValue(undefined),
refreshActiveThreads: vi.fn().mockResolvedValue(undefined),
apply: vi.fn(),
},
appServerContext: { codexPath: "codex", vaultPath: "/vault" },

25
tests/features/chat/host/session-runtime.test.ts
View file

@ -58,7 +58,7 @@ describe("ChatPanelSessionRuntime actions", () => {
environment: {
plugin: {
threadCatalog: {
refreshActive: refresh,
refreshActiveThreads: refresh,
},
},
},
@ -133,7 +133,7 @@ describe("ChatPanelSessionRuntime actions", () => {
const { runtime, stateStore, deferredTasks, threadStreamScrollBinding } = sessionRuntimeFixture({
environment: {
plugin: {
threadCatalog: { observeActive: vi.fn(() => unsubscribeThreads) },
threadCatalog: { observeActiveThreadsResult: vi.fn(() => unsubscribeThreads) },
appServerQueries: {
observeAppServerMetadataResources: vi.fn(() => unsubscribeMetadata),
},
@ -196,6 +196,7 @@ describe("ChatPanelSessionRuntime actions", () => {
plugin?: {
workspace?: Partial<ChatPanelEnvironment["plugin"]["workspace"]>;
threadCatalog?: Partial<ChatPanelEnvironment["plugin"]["threadCatalog"]>;
threadOperationCoordinator?: Partial<ChatPanelEnvironment["plugin"]["threadOperationCoordinator"]>;
appServerQueries?: Partial<ChatPanelEnvironment["plugin"]["appServerQueries"]>;
settings?: ChatPanelEnvironment["plugin"]["settings"];
appServerContext?: ChatPanelEnvironment["plugin"]["appServerContext"];
@ -268,6 +269,11 @@ describe("ChatPanelSessionRuntime actions", () => {
},
appServerQueries,
threadCatalog,
threadOperationCoordinator: {
apply: overrides.plugin?.threadOperationCoordinator?.apply ?? vi.fn(),
beginForkPublication:
overrides.plugin?.threadOperationCoordinator?.beginForkPublication ?? vi.fn(() => ({ record: vi.fn(), finish: vi.fn() })),
},
threadNameMutations: createKeyedOperationQueue(),
threadGoalOperations: createThreadGoalOperationCoordinator(),
runtimeSettingsCommitQueue: createKeyedOperationQueue(),
@ -285,14 +291,13 @@ describe("ChatPanelSessionRuntime actions", () => {
overrides: Partial<ChatPanelEnvironment["plugin"]["threadCatalog"]> = {},
): ChatPanelEnvironment["plugin"]["threadCatalog"] {
return {
hasMoreActive: vi.fn(() => false),
loadMoreActive: vi.fn().mockResolvedValue([]),
loadActive: vi.fn().mockResolvedValue([]),
refreshActive: vi.fn().mockResolvedValue([]),
activeSnapshot: vi.fn(() => null),
recentActiveSnapshot: vi.fn(() => null),
observeActive: vi.fn(() => () => undefined),
apply: vi.fn(),
hasMoreActiveThreads: vi.fn(() => false),
loadMoreActiveThreads: vi.fn().mockResolvedValue([]),
fetchActiveThreads: vi.fn().mockResolvedValue([]),
refreshActiveThreads: vi.fn().mockResolvedValue([]),
activeThreadsSnapshot: vi.fn(() => null),
recentActiveThreadsSnapshot: vi.fn(() => null),
observeActiveThreadsResult: vi.fn(() => () => undefined),
...overrides,
};
}

49
tests/features/chat/host/view-connection-harness.ts
View file

@ -11,7 +11,7 @@ import type { SharedServerMetadata, SharedServerMetadataResource } from "../../.
import type { Thread } from "../../../../src/domain/threads/model";
import { createThreadGoalOperationCoordinator } from "../../../../src/features/chat/application/threads/goal-actions";
import type { ChatRuntimeView, ChatViewRuntimeOwner, CodexChatHost } from "../../../../src/features/chat/host/contracts";
import type { ThreadCatalogEvent } from "../../../../src/features/threads/catalog/thread-catalog";
import type { ThreadOperationEvent } from "../../../../src/features/threads/workflows/thread-operation-event";
import { type CodexPanelSettings, DEFAULT_SETTINGS } from "../../../../src/settings/model";
import { createKeyedOperationQueue } from "../../../../src/shared/runtime/keyed-operation-queue";
import { notices } from "../../../mocks/obsidian";
@ -412,9 +412,9 @@ export interface ChatHostFixtureOverrides {
openTurnDiff?: CodexChatHost["workspace"]["openTurnDiff"];
notifyPanelActivityChanged?: CodexChatHost["workspace"]["notifyPanelActivityChanged"];
openSideChat?: CodexChatHost["workspace"]["openSideChat"];
applyThreadCatalogEvent?: CodexChatHost["threadCatalog"]["apply"];
refreshActive?: CodexChatHost["threadCatalog"]["refreshActive"];
activeSnapshot?: CodexChatHost["threadCatalog"]["activeSnapshot"];
applyThreadOperationEvent?: CodexChatHost["threadOperationCoordinator"]["apply"];
refreshActiveThreads?: CodexChatHost["threadCatalog"]["refreshActiveThreads"];
activeThreadsSnapshot?: CodexChatHost["threadCatalog"]["activeThreadsSnapshot"];
appServerMetadataSnapshot?: CodexChatHost["appServerQueries"]["appServerMetadataSnapshot"];
modelsSnapshot?: CodexChatHost["appServerQueries"]["modelsSnapshot"];
fetchModels?: CodexChatHost["appServerQueries"]["fetchModels"];
@ -426,7 +426,7 @@ export interface ChatHostFixtureOverrides {
}
export function chatHost(overrides: ChatHostFixtureOverrides = {}): TestCodexChatHost {
let activeThreads = overrides.activeSnapshot?.() ?? null;
let activeThreads = overrides.activeThreadsSnapshot?.() ?? null;
let metadata = overrides.appServerMetadataSnapshot?.() ?? null;
let models = overrides.modelsSnapshot?.() ?? null;
const activeThreadResultListeners = new Set<(result: ObservedPaginatedResult<readonly Thread[]>) => void>();
@ -523,7 +523,7 @@ export function chatHost(overrides: ChatHostFixtureOverrides = {}): TestCodexCha
activeThreads = [thread, ...(activeThreads?.filter((item) => item.id !== thread.id) ?? [])];
emitActiveThreads();
};
const applyThreadCatalogEvent = (event: ThreadCatalogEvent): void => {
const applyThreadOperationEvent = (event: ThreadOperationEvent): void => {
switch (event.type) {
case "thread-archived":
case "thread-deleted":
@ -620,11 +620,10 @@ export function chatHost(overrides: ChatHostFixtureOverrides = {}): TestCodexCha
},
},
threadCatalog: {
apply: overrides.applyThreadCatalogEvent ?? applyThreadCatalogEvent,
hasMoreActive: vi.fn(() => false),
loadMoreActive: vi.fn(async () => activeThreads ?? []),
refreshActive:
overrides.refreshActive ??
hasMoreActiveThreads: vi.fn(() => false),
loadMoreActiveThreads: vi.fn(async () => activeThreads ?? []),
refreshActiveThreads:
overrides.refreshActiveThreads ??
(vi.fn(async () => {
const client = connectionMock.state.client;
if (!client) return activeThreads ?? [];
@ -638,8 +637,8 @@ export function chatHost(overrides: ChatHostFixtureOverrides = {}): TestCodexCha
activeThreads = response.data.map(threadFromRecord);
for (const listener of activeThreadResultListeners) listener(paginatedQueryResult(activeThreads));
return activeThreads;
}) as CodexChatHost["threadCatalog"]["refreshActive"]),
loadActive: vi.fn(async () => {
}) as CodexChatHost["threadCatalog"]["refreshActiveThreads"]),
fetchActiveThreads: vi.fn(async () => {
const client = connectionMock.state.client;
if (!client) return activeThreads ?? [];
const request = client["request"] as (method: string, params: Record<string, unknown>) => Promise<{ data: ThreadRecord[] }>;
@ -653,13 +652,27 @@ export function chatHost(overrides: ChatHostFixtureOverrides = {}): TestCodexCha
for (const listener of activeThreadResultListeners) listener(paginatedQueryResult(activeThreads));
return activeThreads;
}),
activeSnapshot: overrides.activeSnapshot ?? vi.fn(() => activeThreads),
recentActiveSnapshot: overrides.activeSnapshot ?? vi.fn(() => activeThreads),
observeActive: (listener, options = {}) => {
activeThreadsSnapshot: overrides.activeThreadsSnapshot ?? vi.fn(() => activeThreads),
recentActiveThreadsSnapshot: vi.fn(() => activeThreads),
observeActiveThreadsResult: (listener, options = {}) => {
activeThreadResultListeners.add(listener);
if ((options.emitCurrent ?? true) && activeThreads) listener(paginatedQueryResult(activeThreads));
return () => {
activeThreadResultListeners.delete(listener);
return () => activeThreadResultListeners.delete(listener);
},
},
threadOperationCoordinator: {
apply: overrides.applyThreadOperationEvent ?? applyThreadOperationEvent,
beginForkPublication: (sourceThreadId) => {
let replacement: Thread | null = null;
return {
record: (thread) => {
replacement = thread;
},
finish: (options = {}) => {
if (!replacement) return;
applyThreadOperationEvent({ type: "thread-upserted", thread: replacement, forkedFromThreadId: sourceThreadId });
if (options.sourceArchived ?? false) applyThreadOperationEvent({ type: "thread-archived", threadId: sourceThreadId });
},
};
},
},

2
tests/features/chat/host/view-restoration.test.ts
View file

@ -180,7 +180,7 @@ describe("CodexChatView workspace restoration", () => {
const cachedThread = threadFixture("thread-cached");
const view = await chatView({
host: chatHost({
activeSnapshot: vi.fn(() => [cachedThread] as never[]),
activeThreadsSnapshot: vi.fn(() => [cachedThread] as never[]),
appServerMetadataSnapshot: vi.fn(
() =>
({

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

@ -56,7 +56,7 @@ describe("threadPickerSuggestions", () => {
it("uses the native searching empty state until the complete inventory resolves", async () => {
const pending = deferred<readonly Thread[]>();
const host = threadPickerHost([thread({ id: "recent" })]);
host.threadCatalog.searchActive = () => pending.promise;
host.threadCatalog.fetchActiveThreadSearchInventory = () => pending.promise;
const modal = await openedThreadPicker(host);
const suggestions = modal.getSuggestions("needle");
@ -71,7 +71,7 @@ describe("threadPickerSuggestions", () => {
it("clears the searching empty state when the query is cleared", async () => {
const pending = deferred<readonly Thread[]>();
const host = threadPickerHost([thread({ id: "recent" })]);
host.threadCatalog.searchActive = () => pending.promise;
host.threadCatalog.fetchActiveThreadSearchInventory = () => pending.promise;
const modal = await openedThreadPicker(host);
const suggestions = modal.getSuggestions("needle");
@ -142,7 +142,7 @@ describe("thread picker lifecycle", () => {
it("finishes when the initial inventory fails to load", async () => {
const host = threadPickerHost([], [], [], false);
host.threadCatalog.loadActive = async () => {
host.threadCatalog.fetchActiveThreads = async () => {
throw new Error("inventory unavailable");
};
const onClosed = vi.fn();
@ -156,7 +156,7 @@ describe("thread picker lifecycle", () => {
it("does not open after being replaced during the initial inventory load", async () => {
const pending = deferred<readonly Thread[]>();
const host = threadPickerHost([], [], [], false);
host.threadCatalog.loadActive = () => pending.promise;
host.threadCatalog.fetchActiveThreads = () => pending.promise;
const open = vi.spyOn(SuggestModal.prototype, "open");
const onClosed = vi.fn();
@ -245,20 +245,20 @@ function threadPickerHost(
completeHistoryLoads: 0,
sharedRefreshes: 0,
threadCatalog: {
activeSnapshot: () => (hasSharedSnapshot ? firstPage : null),
recentActiveSnapshot: () => (hasSharedSnapshot ? firstPage : null),
loadActive: async () => loadedThreads,
searchActive: async () => {
activeThreadsSnapshot: () => (hasSharedSnapshot ? firstPage : null),
recentActiveThreadsSnapshot: () => (hasSharedSnapshot ? firstPage : null),
fetchActiveThreads: async () => loadedThreads,
fetchActiveThreadSearchInventory: async () => {
host.completeHistoryLoads += 1;
return completeHistory;
},
refreshActive: async () => {
refreshActiveThreads: async () => {
host.sharedRefreshes += 1;
return loadedThreads;
},
hasMoreActive: () => false,
loadMoreActive: async () => loadedThreads,
observeActive: () => () => undefined,
hasMoreActiveThreads: () => false,
loadMoreActiveThreads: async () => loadedThreads,
observeActiveThreadsResult: () => () => undefined,
},
openThreadInCurrentView: async (threadId) => {
openedCurrent.push(threadId);

96
tests/features/threads-view/view.test.ts
View file

@ -159,9 +159,9 @@ describe("CodexThreadsView", () => {
const view = await threadsView(
threadsHost({
threadCatalog: {
loadActive: vi.fn().mockResolvedValue([returned]),
refreshActive: vi.fn().mockResolvedValue([returned]),
observeActive: vi.fn((listener: (result: ObservedPaginatedResult<readonly Thread[]>) => void) => {
fetchActiveThreads: vi.fn().mockResolvedValue([returned]),
refreshActiveThreads: vi.fn().mockResolvedValue([returned]),
observeActiveThreadsResult: vi.fn((listener: (result: ObservedPaginatedResult<readonly Thread[]>) => void) => {
observer = listener;
return () => undefined;
}),
@ -238,9 +238,9 @@ describe("CodexThreadsView", () => {
const view = await threadsView(
threadsHost({
threadCatalog: {
refreshActive: vi.fn(async () => [first]),
hasMoreActive: vi.fn(() => hasMore),
loadMoreActive,
refreshActiveThreads: vi.fn(async () => [first]),
hasMoreActiveThreads: vi.fn(() => hasMore),
loadMoreActiveThreads: loadMoreActive,
},
}),
);
@ -263,10 +263,10 @@ describe("CodexThreadsView", () => {
const view = await threadsView(
threadsHost({
threadCatalog: {
activeSnapshot: vi.fn(() => [first]),
loadActive: vi.fn().mockResolvedValue([first]),
hasMoreActive: vi.fn(() => true),
loadMoreActive,
activeThreadsSnapshot: vi.fn(() => [first]),
fetchActiveThreads: vi.fn().mockResolvedValue([first]),
hasMoreActiveThreads: vi.fn(() => true),
loadMoreActiveThreads: loadMoreActive,
},
}),
);
@ -286,7 +286,7 @@ describe("CodexThreadsView", () => {
});
const host = threadsHost({
threadCatalog: {
refreshActive: refresh,
refreshActiveThreads: refresh,
},
});
const view = await threadsView(host);
@ -309,7 +309,7 @@ describe("CodexThreadsView", () => {
const view = await threadsView(
threadsHost({
threadCatalog: {
activeSnapshot: vi.fn(() => [threadFixture({ id: "cached", preview: "Cached thread" })]),
activeThreadsSnapshot: vi.fn(() => [threadFixture({ id: "cached", preview: "Cached thread" })]),
},
}),
);
@ -322,13 +322,13 @@ describe("CodexThreadsView", () => {
const view = await threadsView(
threadsHost({
threadCatalog: {
refreshActive: vi.fn(
refreshActiveThreads: vi.fn(
() =>
new Promise(() => {
// Keep the initial refresh pending; this test drives observed query results directly.
}),
),
observeActive: vi.fn((listener: (result: ObservedPaginatedResult<readonly Thread[]>) => void) => {
observeActiveThreadsResult: vi.fn((listener: (result: ObservedPaginatedResult<readonly Thread[]>) => void) => {
observedThreads = listener;
return () => undefined;
}),
@ -349,13 +349,13 @@ describe("CodexThreadsView", () => {
const view = await threadsView(
threadsHost({
threadCatalog: {
loadActive: vi.fn(
fetchActiveThreads: vi.fn(
() =>
new Promise(() => {
// Drive the shared query state directly.
}),
),
observeActive: vi.fn((listener: (result: ObservedPaginatedResult<readonly Thread[]>) => void) => {
observeActiveThreadsResult: vi.fn((listener: (result: ObservedPaginatedResult<readonly Thread[]>) => void) => {
observedThreads = listener;
return () => undefined;
}),
@ -377,9 +377,9 @@ describe("CodexThreadsView", () => {
"thread/list": vi.fn().mockResolvedValue({ data: [threadFixture({ id: "thread", preview: "Thread preview" })] }),
"thread/archive": archiveThread,
});
const applyThreadCatalogEvent = vi.fn();
const applyThreadOperationEvent = vi.fn();
const host = threadsHost({
threadCatalog: { apply: applyThreadCatalogEvent },
threadEvents: { apply: applyThreadOperationEvent },
});
const view = await threadsView(host);
@ -389,7 +389,7 @@ describe("CodexThreadsView", () => {
await waitForAsyncWork(() => {
expect(archiveThread).toHaveBeenCalledWith({ threadId: "thread" });
expect(applyThreadCatalogEvent).toHaveBeenCalledWith({
expect(applyThreadOperationEvent).toHaveBeenCalledWith({
type: "thread-archived",
threadId: "thread",
});
@ -421,9 +421,9 @@ describe("CodexThreadsView", () => {
"thread/list": vi.fn().mockResolvedValue({ data: [threadFixture({ id: "thread", preview: "Thread preview" })] }),
"thread/name/set": renameThreadRequest,
});
const applyThreadCatalogEvent = vi.fn();
const applyThreadOperationEvent = vi.fn();
const host = threadsHost({
threadCatalog: { apply: applyThreadCatalogEvent },
threadEvents: { apply: applyThreadOperationEvent },
});
const view = await threadsView(host);
@ -437,7 +437,7 @@ describe("CodexThreadsView", () => {
await waitForAsyncWork(() => {
expect(renameThreadRequest).toHaveBeenCalledWith({ threadId: "thread", name: "Renamed thread" });
expect(applyThreadCatalogEvent).toHaveBeenCalledWith({
expect(applyThreadOperationEvent).toHaveBeenCalledWith({
type: "thread-renamed",
threadId: "thread",
name: "Renamed thread",
@ -452,9 +452,9 @@ describe("CodexThreadsView", () => {
"thread/list": vi.fn().mockResolvedValue({ data: [threadFixture({ id: "thread", preview: "Thread preview" })] }),
"thread/name/set": renameThreadRequest,
});
const applyThreadCatalogEvent = vi.fn();
const applyThreadOperationEvent = vi.fn();
const host = threadsHost({
threadCatalog: { apply: applyThreadCatalogEvent },
threadEvents: { apply: applyThreadOperationEvent },
});
const view = await threadsView(host);
@ -477,7 +477,7 @@ describe("CodexThreadsView", () => {
saved.resolve({});
await waitForAsyncWork(() => {
expect(applyThreadCatalogEvent).toHaveBeenCalledWith({
expect(applyThreadOperationEvent).toHaveBeenCalledWith({
type: "thread-renamed",
threadId: "thread",
name: "Saved title",
@ -822,7 +822,11 @@ function threadsHost(overrides: Record<string, unknown> = {}) {
"threadCatalog" in overrides && overrides["threadCatalog"] !== null && typeof overrides["threadCatalog"] === "object"
? (overrides["threadCatalog"] as Partial<ThreadsViewHost["threadCatalog"]>)
: {};
const hostOverrides = Object.fromEntries(Object.entries(overrides).filter(([key]) => key !== "threadCatalog"));
const threadEventOverrides =
"threadEvents" in overrides && overrides["threadEvents"] !== null && typeof overrides["threadEvents"] === "object"
? (overrides["threadEvents"] as Partial<ThreadsViewHost["threadEvents"]>)
: {};
const hostOverrides = Object.fromEntries(Object.entries(overrides).filter(([key]) => key !== "threadCatalog" && key !== "threadEvents"));
const activeObservers = new Set<(result: ObservedPaginatedResult<readonly Thread[]>) => void>();
const emitActive = (threads: readonly Thread[]): void => {
for (const observer of activeObservers) observer(queryResult(threads));
@ -846,6 +850,9 @@ function threadsHost(overrides: Record<string, unknown> = {}) {
vaultPath: "/vault",
threadNameMutations: createKeyedOperationQueue(),
threadOperationsTransport: createThreadOperationsTransport(clientAccess),
threadEvents: {
apply: threadEventOverrides.apply ?? vi.fn(),
},
threadTitleTransport: createThreadTitleTransport({
clientAccess,
codexPath: "codex",
@ -858,18 +865,19 @@ function threadsHost(overrides: Record<string, unknown> = {}) {
openThreadInAvailableView: vi.fn().mockResolvedValue(undefined),
openPanelActivities: vi.fn(() => []),
threadCatalog: {
apply: vi.fn(),
hasMoreActive:
typeof threadCatalogOverrides["hasMoreActive"] === "function" ? threadCatalogOverrides["hasMoreActive"] : vi.fn(() => false),
hasMoreActiveThreads:
typeof threadCatalogOverrides["hasMoreActiveThreads"] === "function"
? threadCatalogOverrides["hasMoreActiveThreads"]
: vi.fn(() => false),
...threadCatalogOverrides,
loadMoreActive: vi.fn(async () => {
const load = threadCatalogOverrides["loadMoreActive"];
loadMoreActiveThreads: vi.fn(async () => {
const load = threadCatalogOverrides["loadMoreActiveThreads"];
const threads = typeof load === "function" ? await load() : [];
emitActive(threads);
return threads;
}),
loadActive: vi.fn(async () => {
const load = threadCatalogOverrides["loadActive"];
fetchActiveThreads: vi.fn(async () => {
const load = threadCatalogOverrides["fetchActiveThreads"];
if (typeof load === "function") {
const threads = await load();
emitActive(threads);
@ -886,8 +894,8 @@ function threadsHost(overrides: Record<string, unknown> = {}) {
emitActive(threads);
return threads;
}),
refreshActive: vi.fn(async () => {
const refresh = threadCatalogOverrides["refreshActive"];
refreshActiveThreads: vi.fn(async () => {
const refresh = threadCatalogOverrides["refreshActiveThreads"];
if (typeof refresh === "function") {
const threads = await refresh();
emitActive(threads);
@ -906,15 +914,17 @@ function threadsHost(overrides: Record<string, unknown> = {}) {
emitActive(threads);
return threads;
}),
activeSnapshot:
typeof threadCatalogOverrides["activeSnapshot"] === "function" ? threadCatalogOverrides["activeSnapshot"] : vi.fn(() => null),
recentActiveSnapshot:
typeof threadCatalogOverrides["recentActiveSnapshot"] === "function"
? threadCatalogOverrides["recentActiveSnapshot"]
activeThreadsSnapshot:
typeof threadCatalogOverrides["activeThreadsSnapshot"] === "function"
? threadCatalogOverrides["activeThreadsSnapshot"]
: vi.fn(() => null),
observeActive:
typeof threadCatalogOverrides["observeActive"] === "function"
? threadCatalogOverrides["observeActive"]
recentActiveThreadsSnapshot:
typeof threadCatalogOverrides["recentActiveThreadsSnapshot"] === "function"
? threadCatalogOverrides["recentActiveThreadsSnapshot"]
: vi.fn(() => null),
observeActiveThreadsResult:
typeof threadCatalogOverrides["observeActiveThreadsResult"] === "function"
? threadCatalogOverrides["observeActiveThreadsResult"]
: vi.fn((observer: (result: ObservedPaginatedResult<readonly Thread[]>) => void) => {
activeObservers.add(observer);
return () => activeObservers.delete(observer);

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

@ -1,155 +0,0 @@
import { describe, expect, it, vi } from "vitest";
import type { ObservedPaginatedResultListener, 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 } from "../../../../src/features/threads/catalog/thread-catalog";
describe("ThreadCatalog", () => {
it("projects rename and archive events through the shared query owner", () => {
const store = catalogStore({
active: [thread("active"), thread("other")],
archived: [thread("archived", true)],
});
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" });
expect(catalog.activeSnapshot()).toEqual([thread("other")]);
expect(catalog.archivedSnapshot()).toEqual([{ ...thread("active"), name: "Renamed", archived: true }, thread("archived", true)]);
expect(onEventApplied).toHaveBeenCalledTimes(2);
});
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" },
]);
});
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 exact rename and delete facts without inventing unknown thread records", () => {
const store = catalogStore({
active: [thread("active"), thread("kept")],
archived: [thread("deleted", true)],
});
const catalog = createThreadCatalog({ store });
catalog.apply({ type: "thread-renamed", threadId: "missing", name: "Not invented" });
catalog.apply({ type: "thread-deleted", threadId: "deleted" });
expect(catalog.activeSnapshot()).toEqual([thread("active"), thread("kept")]);
expect(catalog.archivedSnapshot()).toEqual([]);
});
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.searchActive()).toEqual([thread("active")]);
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,
hasMore: false,
isFetchingNextPage: false,
});
expect(archivedObserver).toHaveBeenCalledWith({ value: [thread("archived", true)], error: null, isFetching: false });
unsubscribeActive();
unsubscribeArchived();
});
});
interface CatalogStoreOptions {
readonly active?: readonly Thread[] | null;
readonly archived?: readonly Thread[] | null;
}
function catalogStore(options: CatalogStoreOptions = {}) {
let active = options.active ?? null;
let archived = options.archived ?? null;
const activeObservers = new Set<ObservedPaginatedResultListener<readonly Thread[]>>();
const archivedObservers = new Set<ObservedResultListener<readonly Thread[]>>();
const appliedMutations: ThreadListMutation[] = [];
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 {
appliedMutations,
activeThreadsSnapshot: () => active,
recentActiveThreadsSnapshot: () => active,
archivedThreadsSnapshot: () => archived,
fetchActiveThreadSearchInventory: async () => active ?? [],
fetchActiveThreads: async () => active ?? [],
hasMoreActiveThreads: () => false,
loadMoreActiveThreads: async () => active ?? [],
refreshActiveThreads: async () => active ?? [],
refreshArchivedThreads: async () => archived ?? [],
applyThreadListMutations: applyMutations,
observeActiveThreadsResult: (
observer: ObservedPaginatedResultListener<readonly Thread[]>,
observeOptions: { emitCurrent?: boolean } = {},
) => {
activeObservers.add(observer);
if (observeOptions.emitCurrent ?? true) {
observer({ value: active, error: null, isFetching: false, hasMore: false, isFetchingNextPage: 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 thread(id: string, archived = false, overrides: Partial<Thread> = {}): Thread {
return {
id,
preview: id,
createdAt: 1,
updatedAt: 1,
name: null,
archived,
provenance: { kind: "interactive" },
...overrides,
};
}

310
tests/features/threads/workflows/thread-operation-coordinator.test.ts Normal file
View file

@ -0,0 +1,310 @@
import { describe, expect, it, vi } from "vitest";
import { applyThreadListMutation, type ThreadListMutation } from "../../../../src/app-server/query/thread-list-mutation";
import type { Thread } from "../../../../src/domain/threads/model";
import { createThreadOperationCoordinator } from "../../../../src/features/threads/workflows/thread-operation-coordinator";
import type { ThreadLifecycleEvent } from "../../../../src/features/threads/workflows/thread-operation-event";
import { projectThreadListChanges } from "../../../../src/features/threads/workflows/thread-read-model-projection";
describe("ThreadOperationCoordinator", () => {
it("commits ordinary facts directly and a completed fork as one ordered fact batch", () => {
const commit = vi.fn();
const coordinator = createThreadOperationCoordinator(commit);
coordinator.apply({ type: "thread-renamed", threadId: "other", name: "Other" });
const publication = coordinator.beginForkPublication("source");
publication.record(thread("child"));
coordinator.apply({ type: "thread-archived", threadId: "child" });
coordinator.apply({ type: "thread-archived", threadId: "source" });
publication.finish({ sourceArchived: true });
expect(commit).toHaveBeenNthCalledWith(1, [{ type: "thread-renamed", threadId: "other", name: "Other" }]);
expect(commit).toHaveBeenNthCalledWith(2, [
{ type: "thread-upserted", thread: thread("child") },
{ type: "thread-archived", threadId: "child" },
{ type: "thread-archived", threadId: "source" },
]);
});
it("holds unclaimed fork notifications until every publication for the source finishes", () => {
const commit = vi.fn();
const coordinator = createThreadOperationCoordinator(commit);
const first = coordinator.beginForkPublication("source");
const second = coordinator.beginForkPublication("source");
first.record(thread("first"));
second.record(thread("second"));
coordinator.apply({ type: "thread-upserted", thread: thread("unclaimed"), forkedFromThreadId: "source" });
first.finish();
expect(commit).toHaveBeenCalledOnce();
expect(commit).toHaveBeenLastCalledWith([{ type: "thread-upserted", thread: thread("first") }]);
second.finish();
expect(commit).toHaveBeenNthCalledWith(2, [{ type: "thread-upserted", thread: thread("second") }]);
expect(commit).toHaveBeenNthCalledWith(3, [{ type: "thread-upserted", thread: thread("unclaimed") }]);
});
it("carries lifecycle observed before the fork response into the claimed child", () => {
const commit = vi.fn();
const coordinator = createThreadOperationCoordinator(commit);
const publication = coordinator.beginForkPublication("source");
coordinator.apply({
type: "thread-upserted",
thread: thread("child", false, { preview: "notification" }),
forkedFromThreadId: "source",
});
coordinator.apply({ type: "thread-renamed", threadId: "child", name: "External" });
coordinator.apply({ type: "thread-archived", threadId: "child" });
publication.record(thread("child", false, { preview: "response" }));
publication.finish();
expect(commit).toHaveBeenCalledOnce();
expect(commit).toHaveBeenCalledWith([
{ type: "thread-upserted", thread: thread("child", false, { preview: "response", name: "External" }) },
{ type: "thread-archived", threadId: "child" },
]);
});
it("settles source lifecycle only after every publication for the source finishes", () => {
const commit = vi.fn();
const coordinator = createThreadOperationCoordinator(commit);
const first = coordinator.beginForkPublication("source");
const second = coordinator.beginForkPublication("source");
first.record(thread("first"));
second.record(thread("second"));
coordinator.apply({ type: "thread-archived", threadId: "source" });
first.finish({ sourceArchived: true });
coordinator.apply({ type: "thread-unarchived", threadId: "source" });
second.finish();
expect(commit).toHaveBeenNthCalledWith(1, [{ type: "thread-upserted", thread: thread("first") }]);
expect(commit).toHaveBeenNthCalledWith(2, [{ type: "thread-upserted", thread: thread("second") }]);
});
it("commits a restored source even when a failed fork produced no child", () => {
const commit = vi.fn();
const coordinator = createThreadOperationCoordinator(commit);
const publication = coordinator.beginForkPublication("source");
const restored = thread("source", true, { preview: "restored" });
coordinator.apply({ type: "thread-archived", threadId: "source" });
coordinator.apply({ type: "thread-restored", thread: restored });
publication.finish({ sourceArchived: true });
expect(commit).toHaveBeenCalledWith([{ type: "thread-restored", thread: restored }]);
});
it("projects rename and archive events through the shared query owner", () => {
const store = catalogStore({
active: [thread("active"), thread("other")],
archived: [thread("archived", true)],
});
const onEventApplied = vi.fn();
const coordinator = operationCoordinatorForStore(store, onEventApplied);
coordinator.apply({ type: "thread-renamed", threadId: "active", name: "Renamed" });
coordinator.apply({ type: "thread-archived", threadId: "active" });
expect(store.activeThreadsSnapshot()).toEqual([thread("other")]);
expect(store.archivedThreadsSnapshot()).toEqual([{ ...thread("active"), name: "Renamed", archived: true }, thread("archived", true)]);
expect(onEventApplied).toHaveBeenCalledTimes(2);
});
it("publishes a fork replacement to active observers as one mutation batch", () => {
const store = catalogStore({ active: [thread("source"), thread("other")], archived: [] });
const onEventApplied = vi.fn();
const coordinator = operationCoordinatorForStore(store, onEventApplied);
const publication = coordinator.beginForkPublication("source");
coordinator.apply({
type: "thread-upserted",
thread: thread("forked"),
forkedFromThreadId: "source",
});
publication.record(thread("forked"));
coordinator.apply({ type: "thread-upserted", thread: thread("forked", false, { preview: "resumed" }) });
coordinator.apply({ type: "thread-renamed", threadId: "forked", name: "Forked title" });
coordinator.apply({ type: "thread-archived", threadId: "source" });
expect(store.appliedMutationBatches).toEqual([]);
publication.finish({ sourceArchived: true });
expect(store.appliedMutationBatches).toHaveLength(1);
expect(store.activeThreadsSnapshot()).toEqual([{ ...thread("forked"), preview: "resumed", name: "Forked title" }, thread("other")]);
expect(store.archivedThreadsSnapshot()).toEqual([thread("source", true)]);
expect(onEventApplied).toHaveBeenCalledOnce();
expect(onEventApplied).toHaveBeenCalledWith([
{
type: "thread-upserted",
thread: { ...thread("forked"), preview: "resumed", name: "Forked title" },
},
{ type: "thread-archived", threadId: "source" },
]);
});
it("publishes a normal fork once without removing its source", () => {
const store = catalogStore({ active: [thread("source")] });
const publication = operationCoordinatorForStore(store).beginForkPublication("source");
publication.record(thread("forked"));
publication.finish();
expect(store.appliedMutationBatches).toHaveLength(1);
expect(store.activeThreadsSnapshot()).toEqual([thread("forked"), thread("source")]);
});
it("keeps the source active when an unarchive supersedes a deferred archive", () => {
const store = catalogStore({ active: [thread("source")], archived: [] });
const coordinator = operationCoordinatorForStore(store);
const publication = coordinator.beginForkPublication("source");
publication.record(thread("forked"));
coordinator.apply({ type: "thread-archived", threadId: "source" });
coordinator.apply({ type: "thread-unarchived", threadId: "source" });
publication.finish({ sourceArchived: true });
expect(store.appliedMutationBatches).toHaveLength(1);
expect(store.activeThreadsSnapshot()).toEqual([thread("forked"), thread("source")]);
expect(store.archivedThreadsSnapshot()).toEqual([]);
});
it("publishes the restored source record that supersedes a deferred archive", () => {
const store = catalogStore({ active: [thread("source")], archived: [] });
const coordinator = operationCoordinatorForStore(store);
const publication = coordinator.beginForkPublication("source");
publication.record(thread("forked"));
const restoredSource = thread("source", true, { preview: "restored preview", updatedAt: 4 });
coordinator.apply({ type: "thread-archived", threadId: "source" });
coordinator.apply({ type: "thread-restored", thread: restoredSource });
publication.finish({ sourceArchived: true });
expect(store.appliedMutationBatches).toHaveLength(1);
expect(store.activeThreadsSnapshot()).toEqual([thread("forked"), { ...restoredSource, archived: false }]);
expect(store.archivedThreadsSnapshot()).toEqual([]);
});
it("does not resurrect a fork child archived before publication finishes", () => {
const store = catalogStore({ active: [thread("source")], archived: [] });
const coordinator = operationCoordinatorForStore(store);
const publication = coordinator.beginForkPublication("source");
publication.record(thread("forked"));
coordinator.apply({ type: "thread-archived", threadId: "forked" });
coordinator.apply({ type: "thread-archived", threadId: "source" });
publication.finish({ sourceArchived: true });
expect(store.appliedMutationBatches).toHaveLength(1);
expect(store.activeThreadsSnapshot()).toEqual([]);
expect(store.archivedThreadsSnapshot()).toEqual([thread("source", true), thread("forked", true)]);
});
it("does not delay unrelated catalog events during a fork publication", () => {
const store = catalogStore({ active: [thread("source")] });
const coordinator = operationCoordinatorForStore(store);
const publication = coordinator.beginForkPublication("source");
coordinator.apply({ type: "thread-upserted", thread: thread("unrelated") });
expect(store.activeThreadsSnapshot()).toEqual([thread("unrelated"), thread("source")]);
publication.finish();
});
it("does not invent an archived record when an archive event lacks a source snapshot", () => {
const store = catalogStore();
const coordinator = operationCoordinatorForStore(store);
coordinator.apply({ type: "thread-archived", threadId: "unknown" });
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 coordinator = operationCoordinatorForStore(store);
coordinator.apply({ type: "thread-restored", thread: thread("restored", true) });
coordinator.apply({ type: "thread-unarchived", threadId: "unarchived" });
expect(store.activeThreadsSnapshot()).toEqual([thread("unarchived"), thread("restored"), thread("active")]);
expect(store.archivedThreadsSnapshot()).toEqual([]);
});
it("patches exact rename and delete facts without inventing unknown thread records", () => {
const store = catalogStore({
active: [thread("active"), thread("kept")],
archived: [thread("deleted", true)],
});
const coordinator = operationCoordinatorForStore(store);
coordinator.apply({ type: "thread-renamed", threadId: "missing", name: "Not invented" });
coordinator.apply({ type: "thread-deleted", threadId: "deleted" });
expect(store.activeThreadsSnapshot()).toEqual([thread("active"), thread("kept")]);
expect(store.archivedThreadsSnapshot()).toEqual([]);
});
});
function operationCoordinatorForStore(
store: ReturnType<typeof catalogStore>,
onEventsApplied?: (events: readonly ThreadLifecycleEvent[]) => void,
) {
return createThreadOperationCoordinator((events) => {
store.applyThreadListMutations(projectThreadListChanges(store, events));
onEventsApplied?.(events);
});
}
interface CatalogStoreOptions {
readonly active?: readonly Thread[] | null;
readonly archived?: readonly Thread[] | null;
}
function catalogStore(options: CatalogStoreOptions = {}) {
let active = options.active ?? null;
let archived = options.archived ?? null;
const appliedMutations: ThreadListMutation[] = [];
const appliedMutationBatches: ThreadListMutation[][] = [];
const applyMutations = (mutations: readonly ThreadListMutation[]): void => {
appliedMutationBatches.push([...mutations]);
appliedMutations.push(...mutations);
for (const mutation of mutations) {
if (mutation.list === "active") {
active = applyThreadListMutation(active, mutation);
} else {
archived = applyThreadListMutation(archived, mutation);
}
}
};
return {
appliedMutations,
appliedMutationBatches,
activeThreadsSnapshot: () => active,
archivedThreadsSnapshot: () => archived,
applyThreadListMutations: applyMutations,
};
}
function thread(id: string, archived = false, overrides: Partial<Thread> = {}): Thread {
return {
id,
preview: id,
createdAt: 1,
updatedAt: 1,
name: null,
archived,
provenance: { kind: "interactive" },
...overrides,
};
}

2
tests/features/threads/workflows/thread-operations.test.ts
View file

@ -236,7 +236,7 @@ function operationsFixture(options: { client?: MockClient | null | (() => MockCl
vaultConfigDir: "vault-config",
},
archiveDestination: archiveDestinationFactory,
catalog,
operationEvents: catalog,
referenceThreads: () => options.referenceThreads ?? [],
notice,
};

57
tests/features/threads/workflows/thread-read-model-projection.test.ts Normal file
View file

@ -0,0 +1,57 @@
import { describe, expect, it } from "vitest";
import type { ThreadListMutation } from "../../../../src/app-server/query/thread-list-mutation";
import type { Thread } from "../../../../src/domain/threads/model";
import { projectThreadListChanges } from "../../../../src/features/threads/workflows/thread-read-model-projection";
describe("ThreadReadModelProjection", () => {
it("projects an ordered operation event batch without mutating its snapshots", () => {
const active = [thread("source"), thread("other")];
const archived: Thread[] = [];
const changes = projectThreadListChanges(
{
activeThreadsSnapshot: () => active,
archivedThreadsSnapshot: () => archived,
},
[
{ type: "thread-upserted", thread: thread("child") },
{ type: "thread-archived", threadId: "child" },
{ type: "thread-archived", threadId: "source" },
],
);
expect(changes).toEqual([
{ kind: "upsert", list: "active", thread: thread("child") },
{ kind: "remove", list: "active", threadId: "child" },
{ kind: "upsert", list: "archived", thread: thread("child", true) },
{ kind: "remove", list: "active", threadId: "source" },
{ kind: "upsert", list: "archived", thread: thread("source", true) },
] satisfies ThreadListMutation[]);
expect(active).toEqual([thread("source"), thread("other")]);
expect(archived).toEqual([]);
});
it("falls back to refresh when an event needs a record absent from the snapshot", () => {
expect(
projectThreadListChanges({ activeThreadsSnapshot: () => null, archivedThreadsSnapshot: () => null }, [
{ type: "thread-archived", threadId: "unknown" },
]),
).toEqual([
{ kind: "remove", list: "active", threadId: "unknown" },
{ kind: "refresh", list: "archived" },
]);
});
});
function thread(id: string, archived = false): Thread {
return {
id,
preview: id,
name: null,
archived,
createdAt: 1,
updatedAt: 1,
provenance: { kind: "interactive" },
};
}

24
tests/plugin-runtime.integration.test.ts
View file

@ -34,6 +34,10 @@ function threadCatalog(plugin: CodexPanelPlugin) {
return currentChatHost(plugin).threadCatalog;
}
function threadOperationCoordinator(plugin: CodexPanelPlugin) {
return currentChatHost(plugin).threadOperationCoordinator;
}
const capturedChatHosts = new WeakMap<CodexPanelPlugin, { current: CodexChatHost | null }>();
function currentChatHost(plugin: CodexPanelPlugin): CodexChatHost {
@ -75,7 +79,7 @@ describe("CodexPanelPlugin runtime integration", () => {
const secondRefresh = vi.spyOn((secondLeaf.view as CodexChatView).surface, "refreshSharedThreads").mockResolvedValue(undefined);
const plugin = await pluginWithLeaves([firstLeaf, secondLeaf]);
threadCatalog(plugin).apply({ type: "thread-archived", threadId: "thread-1" });
threadOperationCoordinator(plugin).apply({ type: "thread-archived", threadId: "thread-1" });
expect(firstArchived).toHaveBeenCalledWith("thread-1");
expect(secondArchived).toHaveBeenCalledWith("thread-1");
@ -99,7 +103,7 @@ describe("CodexPanelPlugin runtime integration", () => {
vi.spyOn((otherLeaf.view as CodexChatView).surface, "refreshSharedThreads").mockResolvedValue(undefined);
const plugin = await pluginWithLeaves([restoredMatchingLeaf, matchingLeaf, otherLeaf]);
threadCatalog(plugin).apply({ type: "thread-archived", threadId: "thread-1" });
threadOperationCoordinator(plugin).apply({ type: "thread-archived", threadId: "thread-1" });
expect(matchingArchived).toHaveBeenCalledWith("thread-1");
expect(restoredMatchingLeaf.detach).not.toHaveBeenCalled();
@ -126,7 +130,7 @@ describe("CodexPanelPlugin runtime integration", () => {
const secondRefresh = vi.spyOn((secondLeaf.view as CodexChatView).surface, "refreshSharedThreads").mockResolvedValue(undefined);
const plugin = await pluginWithLeaves([firstLeaf, secondLeaf]);
threadCatalog(plugin).apply({ type: "thread-renamed", threadId: "thread-1", name: "Renamed thread" });
threadOperationCoordinator(plugin).apply({ type: "thread-renamed", threadId: "thread-1", name: "Renamed thread" });
expect(firstRenamed).toHaveBeenCalledWith("thread-1", "Renamed thread");
expect(secondRenamed).toHaveBeenCalledWith("thread-1", "Renamed thread");
@ -151,22 +155,22 @@ describe("CodexPanelPlugin runtime integration", () => {
),
);
const first = threadCatalog(plugin).refreshActive();
const first = threadCatalog(plugin).refreshActiveThreads();
const staleFirst = expect(first).rejects.toThrow("Codex execution runtime was disposed while work was in progress.");
await flushMicrotasks();
await publishCodexPath(plugin, "codex-b");
const second = threadCatalog(plugin).refreshActive();
const second = threadCatalog(plugin).refreshActiveThreads();
await flushMicrotasks();
expect(withShortLivedAppServerClientMock).toHaveBeenCalledTimes(2);
await expect(second).resolves.toEqual([thread("second")]);
expect(threadCatalog(plugin).activeSnapshot()).toEqual([thread("second")]);
expect(threadCatalog(plugin).activeThreadsSnapshot()).toEqual([thread("second")]);
resolveFirst([thread("first")]);
await staleFirst;
expect(threadCatalog(plugin).activeSnapshot()).toEqual([thread("second")]);
expect(threadCatalog(plugin).activeThreadsSnapshot()).toEqual([thread("second")]);
await publishCodexPath(plugin, "codex-a");
expect(threadCatalog(plugin).activeSnapshot()).toBeNull();
expect(threadCatalog(plugin).activeThreadsSnapshot()).toBeNull();
});
it("runs shared queries through a runtime-owned short-lived client", async () => {
@ -177,7 +181,7 @@ describe("CodexPanelPlugin runtime integration", () => {
const plugin = await pluginWithLeaves([]);
await publishCodexPath(plugin, "codex-b");
await expect(threadCatalog(plugin).refreshActive()).resolves.toEqual([thread("matching-context")]);
await expect(threadCatalog(plugin).refreshActiveThreads()).resolves.toEqual([thread("matching-context")]);
expect(withShortLivedAppServerClientMock).toHaveBeenCalledWith(
"codex-b",
@ -186,7 +190,7 @@ describe("CodexPanelPlugin runtime integration", () => {
{},
expect.objectContaining({ created: expect.any(Function), disposed: expect.any(Function) }),
);
expect(threadCatalog(plugin).activeSnapshot()).toEqual([thread("matching-context")]);
expect(threadCatalog(plugin).activeThreadsSnapshot()).toEqual([thread("matching-context")]);
});
it("uses a short-lived client when the operation declares an unhandled server-request policy", async () => {

13
tests/settings/test-support.ts
View file

@ -6,7 +6,7 @@ import type { CatalogHookMetadata, CatalogModel } from "../../src/app-server/pro
import type { ThreadRecord } from "../../src/app-server/protocol/thread";
import type { ModelMetadata, ReasoningEffort } from "../../src/domain/catalog/metadata";
import type { Thread } from "../../src/domain/threads/model";
import type { ThreadCatalogEvent } from "../../src/features/threads/catalog/thread-catalog";
import type { ThreadOperationEvent } from "../../src/features/threads/workflows/thread-operation-event";
import { createSettingsAppServerDynamicData } from "../../src/settings/app-server-dynamic-data";
import type { SettingsDynamicDataAccess } from "../../src/settings/dynamic-data";
import type { CodexPanelSettingTabHost } from "../../src/settings/host";
@ -194,7 +194,7 @@ export interface SettingsTabHostOptions {
archivedSnapshot?: Thread[] | null;
refreshArchived?: () => Promise<readonly Thread[]>;
observeArchived?: SettingsDynamicDataAccess["observeArchivedThreadsResult"];
applyThreadCatalogEvent?: (event: ThreadCatalogEvent) => void;
applyThreadCatalogEvent?: (event: ThreadOperationEvent) => void;
dynamicData?: SettingsDynamicDataAccess;
settings?: Partial<{
threadNamingModel: string | null;
@ -222,11 +222,11 @@ export function settingsTabHost(options: SettingsTabHostOptions = {}): CodexPane
observeModelsResult: options.observeModels ?? vi.fn(() => () => undefined),
};
const threadCatalog = {
archivedSnapshot: vi.fn(() => options.archivedSnapshot ?? null),
refreshArchived: options.refreshArchived ?? vi.fn().mockResolvedValue(options.archivedThreads ?? defaultArchivedThreads),
observeArchived: options.observeArchived ?? vi.fn(() => () => undefined),
apply: options.applyThreadCatalogEvent ?? vi.fn(),
archivedThreadsSnapshot: vi.fn(() => options.archivedSnapshot ?? null),
refreshArchivedThreads: options.refreshArchived ?? vi.fn().mockResolvedValue(options.archivedThreads ?? defaultArchivedThreads),
observeArchivedThreadsResult: options.observeArchived ?? vi.fn(() => () => undefined),
};
const threadEvents = { apply: options.applyThreadCatalogEvent ?? vi.fn() };
const clientAccess = {
withClient: async <T>(operation: (client: AppServerClient) => Promise<T>, clientOptions?: AppServerClientAccessOptions): Promise<T> => {
const contextKey = appServerQueries.contextKey();
@ -249,6 +249,7 @@ export function settingsTabHost(options: SettingsTabHostOptions = {}): CodexPane
clientAccess,
appServerQueries,
threadCatalog,
threadEvents,
});
let dynamicData = options.dynamicData ?? createDynamicData();
const host: CodexPanelSettingTabHost = {

17
tests/support/plugin-fixtures.ts
View file

@ -158,14 +158,17 @@ function chatHostFixture(): CodexChatHost {
observeModelsResult: vi.fn(() => () => undefined),
},
threadCatalog: {
fetchActiveThreads: vi.fn(() => Promise.resolve([])),
refreshActiveThreads: vi.fn(() => Promise.resolve([])),
activeThreadsSnapshot: vi.fn(() => null),
recentActiveThreadsSnapshot: vi.fn(() => null),
hasMoreActiveThreads: vi.fn(() => false),
loadMoreActiveThreads: vi.fn(() => Promise.resolve([])),
observeActiveThreadsResult: vi.fn(() => () => undefined),
},
threadOperationCoordinator: {
apply: vi.fn(),
loadActive: vi.fn(() => Promise.resolve([])),
refreshActive: vi.fn(() => Promise.resolve([])),
activeSnapshot: vi.fn(() => null),
recentActiveSnapshot: vi.fn(() => null),
hasMoreActive: vi.fn(() => false),
loadMoreActive: vi.fn(() => Promise.resolve([])),
observeActive: vi.fn(() => () => undefined),
beginForkPublication: vi.fn(() => ({ record: vi.fn(), finish: vi.fn() })),
},
};
}