From 05ea87643aba626500df903635cc4a778d50a220 Mon Sep 17 00:00:00 2001 From: murashit Date: Wed, 22 Jul 2026 11:19:13 +0900 Subject: [PATCH] refactor(threads): coordinate fork publication atomically --- docs/design.md | 2 + src/execution-runtime.ts | 23 +- .../chat/app-server/inbound/handler.ts | 8 +- .../app-server/inbound/notification-plan.ts | 26 +- .../threads/thread-management-actions.ts | 34 +- .../chat/host/bundles/connection-bundle.ts | 6 +- .../chat/host/bundles/shell-bundle.ts | 2 +- .../chat/host/bundles/thread-bundle.ts | 8 +- src/features/chat/host/contracts.ts | 6 +- src/features/chat/host/session-runtime.ts | 2 +- .../chat/host/session/shared-state-binding.ts | 12 +- src/features/thread-picker/modal.obsidian.ts | 10 +- src/features/threads-view/session.ts | 22 +- .../threads/catalog/thread-catalog.ts | 117 +------ .../workflows/thread-operation-coordinator.ts | 184 +++++++++++ .../workflows/thread-operation-event.ts | 19 ++ .../threads/workflows/thread-operations.ts | 8 +- .../workflows/thread-read-model-projection.ts | 73 +++++ src/plugin-runtime.ts | 12 +- src/settings/app-server-dynamic-data.ts | 19 +- tests/execution-runtime.test.ts | 2 +- .../chat/app-server/inbound/handler.test.ts | 32 +- .../threads/auto-title-coordinator.test.ts | 2 +- .../threads/thread-management-actions.test.ts | 41 ++- .../chat/host/connection-bundle.test.ts | 2 +- .../chat/host/session-runtime.test.ts | 25 +- .../chat/host/view-connection-harness.ts | 49 ++- .../chat/host/view-restoration.test.ts | 2 +- tests/features/thread-picker/modal.test.ts | 24 +- tests/features/threads-view/view.test.ts | 96 +++--- .../threads/catalog/thread-catalog.test.ts | 155 --------- .../thread-operation-coordinator.test.ts | 310 ++++++++++++++++++ .../workflows/thread-operations.test.ts | 2 +- .../thread-read-model-projection.test.ts | 57 ++++ tests/plugin-runtime.integration.test.ts | 24 +- tests/settings/test-support.ts | 13 +- tests/support/plugin-fixtures.ts | 17 +- 37 files changed, 981 insertions(+), 465 deletions(-) create mode 100644 src/features/threads/workflows/thread-operation-coordinator.ts create mode 100644 src/features/threads/workflows/thread-operation-event.ts create mode 100644 src/features/threads/workflows/thread-read-model-projection.ts delete mode 100644 tests/features/threads/catalog/thread-catalog.test.ts create mode 100644 tests/features/threads/workflows/thread-operation-coordinator.test.ts create mode 100644 tests/features/threads/workflows/thread-read-model-projection.test.ts diff --git a/docs/design.md b/docs/design.md index 5f6f1a65..8d8bf996 100644 --- a/docs/design.md +++ b/docs/design.md @@ -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. diff --git a/src/execution-runtime.ts b/src/execution-runtime.ts index 5ccf9dd1..0dfbe5ec 100644 --- a/src/execution-runtime.ts +++ b/src/execution-runtime.ts @@ -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; openThreadInCurrentView(threadId: string): Promise; openThreadInAvailableView(threadId: string): Promise; @@ -47,6 +53,7 @@ export class CodexExecutionRuntime implements AppServerClientAccess { private readonly context: Readonly; private readonly appServerQueries: AppServerQueryCache; private readonly threadCatalog: ThreadCatalog; + private readonly threadOperationCoordinator: ThreadOperationCoordinator; readonly settingsDynamicData: SettingsDynamicDataAccess; private readonly threadNameMutations = createKeyedOperationQueue(); 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(), diff --git a/src/features/chat/app-server/inbound/handler.ts b/src/features/chat/app-server/inbound/handler.ts index 3e0d8bcf..21b91a60 100644 --- a/src/features/chat/app-server/inbound/handler.ts +++ b/src/features/chat/app-server/inbound/handler.ts @@ -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; } } diff --git a/src/features/chat/app-server/inbound/notification-plan.ts b/src/features/chat/app-server/inbound/notification-plan.ts index ffc1bc92..5c1e95f8 100644 --- a/src/features/chat/app-server/inbound/notification-plan.ts +++ b/src/features/chat/app-server/inbound/notification-plan.ts @@ -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 }; diff --git a/src/features/chat/application/threads/thread-management-actions.ts b/src/features/chat/application/threads/thread-management-actions.ts index 1d72672f..94373a5f 100644 --- a/src/features/chat/application/threads/thread-management-actions.ts +++ b/src/features/chat/application/threads/thread-management-actions.ts @@ -23,10 +23,15 @@ export interface ThreadManagementActionsHost { setComposerText: (text: string) => void; openThreadInNewView: (threadId: string) => Promise; 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; archiveThread(threadId: string, options?: { saveMarkdown?: boolean }): Promise; @@ -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 { +): Promise { 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( diff --git a/src/features/chat/host/bundles/connection-bundle.ts b/src/features/chat/host/bundles/connection-bundle.ts index 3534d841..56c7f0a9 100644 --- a/src/features/chat/host/bundles/connection-bundle.ts +++ b/src/features/chat/host/bundles/connection-bundle.ts @@ -135,7 +135,7 @@ export function createConnectionBundle( appServerMetadataSnapshot: () => environment.plugin.appServerQueries.appServerMetadataSnapshot(), }); const refreshSharedThreads = async (): Promise => { - 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), diff --git a/src/features/chat/host/bundles/shell-bundle.ts b/src/features/chat/host/bundles/shell-bundle.ts index 124328ca..736574d8 100644 --- a/src/features/chat/host/bundles/shell-bundle.ts +++ b/src/features/chat/host/bundles/shell-bundle.ts @@ -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; diff --git a/src/features/chat/host/bundles/thread-bundle.ts b/src/features/chat/host/bundles/thread-bundle.ts index 5ad71591..a9414127 100644 --- a/src/features/chat/host/bundles/thread-bundle.ts +++ b/src/features/chat/host/bundles/thread-bundle.ts @@ -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), diff --git a/src/features/chat/host/contracts.ts b/src/features/chat/host/contracts.ts index 69194722..47b26593 100644 --- a/src/features/chat/host/contracts.ts +++ b/src/features/chat/host/contracts.ts @@ -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; readonly threadTitleTransport: ThreadTitleTransport; readonly threadGoalOperations: ThreadGoalOperationCoordinator; @@ -47,7 +49,7 @@ export interface WorkspacePanels { openSideChat(sourceThreadId: string, sourceThreadTitle: string | null): Promise; } -type ChatThreadCatalog = ThreadCatalogPaginatedActiveReader & ThreadCatalogEventSink; +type ChatThreadCatalog = ThreadCatalogPaginatedActiveReader; interface ChatAppServerQueries { appServerMetadataSnapshot(): SharedServerMetadata | null; diff --git a/src/features/chat/host/session-runtime.ts b/src/features/chat/host/session-runtime.ts index 907b2934..a5f72dcb 100644 --- a/src/features/chat/host/session-runtime.ts +++ b/src/features/chat/host/session-runtime.ts @@ -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); diff --git a/src/features/chat/host/session/shared-state-binding.ts b/src/features/chat/host/session/shared-state-binding.ts index 64b3cd7c..208550c7 100644 --- a/src/features/chat/host/session/shared-state-binding.ts +++ b/src/features/chat/host/session/shared-state-binding.ts @@ -6,9 +6,9 @@ import type { ChatStateStore } from "../../application/state/store"; type ThreadObserver = (result: ObservedPaginatedResult) => 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), ); }, diff --git a/src/features/thread-picker/modal.obsidian.ts b/src/features/thread-picker/modal.obsidian.ts index d2ebde8a..af922f45 100644 --- a/src/features/thread-picker/modal.obsidian.ts +++ b/src/features/thread-picker/modal.obsidian.ts @@ -34,11 +34,11 @@ export function openThreadPicker(host: ThreadPickerHost, onClosed: () => void): }; const loadAndOpen = async (): Promise => { 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 { private async loadCompleteThreadList(): Promise { 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)]); diff --git a/src/features/threads-view/session.ts b/src/features/threads-view/session.ts index 94824f7a..533a48ba 100644 --- a/src/features/threads-view/session.ts +++ b/src/features/threads-view/session.ts @@ -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; 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 { - await this.requestThreads(() => this.host.threadCatalog.refreshActive()); + await this.requestThreads(() => this.host.threadCatalog.refreshActiveThreads()); } private async load(): Promise { - await this.requestThreads(() => this.host.threadCatalog.loadActive()); + await this.requestThreads(() => this.host.threadCatalog.fetchActiveThreads()); } private async requestThreads(request: () => Promise): Promise { @@ -151,9 +153,9 @@ export class ThreadsViewSession { async loadMore(): Promise { 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(), diff --git a/src/features/threads/catalog/thread-catalog.ts b/src/features/threads/catalog/thread-catalog.ts index 96f3258d..b7b54f51 100644 --- a/src/features/threads/catalog/thread-catalog.ts +++ b/src/features/threads/catalog/thread-catalog.ts @@ -5,129 +5,32 @@ import type { Thread } from "../../../domain/threads/model"; type ActiveThreadListObserver = ObservedPaginatedResultListener; type ArchivedThreadListObserver = ObservedResultListener; -interface ThreadCatalogStore { +export interface ThreadCatalogPaginatedActiveReader { activeThreadsSnapshot(): readonly Thread[] | null; recentActiveThreadsSnapshot(): readonly Thread[] | null; - archivedThreadsSnapshot(): readonly Thread[] | null; - fetchActiveThreadSearchInventory(): Promise; fetchActiveThreads(): Promise; + refreshActiveThreads(): Promise; + observeActiveThreadsResult(observer: ActiveThreadListObserver, options?: { emitCurrent?: boolean }): () => void; hasMoreActiveThreads(): boolean; loadMoreActiveThreads(): Promise; - refreshActiveThreads(): Promise; - refreshArchivedThreads(): Promise; - 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; - refreshActive(): Promise; - observeActive(observer: ActiveThreadListObserver, options?: { emitCurrent?: boolean }): () => void; - hasMoreActive(): boolean; - loadMoreActive(): Promise; } export interface ThreadCatalogSearchReader { - searchActive(): Promise; + fetchActiveThreadSearchInventory(): Promise; } export interface ThreadCatalogArchivedReader { - archivedSnapshot(): readonly Thread[] | null; - refreshArchived(): Promise; - observeArchived(observer: ArchivedThreadListObserver, options?: { emitCurrent?: boolean }): () => void; + archivedThreadsSnapshot(): readonly Thread[] | null; + refreshArchivedThreads(): Promise; + 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 {} diff --git a/src/features/threads/workflows/thread-operation-coordinator.ts b/src/features/threads/workflows/thread-operation-coordinator.ts new file mode 100644 index 00000000..8bbed4fa --- /dev/null +++ b/src/features/threads/workflows/thread-operation-coordinator.ts @@ -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; + unclaimedChildren: Map; + sourceState: "active" | "archived" | null; + restoredSource: Thread | null; +} + +export function createThreadOperationCoordinator(committer: ThreadOperationCommitter): ThreadOperationCoordinator { + const sourceGroups = new Map(); + const pendingChildrenByThread = new Map(); + + 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; +} diff --git a/src/features/threads/workflows/thread-operation-event.ts b/src/features/threads/workflows/thread-operation-event.ts new file mode 100644 index 00000000..e9bc751f --- /dev/null +++ b/src/features/threads/workflows/thread-operation-event.ts @@ -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 + | { type: "thread-upserted"; thread: Thread; forkedFromThreadId?: string | null }; + +export interface ThreadOperationEventSink { + apply(event: ThreadOperationEvent): void; +} + +export type ThreadOperationCommitter = (events: readonly ThreadLifecycleEvent[]) => void; diff --git a/src/features/threads/workflows/thread-operations.ts b/src/features/threads/workflows/thread-operations.ts index 29d6264f..fe5301bf 100644 --- a/src/features/threads/workflows/thread-operations.ts +++ b/src/features/threads/workflows/thread-operations.ts @@ -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 }; } diff --git a/src/features/threads/workflows/thread-read-model-projection.ts b/src/features/threads/workflows/thread-read-model-projection.ts new file mode 100644 index 00000000..01f8fe49 --- /dev/null +++ b/src/features/threads/workflows/thread-read-model-projection.ts @@ -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; +} diff --git a/src/plugin-runtime.ts b/src/plugin-runtime.ts index 4309b63c..4176808a 100644 --- a/src/plugin-runtime.ts +++ b/src/plugin-runtime.ts @@ -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), diff --git a/src/settings/app-server-dynamic-data.ts b/src/settings/app-server-dynamic-data.ts index 50356f56..59a07420 100644 --- a/src/settings/app-server-dynamic-data.ts +++ b/src/settings/app-server-dynamic-data.ts @@ -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; } -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 }); }), }; } diff --git a/tests/execution-runtime.test.ts b/tests/execution-runtime.test.ts index 59a80163..2bf53eec 100644 --- a/tests/execution-runtime.test.ts +++ b/tests/execution-runtime.test.ts @@ -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(), diff --git a/tests/features/chat/app-server/inbound/handler.test.ts b/tests/features/chat/app-server/inbound/handler.test.ts index 2e6bb6c7..edf16df9 100644 --- a/tests/features/chat/app-server/inbound/handler.test.ts +++ b/tests/features/chat/app-server/inbound/handler.test.ts @@ -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 & { currentState(): ChatState; }; -function handlerForState(state = chatStateFixture(), actions: Partial = {}): TestChatInboundHandler { +function handlerForState( + state = chatStateFixture(), + actions: Partial & { + 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); + + 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, }); }); diff --git a/tests/features/chat/application/threads/auto-title-coordinator.test.ts b/tests/features/chat/application/threads/auto-title-coordinator.test.ts index aab22aac..ff497bd2 100644 --- a/tests/features/chat/application/threads/auto-title-coordinator.test.ts +++ b/tests/features/chat/application/threads/auto-title-coordinator.test.ts @@ -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({ diff --git a/tests/features/chat/application/threads/thread-management-actions.test.ts b/tests/features/chat/application/threads/thread-management-actions.test.ts index 81de079a..89f87179 100644 --- a/tests/features/chat/application/threads/thread-management-actions.test.ts +++ b/tests/features/chat/application/threads/thread-management-actions.test.ts @@ -32,7 +32,7 @@ type ThreadManagementActionsHostMock = Omit< | "openThreadInCurrentPanel" | "openThreadInNewView" | "operations" - | "recordThread" + | "beginThreadForkPublication" | "setComposerText" | "setStatus" | "threadTransport" @@ -44,7 +44,11 @@ type ThreadManagementActionsHostMock = Omit< setComposerText: Mock; openThreadInNewView: Mock; openThreadInCurrentPanel: Mock; - recordThread: Mock; + forkPublication: { + record: Mock["record"]>; + finish: Mock["finish"]>; + }; + beginThreadForkPublication: Mock; }; 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().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(), + forkPublication, + beginThreadForkPublication: vi.fn().mockReturnValue(forkPublication), threadHasPendingOrRunningPanel: vi.fn(() => false), }; } diff --git a/tests/features/chat/host/connection-bundle.test.ts b/tests/features/chat/host/connection-bundle.test.ts index a526cfe4..27bf1e94 100644 --- a/tests/features/chat/host/connection-bundle.test.ts +++ b/tests/features/chat/host/connection-bundle.test.ts @@ -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" }, diff --git a/tests/features/chat/host/session-runtime.test.ts b/tests/features/chat/host/session-runtime.test.ts index 6db0fec1..efc30269 100644 --- a/tests/features/chat/host/session-runtime.test.ts +++ b/tests/features/chat/host/session-runtime.test.ts @@ -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; threadCatalog?: Partial; + threadOperationCoordinator?: Partial; appServerQueries?: Partial; 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"] { 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, }; } diff --git a/tests/features/chat/host/view-connection-harness.ts b/tests/features/chat/host/view-connection-harness.ts index 6f71fa74..7248cfde 100644 --- a/tests/features/chat/host/view-connection-harness.ts +++ b/tests/features/chat/host/view-connection-harness.ts @@ -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) => 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) => 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 }); + }, }; }, }, diff --git a/tests/features/chat/host/view-restoration.test.ts b/tests/features/chat/host/view-restoration.test.ts index 102a317e..0fe20c1b 100644 --- a/tests/features/chat/host/view-restoration.test.ts +++ b/tests/features/chat/host/view-restoration.test.ts @@ -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( () => ({ diff --git a/tests/features/thread-picker/modal.test.ts b/tests/features/thread-picker/modal.test.ts index b226f8da..93495922 100644 --- a/tests/features/thread-picker/modal.test.ts +++ b/tests/features/thread-picker/modal.test.ts @@ -56,7 +56,7 @@ describe("threadPickerSuggestions", () => { it("uses the native searching empty state until the complete inventory resolves", async () => { const pending = deferred(); 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(); 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(); 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); diff --git a/tests/features/threads-view/view.test.ts b/tests/features/threads-view/view.test.ts index 6aff253a..333dbb7e 100644 --- a/tests/features/threads-view/view.test.ts +++ b/tests/features/threads-view/view.test.ts @@ -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) => void) => { + fetchActiveThreads: vi.fn().mockResolvedValue([returned]), + refreshActiveThreads: vi.fn().mockResolvedValue([returned]), + observeActiveThreadsResult: vi.fn((listener: (result: ObservedPaginatedResult) => 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) => void) => { + observeActiveThreadsResult: vi.fn((listener: (result: ObservedPaginatedResult) => 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) => void) => { + observeActiveThreadsResult: vi.fn((listener: (result: ObservedPaginatedResult) => 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 = {}) { "threadCatalog" in overrides && overrides["threadCatalog"] !== null && typeof overrides["threadCatalog"] === "object" ? (overrides["threadCatalog"] as Partial) : {}; - 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) + : {}; + const hostOverrides = Object.fromEntries(Object.entries(overrides).filter(([key]) => key !== "threadCatalog" && key !== "threadEvents")); const activeObservers = new Set<(result: ObservedPaginatedResult) => void>(); const emitActive = (threads: readonly Thread[]): void => { for (const observer of activeObservers) observer(queryResult(threads)); @@ -846,6 +850,9 @@ function threadsHost(overrides: Record = {}) { 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 = {}) { 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 = {}) { 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 = {}) { 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) => void) => { activeObservers.add(observer); return () => activeObservers.delete(observer); diff --git a/tests/features/threads/catalog/thread-catalog.test.ts b/tests/features/threads/catalog/thread-catalog.test.ts deleted file mode 100644 index e77f6da3..00000000 --- a/tests/features/threads/catalog/thread-catalog.test.ts +++ /dev/null @@ -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>(); - const archivedObservers = new Set>(); - 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, - 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, 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 { - return { - id, - preview: id, - createdAt: 1, - updatedAt: 1, - name: null, - archived, - provenance: { kind: "interactive" }, - ...overrides, - }; -} diff --git a/tests/features/threads/workflows/thread-operation-coordinator.test.ts b/tests/features/threads/workflows/thread-operation-coordinator.test.ts new file mode 100644 index 00000000..ab3efc7f --- /dev/null +++ b/tests/features/threads/workflows/thread-operation-coordinator.test.ts @@ -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, + 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 { + return { + id, + preview: id, + createdAt: 1, + updatedAt: 1, + name: null, + archived, + provenance: { kind: "interactive" }, + ...overrides, + }; +} diff --git a/tests/features/threads/workflows/thread-operations.test.ts b/tests/features/threads/workflows/thread-operations.test.ts index 16549e2b..c06e2d6d 100644 --- a/tests/features/threads/workflows/thread-operations.test.ts +++ b/tests/features/threads/workflows/thread-operations.test.ts @@ -236,7 +236,7 @@ function operationsFixture(options: { client?: MockClient | null | (() => MockCl vaultConfigDir: "vault-config", }, archiveDestination: archiveDestinationFactory, - catalog, + operationEvents: catalog, referenceThreads: () => options.referenceThreads ?? [], notice, }; diff --git a/tests/features/threads/workflows/thread-read-model-projection.test.ts b/tests/features/threads/workflows/thread-read-model-projection.test.ts new file mode 100644 index 00000000..4a65d463 --- /dev/null +++ b/tests/features/threads/workflows/thread-read-model-projection.test.ts @@ -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" }, + }; +} diff --git a/tests/plugin-runtime.integration.test.ts b/tests/plugin-runtime.integration.test.ts index 3dba5ad3..400e70c0 100644 --- a/tests/plugin-runtime.integration.test.ts +++ b/tests/plugin-runtime.integration.test.ts @@ -34,6 +34,10 @@ function threadCatalog(plugin: CodexPanelPlugin) { return currentChatHost(plugin).threadCatalog; } +function threadOperationCoordinator(plugin: CodexPanelPlugin) { + return currentChatHost(plugin).threadOperationCoordinator; +} + const capturedChatHosts = new WeakMap(); 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 () => { diff --git a/tests/settings/test-support.ts b/tests/settings/test-support.ts index 272c5fba..63c52d25 100644 --- a/tests/settings/test-support.ts +++ b/tests/settings/test-support.ts @@ -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; 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 (operation: (client: AppServerClient) => Promise, clientOptions?: AppServerClientAccessOptions): Promise => { 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 = { diff --git a/tests/support/plugin-fixtures.ts b/tests/support/plugin-fixtures.ts index e7b78fc3..d750ada7 100644 --- a/tests/support/plugin-fixtures.ts +++ b/tests/support/plugin-fixtures.ts @@ -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() })), }, }; }