feat(protocol): thread/list, thread/fork and thread/archive

apps/desktop/src/lib/protocol-agent.test.ts

@@ -28,6 +28,7 @@ class FakeTransport implements ProtocolTransport {
         interactiveRequests: true,
         reviewActions: true,
         reasoningDeltas: true,
+        threadManagement: true,
         configDiagnostics: true,
         diagnosticExport: true,
         workspaceDiff: true,

apps/desktop/src/lib/protocol-agent.ts

@@ -4,6 +4,7 @@ import {
   type ProtocolEvent,
   type ProtocolMethod,
   type ReviewFindingPayload,
+  type ThreadListResult,
   type ThreadSnapshot,
   type TurnSnapshot,
   type WorkspaceDiffResult,
@@ -76,6 +77,21 @@ export class DesktopProtocolAgent {
     return { turnId: turn.id, threadId };
   }
 
+  async listThreads(): Promise<ThreadListResult | null> {
+    const initialized = await this.transport.connect();
+    // Null, not an empty list: "the server can't tell me" and "there are no
+    // threads" must not look the same to a caller deciding whether to fall back.
+    if (!initialized.capabilities.threadManagement) return null;
+    return this.transport.request<ThreadListResult>('thread/list', {});
+  }
+
+  async archiveThread(threadId: string): Promise<boolean> {
+    const initialized = await this.transport.connect();
+    if (!initialized.capabilities.threadManagement) return false;
+    await this.transport.request('thread/archive', { threadId });
+    return true;
+  }
+
   async resume(threadId: string): Promise<ThreadSnapshot> {
     await this.transport.connect();
     if (this.threadId && this.threadId !== threadId) {
@@ -363,6 +379,14 @@ export function startProtocolTurn(args: StartProtocolTurnArgs) {
   return defaultAgent.start(args);
 }
 
+export function listProtocolThreads() {
+  return defaultAgent.listThreads();
+}
+
+export function archiveProtocolThread(threadId: string) {
+  return defaultAgent.archiveThread(threadId);
+}
+
 export function resumeProtocolThread(threadId: string) {
   return defaultAgent.resume(threadId);
 }

apps/desktop/src/lib/window-shim.ts

@@ -10,6 +10,7 @@ import {
   approveProtocolRequest,
   getConfigDiagnostics,
   installProtocolAgentEmitter,
+  listProtocolThreads,
   resumeProtocolThread,
   startProtocolTurn,
 } from './protocol-agent.js';
@@ -66,6 +67,22 @@ export function installTauriShim(): void {
     },
     sessions: {
       async list() {
+        // The app-server owns the thread index. The Tauri reader stays as a
+        // fallback for a sidecar too old to serve thread/list — two readers of
+        // the same directory is what this is working away from, not toward.
+        try {
+          const listed = await listProtocolThreads();
+          if (listed) {
+            return listed.threads.map((t) => ({
+              id: t.id,
+              title: t.title,
+              cwd: t.cwd,
+              updatedAt: t.updatedAt,
+            }));
+          }
+        } catch {
+          /* fall through to the local reader */
+        }
         const rows = await listSessions();
         return rows.map((r) => ({
           id: r.id,

apps/desktop/src/preview-app.tsx

@@ -152,6 +152,7 @@ let nextThread = 1;
 let nextTurn = 1;
 let activeThreadId = MOCK_SESSIONS[0]!.id;
 let activeTurn: TurnSnapshot | null = null;
+const archivedThreads = new Set<string>();
 const protocolRequests: ProtocolRequest[] = [];
 
 // One fixture thread carries completed items so resume can be exercised against
@@ -272,11 +273,30 @@ async function handleProtocolRequest(request: ProtocolRequest): Promise<void> {
           interactiveRequests: true,
           reviewActions: true,
           reasoningDeltas: true,
+          threadManagement: true,
           workspaceDiff: true,
           configDiagnostics: true,
         },
       });
       break;
+    case 'thread/list':
+      await respond({
+        threads: MOCK_SESSIONS.filter((session) => !archivedThreads.has(session.id)).map(
+          (session) => ({
+            id: session.id,
+            cwd: '/Users/oratis/Projects/DeepCode/test',
+            createdAt: new Date(session.updated_at_secs * 1000).toISOString(),
+            updatedAt: new Date(session.updated_at_secs * 1000).toISOString(),
+            title: session.title,
+            turnCount: 1,
+          }),
+        ),
+      });
+      break;
+    case 'thread/archive':
+      archivedThreads.add(String(request.params.threadId));
+      await respond({ archived: true });
+      break;
     case 'workspace/diff':
       await respond({
         repository: true,

apps/lsp/src/handler.test.ts

@@ -23,6 +23,7 @@ const capabilities: InitializeResult = {
     interactiveRequests: true,
     reviewActions: true,
     reasoningDeltas: true,
+    threadManagement: true,
     configDiagnostics: true,
     diagnosticExport: true,
     workspaceDiff: true,

apps/server/src/server.ts

@@ -225,6 +225,12 @@ export class AppServer {
         return this.lifecycle.readThread(requiredId(request.params, 'threadId'));
       case 'thread/resume':
         return this.resumeThread(requiredId(request.params, 'threadId'));
+      case 'thread/list':
+        return this.lifecycle.listThreads();
+      case 'thread/fork':
+        return this.lifecycle.forkThread(requiredId(request.params, 'threadId'), traceId);
+      case 'thread/archive':
+        return this.lifecycle.archiveThread(requiredId(request.params, 'threadId'));
       case 'turn/start':
         return this.startTurn(request.params, traceId);
       case 'turn/interrupt':

apps/server/src/store.ts

@@ -1,4 +1,4 @@
-import { mkdir, readFile, rename, writeFile } from 'node:fs/promises';
+import { mkdir, readFile, readdir, rename, writeFile } from 'node:fs/promises';
 import { join } from 'node:path';
 import process from 'node:process';
 
@@ -35,12 +35,43 @@ export class FileThreadStore implements ThreadStore {
     await rename(temporaryPath, path);
   }
 
+  async list(): Promise<ThreadSnapshot[]> {
+    let entries: string[];
+    try {
+      entries = await readdir(this.directory);
+    } catch (error) {
+      if ((error as NodeJS.ErrnoException).code === 'ENOENT') return [];
+      throw error;
+    }
+    const threads: ThreadSnapshot[] = [];
+    for (const entry of entries) {
+      if (!entry.endsWith('.json')) continue;
+      // A snapshot written by a concurrent save can be mid-rename; skipping an
+      // unreadable one is better than failing the whole listing.
+      try {
+        threads.push(JSON.parse(await readFile(join(this.directory, entry), 'utf8')));
+      } catch {
+        continue;
+      }
+    }
+    return threads;
+  }
+
+  async archive(threadId: string): Promise<void> {
+    const path = this.pathFor(threadId);
+    const target = join(this.directory, ARCHIVED_DIR);
+    await mkdir(target, { recursive: true });
+    await rename(path, join(target, `${threadId}.json`));
+  }
+
   private pathFor(threadId: string): string {
     if (!validThreadId(threadId)) throw new Error(`Invalid thread id: ${threadId}`);
     return join(this.directory, `${threadId}.json`);
   }
 }
 
+const ARCHIVED_DIR = 'archived';
+
 /**
  * Rich protocol snapshots plus a canonical session-v1 message projection.
  *
@@ -73,6 +104,26 @@ export class CanonicalThreadStore implements ThreadStore {
     await this.sessions.materialize(metaFromThread(thread), messages);
     await this.snapshots.save(thread);
   }
+
+  /**
+   * Snapshots first, then any legacy session that has no snapshot yet — so a
+   * listing shows everything a user has, not just what the app-server has
+   * touched since 0.2.0. Legacy rows are projected lazily and not written.
+   */
+  async list(): Promise<ThreadSnapshot[]> {
+    const threads = await this.snapshots.list();
+    const seen = new Set(threads.map((thread) => thread.id));
+    for (const meta of await this.sessions.list()) {
+      if (seen.has(meta.id)) continue;
+      const session = await this.sessions.load(meta.id);
+      if (session) threads.push(threadFromSession(session.meta, session.messages));
+    }
+    return threads;
+  }
+
+  async archive(threadId: string): Promise<void> {
+    await this.snapshots.archive(threadId);
+  }
 }
 
 function metaFromThread(thread: ThreadSnapshot): SessionMeta {

apps/vscode/src/protocol-runtime.test.ts

@@ -43,6 +43,7 @@ class FakeClient {
         interactiveRequests: true,
         reviewActions: true,
         reasoningDeltas: true,
+        threadManagement: true,
         configDiagnostics: true,
         diagnosticExport: true,
         workspaceDiff: true,

packages/protocol/src/runtime.test.ts

@@ -37,6 +37,7 @@ describe('ProtocolRuntime', () => {
         interactiveRequests: true,
         reviewActions: false,
         reasoningDeltas: false,
+        threadManagement: true,
         configDiagnostics: false,
         diagnosticExport: false,
         workspaceDiff: false,
@@ -164,3 +165,80 @@ describe('ProtocolRuntime', () => {
     ]);
   });
 });
+
+describe('thread management', () => {
+  it('lists threads newest first, with a title from the first user message', async () => {
+    const store = new MemoryThreadStore();
+    const runtime = deterministicRuntime(store);
+    const older = await runtime.startThread('/a');
+    await runtime.startTurn(older.id, { text: 'older question' });
+
+    // Save a second thread with a later updatedAt than the first.
+    await store.save({
+      id: 'thread-newer',
+      cwd: '/b',
+      createdAt: '2030-01-01T00:00:00.000Z',
+      updatedAt: '2030-01-01T00:00:00.000Z',
+      turns: [],
+    });
+
+    const { threads } = await runtime.listThreads();
+    expect(threads.map((t) => t.id)).toEqual(['thread-newer', older.id]);
+    expect(threads[1]!.title).toBe('older question');
+    expect(threads[1]!.turnCount).toBe(1);
+  });
+
+  it('has no title for a thread nobody has spoken in', async () => {
+    const runtime = deterministicRuntime(new MemoryThreadStore());
+    const thread = await runtime.startThread('/a');
+    const { threads } = await runtime.listThreads();
+    expect(threads.find((t) => t.id === thread.id)?.title).toBeUndefined();
+  });
+
+  it('drops an archived thread from the listing', async () => {
+    const runtime = deterministicRuntime(new MemoryThreadStore());
+    const thread = await runtime.startThread('/a');
+    expect(await runtime.archiveThread(thread.id)).toEqual({ archived: true });
+    expect((await runtime.listThreads()).threads).toHaveLength(0);
+  });
+
+  it('refuses to archive a thread that does not exist', async () => {
+    const runtime = deterministicRuntime(new MemoryThreadStore());
+    await expect(runtime.archiveThread('nope')).rejects.toThrow(/nope/);
+  });
+
+  it('forks into a new thread and leaves the original alone', async () => {
+    const runtime = deterministicRuntime(new MemoryThreadStore());
+    const source = await runtime.startThread('/a');
+    await runtime.startTurn(source.id, { text: 'hello' });
+    await runtime.completeTurn(source.id, (await runtime.readThread(source.id)).turns[0]!.id);
+
+    const fork = await runtime.forkThread(source.id);
+    expect(fork.id).not.toBe(source.id);
+    expect(fork.cwd).toBe(source.cwd);
+    expect(fork.turns).toHaveLength(1);
+    expect((await runtime.readThread(source.id)).turns).toHaveLength(1);
+  });
+
+  it('does not carry an in-progress turn into the fork', async () => {
+    const runtime = deterministicRuntime(new MemoryThreadStore());
+    const source = await runtime.startThread('/a');
+    await runtime.startTurn(source.id, { text: 'mid-flight' });
+
+    const fork = await runtime.forkThread(source.id);
+    expect(fork.turns[0]!.status).toBe('interrupted');
+    // …so the fork can immediately start a turn of its own.
+    await expect(runtime.startTurn(fork.id, { text: 'continue' })).resolves.toBeDefined();
+  });
+
+  it('rejects listing and archiving on a store that cannot do them', async () => {
+    const bare = {
+      load: async () => null,
+      save: async () => undefined,
+    };
+    const runtime = deterministicRuntime(bare);
+    expect(runtime.initialize().capabilities.threadManagement).toBe(false);
+    await expect(runtime.listThreads()).rejects.toThrow(/cannot list/);
+    await expect(runtime.archiveThread('x')).rejects.toThrow(/cannot archive/);
+  });
+});