From 3575420670dc78adf77e7e05eb33d568bbe4ad3e Mon Sep 17 00:00:00 2001 From: t Date: Mon, 3 Aug 2026 13:55:47 +0800 Subject: [PATCH] feat(protocol): thread/list, thread/fork and thread/archive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Finding F15 in docs/THREE_WAY_REVIEW.md. The protocol could start, read and resume a thread but not enumerate one, so every client that wanted a thread picker read the session directory itself — the desktop through a Tauri command, which is a second reader of the same data with its own idea of what a row is. Adds the three methods behind a `threadManagement` capability. ThreadStore gains optional list/archive; a store without them leaves the capability off and the runtime rejects the calls rather than letting an empty list pass for the truth. The canonical store lists snapshots first and then projects 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. Fork copies a thread into a new one and leaves the original untouched. An in-progress turn is copied as `interrupted`: nothing is executing the fork, and carrying `in_progress` across would leave it permanently unable to start a turn. The desktop sidebar now lists through the protocol, falling back to the Tauri reader only when the sidecar is too old to serve it. Co-Authored-By: Claude Opus 5 --- apps/desktop/src/lib/protocol-agent.test.ts | 1 + apps/desktop/src/lib/protocol-agent.ts | 24 ++++++ apps/desktop/src/lib/window-shim.ts | 17 ++++ apps/desktop/src/preview-app.tsx | 20 +++++ apps/lsp/src/handler.test.ts | 1 + apps/server/src/server.ts | 6 ++ apps/server/src/store.ts | 53 +++++++++++- apps/vscode/src/protocol-runtime.test.ts | 1 + packages/protocol/src/runtime.test.ts | 78 ++++++++++++++++++ packages/protocol/src/runtime.ts | 89 +++++++++++++++++++++ packages/protocol/src/types.ts | 25 ++++++ 11 files changed, 314 insertions(+), 1 deletion(-) diff --git a/apps/desktop/src/lib/protocol-agent.test.ts b/apps/desktop/src/lib/protocol-agent.test.ts index dcfc1ad..ee4f256 100644 --- a/apps/desktop/src/lib/protocol-agent.test.ts +++ b/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, diff --git a/apps/desktop/src/lib/protocol-agent.ts b/apps/desktop/src/lib/protocol-agent.ts index a561268..1020015 100644 --- a/apps/desktop/src/lib/protocol-agent.ts +++ b/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 { + 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('thread/list', {}); + } + + async archiveThread(threadId: string): Promise { + 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 { 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); } diff --git a/apps/desktop/src/lib/window-shim.ts b/apps/desktop/src/lib/window-shim.ts index f4d755a..23c0c35 100644 --- a/apps/desktop/src/lib/window-shim.ts +++ b/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, diff --git a/apps/desktop/src/preview-app.tsx b/apps/desktop/src/preview-app.tsx index 528f0ea..46744b7 100644 --- a/apps/desktop/src/preview-app.tsx +++ b/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(); 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 { 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, diff --git a/apps/lsp/src/handler.test.ts b/apps/lsp/src/handler.test.ts index 6e27035..d7b60cf 100644 --- a/apps/lsp/src/handler.test.ts +++ b/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, diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 5113d9d..8498eb8 100644 --- a/apps/server/src/server.ts +++ b/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': diff --git a/apps/server/src/store.ts b/apps/server/src/store.ts index ee7599e..6c509f7 100644 --- a/apps/server/src/store.ts +++ b/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 { + 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 { + 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 { + 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 { + await this.snapshots.archive(threadId); + } } function metaFromThread(thread: ThreadSnapshot): SessionMeta { diff --git a/apps/vscode/src/protocol-runtime.test.ts b/apps/vscode/src/protocol-runtime.test.ts index 3450fe7..69d508a 100644 --- a/apps/vscode/src/protocol-runtime.test.ts +++ b/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, diff --git a/packages/protocol/src/runtime.test.ts b/packages/protocol/src/runtime.test.ts index a5834c7..4009beb 100644 --- a/packages/protocol/src/runtime.test.ts +++ b/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/); + }); +}); diff --git a/packages/protocol/src/runtime.ts b/packages/protocol/src/runtime.ts index a735906..f829a11 100644 --- a/packages/protocol/src/runtime.ts +++ b/packages/protocol/src/runtime.ts @@ -6,6 +6,7 @@ import { type InitializeResult, type ProtocolEvent, type ReasoningDeltaEvent, + type ThreadListResult, type ThreadSnapshot, type TransientDeltaEvent, type TurnSnapshot, @@ -19,10 +20,18 @@ function clone(value: T): T { export interface ThreadStore { load(threadId: string): Promise; save(thread: ThreadSnapshot): Promise; + /** + * Optional: listing and archiving need an index the base store doesn't have. + * A store without them leaves `threadManagement` off, and the runtime rejects + * the calls rather than pretending an empty list is the truth. + */ + list?(): Promise; + archive?(threadId: string): Promise; } export class MemoryThreadStore implements ThreadStore { private readonly threads = new Map(); + private readonly archived = new Set(); saveCount = 0; async load(threadId: string): Promise { @@ -34,6 +43,14 @@ export class MemoryThreadStore implements ThreadStore { this.saveCount++; this.threads.set(thread.id, clone(thread)); } + + async list(): Promise { + return [...this.threads.values()].filter((thread) => !this.archived.has(thread.id)).map(clone); + } + + async archive(threadId: string): Promise { + this.archived.add(threadId); + } } export interface ProtocolRuntimeOptions { @@ -49,6 +66,23 @@ export interface ProtocolRuntimeOptions { reasoningDeltas?: boolean; } +/** Title for a thread list row: its first user message, truncated. */ +export function threadTitle(thread: ThreadSnapshot): string | undefined { + for (const turn of thread.turns) { + for (const item of turn.items) { + if (item.type !== 'user_message') continue; + const text = item.payload.text; + if (typeof text !== 'string') continue; + const line = text + .split('\n') + .map((l) => l.trim()) + .find(Boolean); + if (line) return [...line].slice(0, 60).join(''); + } + } + return undefined; +} + export class ProtocolInvariantError extends Error { constructor(message: string) { super(message); @@ -87,6 +121,9 @@ export class ProtocolRuntime { workspaceDiff: this.options.workspaceDiff ?? false, reviewActions: this.options.reviewActions ?? false, reasoningDeltas: this.options.reasoningDeltas ?? false, + threadManagement: + typeof this.options.store.list === 'function' && + typeof this.options.store.archive === 'function', }, }; } @@ -113,6 +150,58 @@ export class ProtocolRuntime { return this.requireThread(threadId); } + async listThreads(): Promise { + const list = this.options.store.list; + if (!list) throw new ProtocolInvariantError('This store cannot list threads'); + const threads = await list.call(this.options.store); + return { + threads: threads + .map((thread) => ({ + id: thread.id, + cwd: thread.cwd, + createdAt: thread.createdAt, + updatedAt: thread.updatedAt, + title: threadTitle(thread), + turnCount: thread.turns.length, + })) + // Newest first: a picker's first row should be what you were just doing. + .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt)), + }; + } + + async archiveThread(threadId: string): Promise<{ archived: boolean }> { + const archive = this.options.store.archive; + if (!archive) throw new ProtocolInvariantError('This store cannot archive threads'); + await this.requireThread(threadId); // 404 rather than silently succeeding + await archive.call(this.options.store, threadId); + return { archived: true }; + } + + /** + * Copy a thread into a new one, leaving the original untouched. + * + * An in-progress turn is copied as `interrupted`: the fork is a new thread + * that nothing is executing, and carrying `in_progress` across would make it + * permanently refuse to start a turn. + */ + async forkThread(threadId: string, traceId?: string): Promise { + const source = await this.requireThread(threadId); + const now = this.now(); + const fork: ThreadSnapshot = { + id: this.newId('thread'), + cwd: source.cwd, + createdAt: now, + updatedAt: now, + turns: source.turns.map((turn) => ({ + ...clone(turn), + status: turn.status === 'in_progress' ? ('interrupted' as const) : turn.status, + })), + }; + await this.options.store.save(fork); + this.emit({ type: 'thread.started', traceId, thread: clone(fork) }); + return clone(fork); + } + async startTurn( threadId: string, input: Record, diff --git a/packages/protocol/src/types.ts b/packages/protocol/src/types.ts index 65a7aba..a3565eb 100644 --- a/packages/protocol/src/types.ts +++ b/packages/protocol/src/types.ts @@ -157,6 +157,12 @@ export interface InitializeResult { reviewActions: boolean; /** Server streams `reasoning.delta` for models that emit reasoning. */ reasoningDeltas: boolean; + /** + * `thread/list`, `thread/fork` and `thread/archive` are served. Without + * these a client has to read the session directory itself, which is how the + * desktop ended up with two ways to see the same threads. + */ + threadManagement: boolean; }; } @@ -241,6 +247,22 @@ export type ReviewActionRequest = export type ReviewActionPayload = ReviewActionRequest & { actionId: string }; +/** A thread as it appears in a list — enough to render a picker, no turns. */ +export interface ThreadListEntry { + id: string; + cwd: string; + createdAt: string; + updatedAt: string; + /** First user message, truncated — absent for a thread with no turns yet. */ + title?: string; + turnCount: number; + archived?: boolean; +} + +export interface ThreadListResult { + threads: ThreadListEntry[]; +} + export type ProtocolMethod = | 'initialize' | 'config/diagnostics' @@ -251,6 +273,9 @@ export type ProtocolMethod = | 'thread/start' | 'thread/read' | 'thread/resume' + | 'thread/list' + | 'thread/fork' + | 'thread/archive' | 'turn/start' | 'turn/interrupt' | 'approval/respond'