From c2547fd0260abe39059472e440dd51bbb6ea2f38 Mon Sep 17 00:00:00 2001 From: murashit Date: Tue, 23 Jun 2026 16:49:46 +0900 Subject: [PATCH] Stabilize thread catalog lifecycle facts --- docs/design.md | 2 +- .../chat/app-server/inbound/handler.ts | 4 + .../app-server/inbound/notification-plan.ts | 3 +- src/features/chat/host/connection-bundle.ts | 3 + src/workspace/thread-catalog.ts | 102 ++++++++++++++++-- .../features/chat/host/session-graph.test.ts | 1 + .../chat/protocol/inbound/handler.test.ts | 7 +- tests/features/chat/view-connection.test.ts | 11 ++ tests/main.test.ts | 1 + tests/workspace/thread-catalog.test.ts | 57 +++++++++- 10 files changed, 178 insertions(+), 13 deletions(-) diff --git a/docs/design.md b/docs/design.md index 591c474c..5a8a127a 100644 --- a/docs/design.md +++ b/docs/design.md @@ -54,7 +54,7 @@ Imperative DOM bridges are allowed when an external API or measurement problem r Multiple panels are separate Obsidian leaves. Treat each panel as its own Codex working surface with independent connection, thread, turn state, composer, and pending requests. -Thread history, archived state, forks, and catalog data should follow app-server semantics. Obsidian integrations such as archive note export are convenience views of Codex state, not replacements for Codex history. +Thread history, archived state, forks, and catalog data should follow app-server semantics. The thread catalog is a read model over app-server list snapshots and lifecycle facts; one list response must not discard newer app-server state. Obsidian integrations such as archive note export are convenience views of Codex state, not replacements for Codex history. Selection rewrite is intentionally scoped to a focused edit-and-review workflow. Avoid expanding it into a broader writing assistant without a separate design decision. diff --git a/src/features/chat/app-server/inbound/handler.ts b/src/features/chat/app-server/inbound/handler.ts index 5a80b14c..663d38f3 100644 --- a/src/features/chat/app-server/inbound/handler.ts +++ b/src/features/chat/app-server/inbound/handler.ts @@ -65,6 +65,7 @@ export interface ChatInboundHandlerActions { applyAppServerMetadataSnapshot: () => void; maybeNameThread: (threadId: string, turnId: string, completedSummary: ThreadConversationSummary | null) => void; recordThreadStarted: (thread: Thread) => void; + recordThreadTouched: (threadId: string, recencyAt: number | null) => void; applyThreadArchived: (threadId: string) => void; recordActiveThreadDeleted: (threadId: string) => void; applyThreadRenamed: (threadId: string, name: string | null) => void; @@ -333,6 +334,9 @@ function runNotificationEffect(context: ChatInboundHandlerContext, effect: ChatN case "record-thread-started": context.actions.recordThreadStarted(effect.thread); return; + case "record-thread-touched": + context.actions.recordThreadTouched(effect.threadId, effect.recencyAt); + return; case "apply-thread-archived": context.actions.applyThreadArchived(effect.threadId); return; diff --git a/src/features/chat/app-server/inbound/notification-plan.ts b/src/features/chat/app-server/inbound/notification-plan.ts index d36b9627..3586d16d 100644 --- a/src/features/chat/app-server/inbound/notification-plan.ts +++ b/src/features/chat/app-server/inbound/notification-plan.ts @@ -59,6 +59,7 @@ export type ChatNotificationEffect = | { type: "apply-app-server-metadata-snapshot" } | { type: "maybe-name-thread"; threadId: string; turnId: string; completedSummary: ThreadConversationSummary | null } | { type: "record-thread-started"; thread: Thread } + | { type: "record-thread-touched"; threadId: string; recencyAt: number | null } | { type: "apply-thread-archived"; threadId: string } | { type: "record-active-thread-deleted"; threadId: string } | { type: "apply-thread-renamed"; threadId: string; name: string | null } @@ -223,7 +224,7 @@ const TURN_LIFECYCLE_PLANNERS = { items: messageStreamItemsWithPendingPromptSubmitHooks(state, notification.params.turn.id), }, ], - effects: [{ type: "refresh-threads" }], + effects: [{ type: "record-thread-touched", threadId: notification.params.threadId, recencyAt: notification.params.turn.startedAt }], }), "turn/completed": (state, notification) => { if (activeTurnId(state) !== notification.params.turn.id) return EMPTY_PLAN; diff --git a/src/features/chat/host/connection-bundle.ts b/src/features/chat/host/connection-bundle.ts index bf46ca0f..fc2676e0 100644 --- a/src/features/chat/host/connection-bundle.ts +++ b/src/features/chat/host/connection-bundle.ts @@ -154,6 +154,9 @@ export function createConnectionBundle( recordThreadStarted: (thread) => { environment.plugin.threadCatalog.recordThreadStarted(thread); }, + recordThreadTouched: (threadId, recencyAt) => { + environment.plugin.threadCatalog.recordThreadTouched(threadId, recencyAt); + }, applyThreadArchived: (threadId) => { environment.plugin.threadCatalog.recordThreadArchived(threadId); }, diff --git a/src/workspace/thread-catalog.ts b/src/workspace/thread-catalog.ts index a7f83061..fc12c6c4 100644 --- a/src/workspace/thread-catalog.ts +++ b/src/workspace/thread-catalog.ts @@ -55,6 +55,10 @@ interface ThreadCatalogThreadForks { recordThreadForked(thread: Thread): void; } +interface ThreadCatalogThreadTouches { + recordThreadTouched(threadId: string, recencyAt?: number | null): void; +} + interface ThreadCatalogThreadRenames { recordThreadRenamed(threadId: string, name: string | null): void; } @@ -75,6 +79,7 @@ export interface ThreadCatalogChatEvents extends ThreadCatalogThreadStarts, ThreadCatalogThreadForks, + ThreadCatalogThreadTouches, ThreadCatalogThreadRenames, ThreadCatalogThreadArchives, ThreadCatalogThreadDeletes {} @@ -90,34 +95,48 @@ export interface ThreadCatalog ThreadCatalogThreadRestores {} export function createThreadCatalog(options: ThreadCatalogOptions): ThreadCatalog { + const activeLifecycleFacts = new Map(); + return { - activeSnapshot: () => options.queries.activeThreadsSnapshot(), - loadActive: () => options.queries.fetchActiveThreads(), - refreshActive: () => options.queries.refreshActiveThreads(), - observeActive: (observer, observeOptions) => options.queries.observeActiveThreadsResult(observer, observeOptions), + activeSnapshot: () => activeThreadsProjection(options.queries.activeThreadsSnapshot(), activeLifecycleFacts), + loadActive: () => loadActiveThreads(options.queries.fetchActiveThreads(), activeLifecycleFacts), + refreshActive: () => loadActiveThreads(options.queries.refreshActiveThreads(), activeLifecycleFacts), + observeActive: (observer, observeOptions) => + options.queries.observeActiveThreadsResult((result) => { + observer({ + ...result, + data: activeThreadsProjection(result.data, activeLifecycleFacts), + }); + }, observeOptions), archivedSnapshot: () => options.queries.archivedThreadsSnapshot(), loadArchived: () => options.queries.fetchArchivedThreads(), refreshArchived: () => options.queries.refreshArchivedThreads(), observeArchived: (observer, observeOptions) => options.queries.observeArchivedThreadsResult(observer, observeOptions), replaceActiveThreadsSnapshot: (threads) => { + activeLifecycleFacts.clear(); options.queries.setActiveThreads(threads); }, replaceArchivedThreadsSnapshot: (threads) => { options.queries.setArchivedThreads(threads); }, recordThreadStarted: (thread) => { - recordActiveThread(options.queries, thread); + recordActiveThread(options.queries, activeLifecycleFacts, thread); }, recordThreadForked: (thread) => { - recordActiveThread(options.queries, thread); + recordActiveThread(options.queries, activeLifecycleFacts, thread); + }, + recordThreadTouched: (threadId, recencyAt) => { + recordActiveThreadTouched(options.queries, activeLifecycleFacts, threadId, recencyAt); }, recordThreadRenamed: (threadId, name) => { + updateActiveLifecycleFact(activeLifecycleFacts, threadId, (thread) => ({ ...thread, name })); options.queries.updateActiveThreads((current) => current ? current.map((thread) => (thread.id === threadId ? { ...thread, name } : thread)) : null, ); options.surfaces.applyThreadRenamed(threadId, name); }, recordThreadArchived: (threadId, archiveOptions) => { + activeLifecycleFacts.delete(threadId); const archivedThread = options.queries.activeThreadsSnapshot()?.find((thread) => thread.id === threadId) ?? null; options.queries.updateActiveThreads((current) => { return current ? current.filter((thread) => thread.id !== threadId) : null; @@ -130,20 +149,87 @@ export function createThreadCatalog(options: ThreadCatalogOptions): ThreadCatalo options.surfaces.applyThreadArchived(threadId, archiveOptions); }, recordThreadDeleted: (threadId) => { + activeLifecycleFacts.delete(threadId); options.queries.updateActiveThreads((current) => (current ? current.filter((thread) => thread.id !== threadId) : null)); options.queries.updateArchivedThreads((current) => (current ? current.filter((thread) => thread.id !== threadId) : null)); }, recordThreadRestored: (thread) => { - recordActiveThread(options.queries, thread); + recordActiveThread(options.queries, activeLifecycleFacts, thread); options.queries.updateArchivedThreads((current) => (current ? current.filter((item) => item.id !== thread.id) : null)); }, }; } -function recordActiveThread(queries: ThreadCatalogQuerySource, thread: Thread): void { +async function loadActiveThreads( + threadsPromise: Promise, + activeLifecycleFacts: Map, +): Promise { + const threads = await threadsPromise; + acknowledgeActiveSnapshot(activeLifecycleFacts, threads); + return activeThreadsProjection(threads, activeLifecycleFacts) ?? []; +} + +function recordActiveThread(queries: ThreadCatalogQuerySource, activeLifecycleFacts: Map, thread: Thread): void { + promoteActiveLifecycleFact(activeLifecycleFacts, thread); queries.updateActiveThreads((current) => promoteThreadInList(current ?? [], thread)); } +function recordActiveThreadTouched( + queries: ThreadCatalogQuerySource, + activeLifecycleFacts: Map, + threadId: string, + recencyAt: number | null | undefined, +): void { + const existingFact = activeLifecycleFacts.get(threadId) ?? null; + let touchedThread = existingFact ? touchedActiveThread(existingFact, recencyAt) : null; + const nextThreads = queries.updateActiveThreads((current) => { + const currentThread = current?.find((thread) => thread.id === threadId) ?? touchedThread; + if (!currentThread) return current; + touchedThread = touchedActiveThread(currentThread, recencyAt); + return promoteThreadInList(current ?? [], touchedThread); + }); + if (!touchedThread) return; + const promotedThread = touchedThread; + promoteActiveLifecycleFact(activeLifecycleFacts, promotedThread); + if (!nextThreads) { + queries.updateActiveThreads(() => [promotedThread]); + } +} + +function activeThreadsProjection( + snapshot: readonly Thread[] | null, + activeLifecycleFacts: ReadonlyMap, +): readonly Thread[] | null { + if (!snapshot && activeLifecycleFacts.size === 0) return null; + const threads = snapshot ?? []; + if (activeLifecycleFacts.size === 0) return threads; + const snapshotThreadIds = new Set(threads.map((thread) => thread.id)); + const missingFactThreads = Array.from(activeLifecycleFacts.values()).filter((thread) => !snapshotThreadIds.has(thread.id)); + if (missingFactThreads.length === 0) return threads; + return [...missingFactThreads.reverse(), ...threads]; +} + +function acknowledgeActiveSnapshot(activeLifecycleFacts: Map, threads: readonly Thread[]): void { + for (const thread of threads) { + activeLifecycleFacts.delete(thread.id); + } +} + +function promoteActiveLifecycleFact(activeLifecycleFacts: Map, thread: Thread): void { + activeLifecycleFacts.delete(thread.id); + activeLifecycleFacts.set(thread.id, thread); +} + +function updateActiveLifecycleFact(activeLifecycleFacts: Map, threadId: string, updater: (thread: Thread) => Thread): void { + const thread = activeLifecycleFacts.get(threadId); + if (!thread) return; + promoteActiveLifecycleFact(activeLifecycleFacts, updater(thread)); +} + +function touchedActiveThread(thread: Thread, recencyAt: number | null | undefined): Thread { + return recencyAt === undefined ? thread : { ...thread, recencyAt }; +} + function promoteThreadInList(threads: readonly Thread[], thread: Thread): readonly Thread[] { const withoutThread = threads.filter((item) => item.id !== thread.id); return [thread, ...withoutThread]; diff --git a/tests/features/chat/host/session-graph.test.ts b/tests/features/chat/host/session-graph.test.ts index c84720ee..cf071fec 100644 --- a/tests/features/chat/host/session-graph.test.ts +++ b/tests/features/chat/host/session-graph.test.ts @@ -317,6 +317,7 @@ describe("createChatPanelSessionGraph actions", () => { recordThreadRenamed: vi.fn(), recordThreadStarted: vi.fn(), recordThreadForked: vi.fn(), + recordThreadTouched: vi.fn(), ...overrides, }; } diff --git a/tests/features/chat/protocol/inbound/handler.test.ts b/tests/features/chat/protocol/inbound/handler.test.ts index e780cbe6..7590d2e3 100644 --- a/tests/features/chat/protocol/inbound/handler.test.ts +++ b/tests/features/chat/protocol/inbound/handler.test.ts @@ -38,6 +38,7 @@ function handlerForState( applyAppServerMetadataSnapshot: vi.fn(), maybeNameThread: vi.fn(), recordThreadStarted: vi.fn(), + recordThreadTouched: vi.fn(), applyThreadArchived: vi.fn(), recordActiveThreadDeleted: vi.fn(), applyThreadRenamed: vi.fn(), @@ -467,7 +468,8 @@ describe("ChatInboundHandler", () => { }, ]); const refreshActiveThreads = vi.fn(); - const handler = handlerForState(state, { refreshActiveThreads }); + const recordThreadTouched = vi.fn(); + const handler = handlerForState(state, { refreshActiveThreads, recordThreadTouched }); handler.handleNotification({ method: "turn/started", @@ -489,7 +491,8 @@ describe("ChatInboundHandler", () => { expect(chatStateMessageStreamItems(handler.currentState()).map((item) => item.id)).toEqual(["local-user-1", "hook-hook-1-1"]); expect(chatStateMessageStreamItems(handler.currentState())[1]).toMatchObject({ id: "hook-hook-1-1", turnId: "turn-active" }); expect(pendingTurnStart(handler.currentState())).toBeNull(); - expect(refreshActiveThreads).toHaveBeenCalledOnce(); + expect(recordThreadTouched).toHaveBeenCalledWith("thread-active", 1); + expect(refreshActiveThreads).not.toHaveBeenCalled(); }); it("moves pre-turn hook runs after the optimistic user message when a turn id is assigned", () => { diff --git a/tests/features/chat/view-connection.test.ts b/tests/features/chat/view-connection.test.ts index fdbc0a58..352d4e7d 100644 --- a/tests/features/chat/view-connection.test.ts +++ b/tests/features/chat/view-connection.test.ts @@ -1340,6 +1340,7 @@ interface ChatHostFixtureOverrides { replaceActiveThreadsSnapshot?: ThreadCatalogSnapshotWriter["replaceActiveThreadsSnapshot"]; recordThreadStarted?: CodexChatHost["threadCatalog"]["recordThreadStarted"]; recordThreadForked?: CodexChatHost["threadCatalog"]["recordThreadForked"]; + recordThreadTouched?: CodexChatHost["threadCatalog"]["recordThreadTouched"]; updateAppServerMetadata?: CodexChatHost["appServerData"]["updateAppServerMetadata"]; refreshActive?: CodexChatHost["threadCatalog"]["refreshActive"]; activeSnapshot?: CodexChatHost["threadCatalog"]["activeSnapshot"]; @@ -1481,6 +1482,16 @@ function chatHost(overrides: ChatHostFixtureOverrides = {}): TestCodexChatHost { activeThreads = [thread, ...(activeThreads?.filter((item) => item.id !== thread.id) ?? [])]; for (const listener of activeThreadResultListeners) listener(queryResult(activeThreads)); }), + recordThreadTouched: + overrides.recordThreadTouched ?? + ((threadId, recencyAt) => { + activeThreads = + activeThreads?.map((thread) => (thread.id === threadId && recencyAt !== undefined ? { ...thread, recencyAt } : thread)) ?? + activeThreads; + if (activeThreads) { + for (const listener of activeThreadResultListeners) listener(queryResult(activeThreads)); + } + }), replaceActiveThreadsSnapshot: overrides.replaceActiveThreadsSnapshot ?? ((threads) => { diff --git a/tests/main.test.ts b/tests/main.test.ts index af0f6183..426769e6 100644 --- a/tests/main.test.ts +++ b/tests/main.test.ts @@ -838,6 +838,7 @@ function chatHostFixture(): CodexChatHost { recordThreadRenamed: vi.fn(), recordThreadStarted: vi.fn(), recordThreadForked: vi.fn(), + recordThreadTouched: vi.fn(), loadActive: vi.fn(() => Promise.resolve([])), refreshActive: vi.fn(() => Promise.resolve([])), activeSnapshot: vi.fn(() => null), diff --git a/tests/workspace/thread-catalog.test.ts b/tests/workspace/thread-catalog.test.ts index 02b8c8dc..00fc26f2 100644 --- a/tests/workspace/thread-catalog.test.ts +++ b/tests/workspace/thread-catalog.test.ts @@ -142,6 +142,51 @@ describe("ThreadCatalog", () => { expect(surfaces.applyThreadArchived).not.toHaveBeenCalled(); expect(surfaces.applyThreadRenamed).not.toHaveBeenCalled(); }); + + it("keeps app-server lifecycle threads visible until the server list acknowledges them", async () => { + const fetchThreads = vi + .fn() + .mockResolvedValueOnce([thread("other")]) + .mockResolvedValueOnce([thread("started"), thread("other")]) + .mockResolvedValueOnce([thread("other")]); + const { catalog } = catalogFixture({ fetchThreads }); + const listener = vi.fn(); + catalog.observeActive(listener); + + catalog.recordThreadStarted(thread("started")); + + await expect(catalog.refreshActive()).resolves.toEqual([thread("started"), thread("other")]); + expect(catalog.activeSnapshot()).toEqual([thread("started"), thread("other")]); + expect(observedActiveThreadIds(listener)).not.toContainEqual(["other"]); + + await expect(catalog.refreshActive()).resolves.toEqual([thread("started"), thread("other")]); + await expect(catalog.refreshActive()).resolves.toEqual([thread("other")]); + expect(catalog.activeSnapshot()).toEqual([thread("other")]); + }); + + it("records active thread touches as catalog ordering facts without open-surface broadcasts", () => { + const { catalog, surfaces } = catalogFixture(); + const listener = vi.fn(); + catalog.observeActive(listener); + catalog.replaceActiveThreadsSnapshot([ + thread("active", false, { updatedAt: 1, recencyAt: 1 }), + thread("other", false, { updatedAt: 10, recencyAt: 10 }), + ]); + + catalog.recordThreadTouched("active", 20); + + expect(catalog.activeSnapshot()).toEqual([ + thread("active", false, { updatedAt: 1, recencyAt: 20 }), + thread("other", false, { updatedAt: 10, recencyAt: 10 }), + ]); + expect(listener).toHaveBeenLastCalledWith( + expect.objectContaining({ + data: [thread("active", false, { updatedAt: 1, recencyAt: 20 }), thread("other", false, { updatedAt: 10, recencyAt: 10 })], + }), + ); + expect(surfaces.applyThreadArchived).not.toHaveBeenCalled(); + expect(surfaces.applyThreadRenamed).not.toHaveBeenCalled(); + }); }); function catalogFixture( @@ -183,7 +228,16 @@ function surfaceActions(): MockSurfaceActions { }; } -function thread(id: string, archived = false): Thread { +function observedActiveThreadIds(listener: Mock): string[][] { + return listener.mock.calls + .map((call) => { + const result = call[0] as { data: readonly Thread[] | null }; + return result.data?.map((item) => item.id) ?? null; + }) + .filter((ids): ids is string[] => ids !== null); +} + +function thread(id: string, archived = false, overrides: Partial = {}): Thread { return { id, preview: id, @@ -191,6 +245,7 @@ function thread(id: string, archived = false): Thread { updatedAt: 1, name: null, archived, + ...overrides, }; }