perf(cli): Cache persisted session catalogs

docs/design/session-list-persisted-catalog-cache.md

@@ -0,0 +1,31 @@
+# Session List persisted catalog cache
+
+## Problem
+
+Organized and metadata-filtered session lists must load the complete persisted session catalog before applying organization, source, ordering, and cursor rules. Concurrent `all`, `pinned`, source-filtered, and LiveTask requests currently repeat the same JSONL and worktree-sidecar reads. The synchronous portions of those scans amplify event-loop lag when a workspace has many or large transcripts.
+
+## Design
+
+The daemon keeps a process-local catalog snapshot keyed by the resolved session runtime root, the exact workspace identity, and active versus archived state. Query, group, source, cursor, trust, and live-merge options are intentionally excluded because they do not change persisted catalog contents.
+
+The first request installs an in-flight Promise before starting the loader. Concurrent requests for the same generation await that Promise. A successful catalog, including worktree sidecars, remains available for two seconds from scan completion. Organization and every live bridge field are merged after lookup on every request. Default numeric pagination and the separate session-info counter do not use the catalog cache.
+
+Each scope has a generation. Explicit metadata, close, delete, archive, and unarchive operations invalidate the affected states. An invalidated in-flight load may finish for requests that already joined it, but its generation cannot repopulate the cache. Failures are never cached and there is no stale-on-error fallback.
+
+Metadata title persistence remains asynchronous in the bridge extension method. Invalidation prevents a pre-mutation generation from being installed, while the live merge exposes the new title immediately; it does not claim that the metadata response waits for durable JSONL persistence. A scan racing that background write can still publish its older file view for the normal two-second snapshot lifetime.
+
+The cache retains at most 50,000 summaries across all workspaces. A snapshot larger than the limit is still returned to its current waiters but is not installed. Expiry timers are unreferenced and identity-checked, and the oldest completed snapshots are evicted before a new snapshot would exceed the limit.
+
+## Consistency and isolation
+
+Daemon-managed callers pass the selected runtime root explicitly, and the complete read runs in that pinned Storage context. Secondary runtimes never fall back to the primary runtime. Read-only trust policy remains request-scoped; sharing the persisted snapshot does not enable live merging or debug logging.
+
+The cache is not a filesystem transaction. Unknown writers can update a file after that file was read but before the snapshot finishes. Such a snapshot can remain visible for two seconds after publication. Live session state and organization changes are not subject to that window.
+
+## Observability
+
+Request spans distinguish physical scans, cache hits, and single-flight waiters before awaiting the shared Promise, so failures retain their cache status. Successful lookups also record archive state, query kind, summary count, scan pages, truncation, and either leader scan duration or cache age. Paths, session identifiers, titles, and source identifiers are never attached.
+
+## Out of scope
+
+This change does not alter public protocols, Web Shell polling, the session-info scan, cross-workspace scan scheduling, core filesystem APIs, or daemon timeout policy. The outer lifecycle timeout fix remains necessary for a single slow cold scan.

packages/cli/src/serve/acp-http/dispatch.ts

@@ -120,6 +120,7 @@ import {
 } from '../workspace-agents.js';
 import {
   InvalidCursorError,
+  invalidateWorkspaceSessionListCache,
   listWorkspaceSessionsForResponse,
 } from '../server.js';
 import { createSessionOrganizationService } from '../session-organization-helpers.js';
@@ -966,6 +967,27 @@ export class AcpDispatcher {
     this.agentManager = createDaemonSubagentManager(boundWorkspace);
   }
 
+  private invalidateSessionLists(
+    archiveStates: readonly SessionArchiveState[],
+  ): void {
+    invalidateWorkspaceSessionListCache({
+      runtimeBaseDir: this.sessionRuntimeBaseDir,
+      workspaceCwd: this.boundWorkspace,
+      archiveStates,
+    });
+  }
+
+  private async runWithSessionListInvalidation<T>(
+    archiveStates: readonly SessionArchiveState[],
+    mutation: () => Promise<T>,
+  ): Promise<T> {
+    try {
+      return await mutation();
+    } finally {
+      this.invalidateSessionLists(archiveStates);
+    }
+  }
+
   private removeOrphanSession(
     sessionId: string,
     removePersistedSession = false,
@@ -1957,6 +1979,7 @@ export class AcpDispatcher {
               parentSessionId,
               ...parsedSource,
             },
+            { runtimeBaseDir: this.sessionRuntimeBaseDir },
           );
           this.replyConn(conn, id, {
             sessions: result.sessions.map((s) => ({
@@ -2053,6 +2076,7 @@ export class AcpDispatcher {
             throw err;
           } finally {
             conn.closingSessions.delete(sessionId);
+            this.invalidateSessionLists(['active']);
           }
           closeLocalSessionStream();
           this.replyConn(conn, id, {});
@@ -2637,13 +2661,18 @@ export class AcpDispatcher {
             const metadata = isObject(params['metadata'])
               ? (params['metadata'] as Record<string, unknown>)
               : {};
-            const result = this.bridge.updateSessionMetadata(
-              sessionId,
-              metadata as unknown as Parameters<
-                HttpAcpBridge['updateSessionMetadata']
-              >[1],
-              this.sessionCtx(conn, sessionId, loopback),
-            );
+            let result: ReturnType<HttpAcpBridge['updateSessionMetadata']>;
+            try {
+              result = this.bridge.updateSessionMetadata(
+                sessionId,
+                metadata as unknown as Parameters<
+                  HttpAcpBridge['updateSessionMetadata']
+                >[1],
+                this.sessionCtx(conn, sessionId, loopback),
+              );
+            } finally {
+              this.invalidateSessionLists(['active']);
+            }
             this.replyConn(conn, id, result as unknown);
           });
           return;
@@ -4306,19 +4335,23 @@ export class AcpDispatcher {
           const ids = this.parseSessionIds(params);
           if (this.rejectActiveLiveSessionMutation(conn, id, ids)) return;
           const svc = new SessionService(this.boundWorkspace);
-          const result = await deleteDaemonSessions({
-            sessionIds: ids,
-            service: svc,
-            bridge: this.bridge,
-            coordinator: this.archiveCoordinator,
-            onError: ({ phase, sessionId, error }) => {
-              const safeSessionId = logSafe(sessionId.slice(0, 8));
-              const safeMessage = logSafe(error);
-              writeStderrLine(
-                `qwen serve: /acp sessions/delete ${phase}Session(${safeSessionId}) failed: ${safeMessage}`,
-              );
-            },
-          });
+          const result = await this.runWithSessionListInvalidation(
+            ['active', 'archived'],
+            () =>
+              deleteDaemonSessions({
+                sessionIds: ids,
+                service: svc,
+                bridge: this.bridge,
+                coordinator: this.archiveCoordinator,
+                onError: ({ phase, sessionId, error }) => {
+                  const safeSessionId = logSafe(sessionId.slice(0, 8));
+                  const safeMessage = logSafe(error);
+                  writeStderrLine(
+                    `qwen serve: /acp sessions/delete ${phase}Session(${safeSessionId}) failed: ${safeMessage}`,
+                  );
+                },
+              }),
+          );
           this.replyConn(conn, id, result as unknown);
           return;
         }
@@ -4329,12 +4362,16 @@ export class AcpDispatcher {
           const svc = new SessionService(this.boundWorkspace, {
             onWarning: logSessionArchiveWarning,
           });
-          const result = await archiveDaemonSessions({
-            sessionIds: ids,
-            service: svc,
-            bridge: this.bridge,
-            coordinator: this.archiveCoordinator,
-          });
+          const result = await this.runWithSessionListInvalidation(
+            ['active', 'archived'],
+            () =>
+              archiveDaemonSessions({
+                sessionIds: ids,
+                service: svc,
+                bridge: this.bridge,
+                coordinator: this.archiveCoordinator,
+              }),
+          );
           this.replyConn(conn, id, {
             archived: result.archived,
             alreadyArchived: result.alreadyArchived,
@@ -4349,11 +4386,15 @@ export class AcpDispatcher {
           const svc = new SessionService(this.boundWorkspace, {
             onWarning: logSessionArchiveWarning,
           });
-          const result = await unarchiveDaemonSessions({
-            sessionIds: ids,
-            service: svc,
-            coordinator: this.archiveCoordinator,
-          });
+          const result = await this.runWithSessionListInvalidation(
+            ['active', 'archived'],
+            () =>
+              unarchiveDaemonSessions({
+                sessionIds: ids,
+                service: svc,
+                coordinator: this.archiveCoordinator,
+              }),
+          );
           this.replyConn(conn, id, {
             unarchived: result.unarchived,
             alreadyActive: result.alreadyActive,

packages/cli/src/serve/acp-http/transport.test.ts

@@ -7390,6 +7390,93 @@ describe('ACP Streamable HTTP transport (over the wire)', () => {
       });
     });
 
+    it('session mutations invalidate active and archived organized catalogs', async () => {
+      await withRuntimeDir(async () => {
+        const sessionId = '550e8400-e29b-41d4-a716-446655440017';
+        await writeStoredSession(sessionId);
+        const connId = await initialize();
+        const stream = await openStream(connId);
+        const reader = frameReader(stream);
+
+        const list = async (
+          id: number,
+          archiveState: 'active' | 'archived',
+        ) => {
+          await post(connId, {
+            jsonrpc: '2.0',
+            id,
+            method: 'session/list',
+            params: {
+              workspaceCwd: TEST_WORKSPACE,
+              view: 'organized',
+              archiveState,
+            },
+          });
+          return reader.next();
+        };
+
+        await expect(list(75, 'active')).resolves.toMatchObject({
+          result: { sessions: [{ sessionId }] },
+        });
+        await expect(list(76, 'archived')).resolves.toMatchObject({
+          result: { sessions: [] },
+        });
+
+        await post(connId, {
+          jsonrpc: '2.0',
+          id: 77,
+          method: '_qwen/sessions/archive',
+          params: { sessionIds: [sessionId] },
+        });
+        expect(await reader.next()).toMatchObject({
+          id: 77,
+          result: { archived: [sessionId], errors: [] },
+        });
+
+        await expect(list(78, 'active')).resolves.toMatchObject({
+          result: { sessions: [] },
+        });
+        await expect(list(79, 'archived')).resolves.toMatchObject({
+          result: { sessions: [{ sessionId, isArchived: true }] },
+        });
+
+        await post(connId, {
+          jsonrpc: '2.0',
+          id: 80,
+          method: '_qwen/sessions/unarchive',
+          params: { sessionIds: [sessionId] },
+        });
+        expect(await reader.next()).toMatchObject({
+          id: 80,
+          result: { unarchived: [sessionId], errors: [] },
+        });
+        await expect(list(81, 'active')).resolves.toMatchObject({
+          result: { sessions: [{ sessionId, isArchived: false }] },
+        });
+        await expect(list(82, 'archived')).resolves.toMatchObject({
+          result: { sessions: [] },
+        });
+
+        await post(connId, {
+          jsonrpc: '2.0',
+          id: 83,
+          method: '_qwen/sessions/delete',
+          params: { sessionIds: [sessionId] },
+        });
+        expect(await reader.next()).toMatchObject({
+          id: 83,
+          result: { removed: [sessionId], errors: [] },
+        });
+        await expect(list(84, 'active')).resolves.toMatchObject({
+          result: { sessions: [] },
+        });
+        await expect(list(85, 'archived')).resolves.toMatchObject({
+          result: { sessions: [] },
+        });
+        reader.close();
+      });
+    });
+
     it('_qwen/session/update_organization assigns a color echoed by session/list', async () => {
       await withRuntimeDir(async () => {
         const sessionId = '550e8400-e29b-41d4-a716-446655440011';

packages/cli/src/serve/live/live-task-service.test.ts

@@ -4,6 +4,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+import path from 'node:path';
 import { beforeEach, describe, expect, it, vi } from 'vitest';
 import { SessionNotFoundError } from '@qwen-code/acp-bridge/bridgeErrors';
 import type {
@@ -31,6 +32,7 @@ const sessionSources = vi.hoisted(
 const removeSessionMock = vi.hoisted(() =>
   vi.fn(async (_sessionId: string) => true),
 );
+const removeSessionRuntimeBaseDirs = vi.hoisted(() => new Array<string>());
 const listWorkspaceSessionsForResponse = vi.hoisted(() => vi.fn());
 
 vi.mock('@qwen-code/qwen-code-core', async (importOriginal) => {
@@ -71,6 +73,7 @@ vi.mock('@qwen-code/qwen-code-core', async (importOriginal) => {
       }
 
       removeSession(sessionId: string) {
+        removeSessionRuntimeBaseDirs.push(actual.Storage.getRuntimeBaseDir());
         return removeSessionMock(sessionId);
       }
     },
@@ -247,13 +250,15 @@ function makeHarness() {
   const runtime = {
     workspaceId: 'conversations',
     workspaceCwd: '/conversations',
+    sessionRuntimeBaseDir: '/runtime/conversations',
     provenance: 'live-conversation',
     bridge,
   } as WorkspaceRuntime;
   const projectBridge = { ...bridge } as AcpSessionBridge;
   const projectRuntime = {
     workspaceId: 'project-1',
     workspaceCwd: '/project',
+    sessionRuntimeBaseDir: '/runtime/project',
     bridge: projectBridge,
   } as WorkspaceRuntime;
   const registry = {
@@ -291,6 +296,7 @@ function makeHarness() {
   return {
     service,
     bridge,
+    projectBridge,
     runtime,
     summaries,
     resident,
@@ -307,6 +313,7 @@ beforeEach(() => {
   parentSessions.clear();
   sessionSources.clear();
   removeSessionMock.mockClear();
+  removeSessionRuntimeBaseDirs.length = 0;
   listWorkspaceSessionsForResponse.mockReset();
   listWorkspaceSessionsForResponse.mockResolvedValue({
     sessions: [],
@@ -365,6 +372,23 @@ describe('LiveTaskService', () => {
       ],
       threads: [{ id: 'ordinary', status: 'idle', updatedAt: 1_785_369_601 }],
     });
+    expect(listWorkspaceSessionsForResponse).toHaveBeenNthCalledWith(
+      1,
+      harness.bridge,
+      '/conversations',
+      expect.objectContaining({ view: 'organized', group: 'all' }),
+      { runtimeBaseDir: '/runtime/conversations' },
+    );
+    expect(listWorkspaceSessionsForResponse).toHaveBeenNthCalledWith(
+      2,
+      harness.projectBridge,
+      '/project',
+      expect.objectContaining({ view: 'organized', group: 'all' }),
+      { runtimeBaseDir: '/runtime/project' },
+    );
+    expect(listWorkspaceSessionsForResponse.mock.calls[1]?.[0]).toBe(
+      harness.projectBridge,
+    );
     expect(harness.bridge.spawnOrAttach).not.toHaveBeenCalled();
   });
 
@@ -897,6 +921,10 @@ describe('LiveTaskService', () => {
     expect(harness.bridge.killSession).toHaveBeenCalledWith('new-task', {
       requireZeroAttaches: true,
     });
+    expect(removeSessionMock).toHaveBeenCalledWith('new-task');
+    expect(removeSessionRuntimeBaseDirs).toEqual([
+      path.resolve('/runtime/conversations'),
+    ]);
     expect(harness.sendPrompt).not.toHaveBeenCalled();
   });
 });