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'