From 5058953a303da967bb81a5d744a754f8823e3ae1 Mon Sep 17 00:00:00 2001 From: DoSun Date: Wed, 6 May 2026 14:32:04 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=E5=9F=BA=E4=BA=8E=20?= =?UTF-8?q?Hono=20=E7=9A=84=20HTTP=20API=20=E6=9C=8D=E5=8A=A1=E5=99=A8?= =?UTF-8?q?=EF=BC=88serve=20=E6=A8=A1=E5=BC=8F=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/server/errors.ts | 59 +++ src/server/eventBus.ts | 100 +++++ src/server/lockfile.ts | 13 - src/server/routes/event.ts | 19 + src/server/routes/find.ts | 36 ++ src/server/routes/health.ts | 14 + src/server/routes/info.ts | 80 ++++ src/server/routes/message.ts | 62 ++++ src/server/routes/permission.ts | 29 ++ src/server/routes/provider.ts | 65 ++++ src/server/routes/question.ts | 32 ++ src/server/routes/session.ts | 267 ++++++++++++++ src/server/server.ts | 65 +++- src/server/serverBanner.ts | 21 +- src/server/serverLog.ts | 17 +- src/server/sessionHandle.ts | 629 ++++++++++++++++++++++++++++++++ src/server/sessionManager.ts | 277 +++++++++++++- src/server/transcriptReader.ts | 305 ++++++++++++++++ src/server/types.ts | 2 +- 19 files changed, 2066 insertions(+), 26 deletions(-) create mode 100644 src/server/errors.ts create mode 100644 src/server/eventBus.ts delete mode 100644 src/server/lockfile.ts create mode 100644 src/server/routes/event.ts create mode 100644 src/server/routes/find.ts create mode 100644 src/server/routes/health.ts create mode 100644 src/server/routes/info.ts create mode 100644 src/server/routes/message.ts create mode 100644 src/server/routes/permission.ts create mode 100644 src/server/routes/provider.ts create mode 100644 src/server/routes/question.ts create mode 100644 src/server/routes/session.ts create mode 100644 src/server/sessionHandle.ts create mode 100644 src/server/transcriptReader.ts diff --git a/src/server/errors.ts b/src/server/errors.ts new file mode 100644 index 000000000..af56c4584 --- /dev/null +++ b/src/server/errors.ts @@ -0,0 +1,59 @@ +import type { Context } from 'hono' +import { HTTPException } from 'hono/http-exception' + +export class ServeError extends Error { + constructor( + public status: number, + public code: string, + message: string, + ) { + super(message) + this.name = 'ServeError' + } + + json() { + return { error: this.code, message: this.message } + } +} + +export function badRequest(message: string): ServeError { + return new ServeError(400, 'BAD_REQUEST', message) +} + +export function unauthorized(message = 'Unauthorized'): ServeError { + return new ServeError(401, 'UNAUTHORIZED', message) +} + +export function notFound(message: string): ServeError { + return new ServeError(404, 'NOT_FOUND', message) +} + +export function conflict(message: string): ServeError { + return new ServeError(409, 'CONFLICT', message) +} + +export function tooManySessions(message: string): ServeError { + return new ServeError(429, 'TOO_MANY_SESSIONS', message) +} + +export function sessionError(message: string): ServeError { + return new ServeError(503, 'SESSION_ERROR', message) +} + +export function errorHandler(err: Error, c: Context) { + if (err instanceof ServeError) { + return c.json(err.json(), { status: err.status }) + } + if (err instanceof HTTPException) { + return c.json( + { error: 'HTTP_ERROR', message: err.message }, + { status: err.status }, + ) + } + const message = + err instanceof Error && err.stack ? err.stack : err.toString() + return c.json( + { error: 'INTERNAL', message }, + { status: 500 }, + ) +} diff --git a/src/server/eventBus.ts b/src/server/eventBus.ts new file mode 100644 index 000000000..3ad404f67 --- /dev/null +++ b/src/server/eventBus.ts @@ -0,0 +1,100 @@ +export type SSEWriter = { + writeSSE: (opts: { event: string; data: string }) => Promise +} + +type SSEClient = { + id: string + writer: SSEWriter | null + rawWrite: ((event: string, data: unknown) => void) | null + sessionIdFilter?: string +} + +type BusEvent = { + event: string + data: unknown +} + +export class EventBus { + private clients = new Map() + private buffer: BusEvent[] = [] + private bufferSize = 100 + private heartbeatInterval: ReturnType | null = null + + addClient(writer: SSEWriter, sessionIdFilter?: string): string { + const id = crypto.randomUUID() + const client: SSEClient = { id, writer, rawWrite: null, sessionIdFilter } + this.clients.set(id, client) + + void writer.writeSSE({ + event: 'connected', + data: JSON.stringify({ type: 'server.connected' }), + }) + + for (const buffered of this.buffer) { + void writer.writeSSE({ + event: buffered.event, + data: JSON.stringify(buffered.data), + }) + } + + return id + } + + removeClient(id: string): void { + this.clients.delete(id) + } + + publish(event: string, data: unknown): void { + const entry: BusEvent = { event, data } + this.buffer.push(entry) + if (this.buffer.length > this.bufferSize) { + this.buffer.shift() + } + const payload = JSON.stringify(data) + for (const client of this.clients.values()) { + if (client.writer) { + if (client.sessionIdFilter) { + const dataObj = data as Record | undefined + if (dataObj?.session_id && dataObj.session_id !== client.sessionIdFilter) { + continue + } + } + void client.writer.writeSSE({ event, data: payload }) + } + } + } + + publishSessionEvent( + sessionId: string, + event: string, + data: Record, + ): void { + this.publish(`session.${event}`, { session_id: sessionId, ...data }) + } + + startHeartbeat(intervalMs = 10000): void { + this.heartbeatInterval = setInterval(() => { + this.publish('heartbeat', { + type: 'server.heartbeat', + ts: Date.now(), + }) + }, intervalMs) + } + + stopHeartbeat(): void { + if (this.heartbeatInterval) { + clearInterval(this.heartbeatInterval) + this.heartbeatInterval = null + } + } + + clientCount(): number { + return this.clients.size + } + + destroy(): void { + this.stopHeartbeat() + this.clients.clear() + this.buffer.length = 0 + } +} diff --git a/src/server/lockfile.ts b/src/server/lockfile.ts deleted file mode 100644 index 7efc3e804..000000000 --- a/src/server/lockfile.ts +++ /dev/null @@ -1,13 +0,0 @@ -// Auto-generated stub — replace with real implementation - -export interface ServerLockInfo { - pid: number - port: number - host: string - httpUrl: string - startedAt: number -} - -export const writeServerLock: (info: ServerLockInfo) => Promise = (async () => {}); -export const removeServerLock: () => Promise = (async () => {}); -export const probeRunningServer: () => Promise = (async () => null); diff --git a/src/server/routes/event.ts b/src/server/routes/event.ts new file mode 100644 index 000000000..8c8327c0c --- /dev/null +++ b/src/server/routes/event.ts @@ -0,0 +1,19 @@ +import { Hono } from 'hono' +import { streamSSE } from 'hono/streaming' +import type { EventBus } from '../eventBus.js' + +export function createEventRoutes(eventBus: EventBus): Hono { + return new Hono().get('/event', async c => { + const sessionIdFilter = c.req.query('session_id') ?? undefined + return streamSSE(c, async stream => { + const clientId = eventBus.addClient(stream, sessionIdFilter) + stream.onAbort(() => { + eventBus.removeClient(clientId) + }) + + while (true) { + await stream.sleep(30000) + } + }) + }) +} diff --git a/src/server/routes/find.ts b/src/server/routes/find.ts new file mode 100644 index 000000000..39f758613 --- /dev/null +++ b/src/server/routes/find.ts @@ -0,0 +1,36 @@ +import { Hono } from 'hono' +import { stat } from 'fs/promises' +import { join } from 'path' +import { badRequest } from '../errors.js' + +export function createFindRoutes(): Hono { + return new Hono().get('/find/file', async c => { + const query = c.req.query('query') + if (!query) throw badRequest('query is required') + + const limit = parseInt(c.req.query('limit') ?? '50', 10) + const dirs = c.req.query('dirs') === 'true' + const cwd = process.cwd() + + try { + const glob = new Bun.Glob(`**/*${query}*`) + const results: string[] = [] + for await (const path of glob.scan({ cwd, dot: false, absolute: false })) { + if (dirs) { + try { + const s = await stat(join(cwd, path)) + if (!s.isDirectory()) continue + } catch { + continue + } + } + results.push(path) + if (results.length >= limit) break + } + + return c.json(results) + } catch { + return c.json([]) + } + }) +} diff --git a/src/server/routes/health.ts b/src/server/routes/health.ts new file mode 100644 index 000000000..bbb2348ab --- /dev/null +++ b/src/server/routes/health.ts @@ -0,0 +1,14 @@ +import { Hono } from 'hono' +import type { SessionManager } from '../sessionManager.js' + +export function createHealthRoutes(sessionManager: SessionManager): Hono { + return new Hono().get('/health', c => { + const uptime = process.uptime() * 1000 + return c.json({ + status: 'ok', + version: MACRO.VERSION, + uptime_ms: Math.round(uptime), + active_sessions: sessionManager.getActiveCount(), + }) + }) +} diff --git a/src/server/routes/info.ts b/src/server/routes/info.ts new file mode 100644 index 000000000..49a179efb --- /dev/null +++ b/src/server/routes/info.ts @@ -0,0 +1,80 @@ +import { Hono } from 'hono' +import { execSync } from 'child_process' +import { homedir } from 'os' +import { getClaudeConfigHomeDir } from '../../utils/envUtils.js' +import { getCommands } from '../../commands.js' +import { getCommandName } from '../../types/command.js' +import type { SessionManager } from '../sessionManager.js' +import type { InitData } from '../sessionHandle.js' + +export function createInfoRoutes(sessionManager: SessionManager): Hono { + return new Hono() + .get('/path', c => { + return c.json({ + home: homedir(), + state: getClaudeConfigHomeDir(), + config: getClaudeConfigHomeDir(), + directory: process.cwd(), + }) + }) + .get('/vcs', c => { + let branch = '' + try { + branch = execSync('git rev-parse --abbrev-ref HEAD', { + encoding: 'utf-8', + timeout: 5000, + }).trim() + } catch {} + return c.json({ branch }) + }) + .get('/command', async c => { + try { + const commands = await getCommands(process.cwd()) + return c.json( + commands + .filter(cmd => !cmd.isHidden) + .map(cmd => ({ + name: getCommandName(cmd), + description: cmd.description, + argumentHint: cmd.argumentHint, + })), + ) + } catch { + return c.json([]) + } + }) + .get('/agent', c => { + const initData = getFirstInitData(sessionManager) + if (initData?.agents && initData.agents.length > 0) { + return c.json( + initData.agents.map(a => ({ + id: a.name, + name: a.name, + description: a.description, + model: a.model, + })), + ) + } + return c.json([]) + }) + .get('/mcp', async c => { + for (const handle of sessionManager.getAllSessions()) { + if (handle.status === 'running') { + try { + const servers = await handle.getMcpStatus() + return c.json({ servers }) + } catch { + continue + } + } + } + return c.json({ servers: [] }) + }) +} + +function getFirstInitData(sessionManager: SessionManager): InitData | null { + for (const handle of sessionManager.getAllSessions()) { + if (handle.initData) return handle.initData + } + return null +} diff --git a/src/server/routes/message.ts b/src/server/routes/message.ts new file mode 100644 index 000000000..6c0c072d9 --- /dev/null +++ b/src/server/routes/message.ts @@ -0,0 +1,62 @@ +import { Hono } from 'hono' +import type { SessionManager } from '../sessionManager.js' +import { notFound } from '../errors.js' +import { + readSessionMessages, + readSessionTodos, + readSessionDiff, +} from '../transcriptReader.js' + +export function createMessageRoutes(sessionManager: SessionManager): Hono { + return new Hono() + .get('/session/:sessionID/message', async c => { + const id = c.req.param('sessionID') + const handle = sessionManager.getSession(id) + + const url = new URL(c.req.url) + const limit = parseInt(url.searchParams.get('limit') ?? '50', 10) + const before = url.searchParams.get('before') ?? undefined + const includeSystem = url.searchParams.get('include_system') === 'true' + + const { messages, nextCursor } = await readSessionMessages({ + sessionId: id, + cwd: handle?.cwd, + limit, + before, + includeSystem, + }) + + if (nextCursor) { + c.header('Link', `<${url.pathname}?limit=${limit}&before=${nextCursor}>; rel="prev"`) + c.header('X-Next-Cursor', nextCursor) + } + + return c.json({ messages }) + }) + .get('/session/:sessionID/todo', async c => { + const id = c.req.param('sessionID') + const handle = sessionManager.getSession(id) + + const todos = await readSessionTodos({ + sessionId: id, + cwd: handle?.cwd, + }) + + return c.json({ todos }) + }) + .get('/session/:sessionID/diff', async c => { + const id = c.req.param('sessionID') + const handle = sessionManager.getSession(id) + + const url = new URL(c.req.url) + const messageID = url.searchParams.get('messageID') ?? undefined + + const { diffs } = await readSessionDiff({ + sessionId: id, + cwd: handle?.cwd, + messageID, + }) + + return c.json({ diffs }) + }) +} diff --git a/src/server/routes/permission.ts b/src/server/routes/permission.ts new file mode 100644 index 000000000..f4911e680 --- /dev/null +++ b/src/server/routes/permission.ts @@ -0,0 +1,29 @@ +import { Hono } from 'hono' +import type { SessionManager } from '../sessionManager.js' +import { notFound } from '../errors.js' + +export function createPermissionRoutes(sessionManager: SessionManager): Hono { + return new Hono() + .get('/permission', c => { + const permissions = sessionManager.getAllPendingPermissions() + return c.json({ permissions }) + }) + .post('/permission/:requestID/reply', async c => { + const requestId = c.req.param('requestID') + const found = sessionManager.findPermissionAcrossSessions(requestId) + if (!found) throw notFound('permission request not found') + + const body = await c.req.json<{ + behavior: 'allow' | 'deny' + updated_input?: Record + message?: string + }>() + + found.handle.replyPermission(requestId, body.behavior, { + updatedInput: body.updated_input, + message: body.message, + }) + + return c.json({ resolved: true }) + }) +} diff --git a/src/server/routes/provider.ts b/src/server/routes/provider.ts new file mode 100644 index 000000000..838ca925a --- /dev/null +++ b/src/server/routes/provider.ts @@ -0,0 +1,65 @@ +import { Hono } from 'hono' +import type { SessionManager } from '../sessionManager.js' +import type { InitData } from '../sessionHandle.js' + +export function createProviderRoutes(sessionManager: SessionManager): Hono { + return new Hono() + .get('/provider', c => { + const initData = getFirstInitData(sessionManager) + const models = initData?.models ?? {} + const modelValues = Object.values(models as Record) + const defaultModel = modelValues[0] ?? '' + + return c.json({ + connected: initData?.account?.apiProvider + ? [initData.account.apiProvider] + : ['anthropic'], + default_model: defaultModel, + providers: [ + { + id: initData?.account?.apiProvider ?? 'anthropic', + name: + initData?.account?.apiProvider === 'firstParty' + ? 'Anthropic' + : initData?.account?.apiProvider ?? 'Anthropic', + connected: true, + models: Object.entries(models as Record).map( + ([key, id]) => ({ + id, + name: key, + }), + ), + }, + ], + }) + }) + .get('/provider/capabilities', c => { + const initData = getFirstInitData(sessionManager) + const models = initData?.models ?? {} + + return c.json({ + connected: [ + { + provider_id: initData?.account?.apiProvider ?? 'anthropic', + provider_name: + initData?.account?.apiProvider === 'firstParty' + ? 'Anthropic' + : initData?.account?.apiProvider ?? 'Anthropic', + models: Object.entries(models as Record).map( + ([key, id]) => ({ + model_id: id, + model_name: key, + }), + ), + }, + ], + }) + }) +} + +function getFirstInitData(sessionManager: SessionManager): InitData | null { + for (const handle of sessionManager.getAllSessions()) { + if (handle.initData) return handle.initData + } + return null +} diff --git a/src/server/routes/question.ts b/src/server/routes/question.ts new file mode 100644 index 000000000..75af140f9 --- /dev/null +++ b/src/server/routes/question.ts @@ -0,0 +1,32 @@ +import { Hono } from 'hono' +import type { SessionManager } from '../sessionManager.js' +import { notFound } from '../errors.js' + +export function createQuestionRoutes(sessionManager: SessionManager): Hono { + return new Hono() + .get('/question', c => { + const questions = sessionManager.getAllPendingQuestions() + return c.json({ questions }) + }) + .post('/question/:requestID/reply', async c => { + const requestId = c.req.param('requestID') + const found = sessionManager.findQuestionAcrossSessions(requestId) + if (!found) throw notFound('question request not found') + + const body = await c.req.json<{ + action: 'accept' + content?: Record + }>() + + found.handle.replyQuestion(requestId, body.action, body.content) + return c.json({ resolved: true }) + }) + .post('/question/:requestID/reject', async c => { + const requestId = c.req.param('requestID') + const found = sessionManager.findQuestionAcrossSessions(requestId) + if (!found) throw notFound('question request not found') + + found.handle.replyQuestion(requestId, 'decline') + return c.json({ resolved: true }) + }) +} diff --git a/src/server/routes/session.ts b/src/server/routes/session.ts new file mode 100644 index 000000000..62a5134e3 --- /dev/null +++ b/src/server/routes/session.ts @@ -0,0 +1,267 @@ +import { Hono } from 'hono' +import { streamSSE } from 'hono/streaming' +import type { SessionManager } from '../sessionManager.js' +import { + badRequest, + notFound, + tooManySessions, + sessionError, + conflict, +} from '../errors.js' + +function ssePrompt( + handle: import('../sessionHandle.js').SessionHandle, + id: string, + content: string, + c: import('hono').Context, +) { + return streamSSE(c, async stream => { + const startTime = Date.now() + let resultWritten = false + let pendingWrite: Promise = Promise.resolve() + + const unsub = handle.onMessage(msg => { + if (resultWritten) return + if (msg.type === 'result') { + resultWritten = true + const info = handle.getInfo() + pendingWrite = pendingWrite.then(() => + stream.writeSSE({ + event: 'result', + data: JSON.stringify({ + ...msg, + session_id: id, + cost_usd: info.cost_usd, + duration_ms: Date.now() - startTime, + }), + }), + ) + unsub() + } else if ( + msg.type === 'assistant' || + msg.type === 'tool_progress' + ) { + pendingWrite = pendingWrite.then(() => + stream.writeSSE({ + event: 'message', + data: JSON.stringify({ ...msg, session_id: id }), + }), + ) + } else if (msg.type === 'control_request') { + pendingWrite = pendingWrite.then(() => + stream.writeSSE({ + event: 'control_request', + data: JSON.stringify({ ...msg, session_id: id }), + }), + ) + } else if (msg.type === 'system') { + pendingWrite = pendingWrite.then(() => + stream.writeSSE({ + event: 'system', + data: JSON.stringify({ ...msg, session_id: id }), + }), + ) + } + }) + + stream.onAbort(() => { + unsub() + }) + + try { + await handle.prompt(content) + } catch { + if (!resultWritten) { + pendingWrite = pendingWrite.then(() => + stream.writeSSE({ + event: 'result', + data: JSON.stringify({ + type: 'result', + subtype: 'error_during_execution', + session_id: id, + duration_ms: Date.now() - startTime, + }), + }), + ) + } + } + + await pendingWrite + + if (!resultWritten) { + await stream.writeSSE({ + event: 'result', + data: JSON.stringify({ + type: 'result', + subtype: 'success', + session_id: id, + duration_ms: Date.now() - startTime, + }), + }) + } + + await stream.sleep(50) + }) +} + +export function createSessionRoutes( + sessionManager: SessionManager, +): Hono { + return new Hono() + .post('/session', async c => { + const body = await c.req.json<{ + cwd?: string + permission_mode?: string + model?: string + system_prompt?: string + resume_session_id?: string + }>() + + try { + const handle = await sessionManager.createSession({ + cwd: body.cwd, + model: body.model, + permissionMode: body.permission_mode, + systemPrompt: body.system_prompt, + resumeSessionId: body.resume_session_id, + execPath: process.execPath, + scriptArgs: process.argv[1] ? [process.argv[1]] : [], + }) + + const initData = handle.initData + + return c.json( + { + session_id: handle.sessionId, + status: handle.status, + cwd: handle.cwd, + created_at: handle.getInfo().created_at, + commands: initData?.commands ?? [], + agents: initData?.agents ?? [], + models: initData?.models ?? {}, + account: initData?.account ?? {}, + }, + 201, + ) + } catch (err) { + const msg = + err instanceof Error ? err.message : 'Failed to create session' + if (msg.includes('Maximum concurrent')) { + throw tooManySessions(msg) + } + throw sessionError(msg) + } + }) + .get('/session', c => { + const url = new URL(c.req.url) + const limit = parseInt(url.searchParams.get('limit') ?? '50', 10) + const offset = parseInt(url.searchParams.get('offset') ?? '0', 10) + + const handles = sessionManager.getAllSessions() + const sessions = handles + .slice(offset, offset + limit) + .map(h => h.getInfo()) + + return c.json({ sessions }) + }) + .get('/session/status', c => { + return c.json({ sessions: sessionManager.getSessionStatuses() }) + }) + .get('/session/:sessionID', c => { + const id = c.req.param('sessionID') + const handle = sessionManager.getSession(id) + if (!handle) throw notFound('session not found') + const info = handle.getInfo() + return c.json({ + ...info, + message_count: handle.messageCount, + usage: handle.usage, + }) + }) + .patch('/session/:sessionID', async c => { + const id = c.req.param('sessionID') + const handle = sessionManager.getSession(id) + if (!handle) throw notFound('session not found') + + const body = await c.req.json<{ + title?: string + model?: string + permission_mode?: string + }>() + + if (body.title) handle.setTitle(body.title) + if (body.model) await handle.setModel(body.model) + if (body.permission_mode) await handle.setPermissionMode(body.permission_mode) + + return c.json({ + session_id: id, + title: handle.title ?? body.title, + model: handle.model, + permission_mode: handle.permissionMode, + }) + }) + .delete('/session/:sessionID', async c => { + const id = c.req.param('sessionID') + const deleted = await sessionManager.deleteSession(id) + if (!deleted) throw notFound('session not found') + return c.json({ deleted: true }) + }) + .post('/session/:sessionID/prompt', async c => { + const id = c.req.param('sessionID') + const handle = sessionManager.getSession(id) + if (!handle) throw notFound('session not found') + + const body = await c.req.json<{ + content: string + files?: string[] + images?: unknown[] + }>() + + if (!body.content) throw badRequest('content is required') + if (handle.prompting) throw conflict('session is already processing a prompt') + return ssePrompt(handle, id, body.content, c) + }) + .post('/session/:sessionID/prompt_async', async c => { + const id = c.req.param('sessionID') + const handle = sessionManager.getSession(id) + if (!handle) throw notFound('session not found') + + const body = await c.req.json<{ + content: string + files?: string[] + images?: unknown[] + }>() + + if (!body.content) throw badRequest('content is required') + if (handle.prompting) throw conflict('session is already processing a prompt') + + handle.prompt(body.content).catch(() => {}) + + return new Response(null, { status: 204 }) + }) + .post('/session/:sessionID/abort', async c => { + const id = c.req.param('sessionID') + const handle = sessionManager.getSession(id) + if (!handle) throw notFound('session not found') + await handle.abort() + return c.json({ aborted: true }) + }) + .post('/session/:sessionID/shell', async c => { + const id = c.req.param('sessionID') + const handle = sessionManager.getSession(id) + if (!handle) throw notFound('session not found') + + const body = await c.req.json<{ command: string }>() + if (!body.command) throw badRequest('command is required') + return ssePrompt(handle, id, body.command, c) + }) + .post('/session/:sessionID/command', async c => { + const id = c.req.param('sessionID') + const handle = sessionManager.getSession(id) + if (!handle) throw notFound('session not found') + + const body = await c.req.json<{ command: string }>() + if (!body.command) throw badRequest('command is required') + return ssePrompt(handle, id, body.command, c) + }) +} diff --git a/src/server/server.ts b/src/server/server.ts index 94a377148..926774e16 100644 --- a/src/server/server.ts +++ b/src/server/server.ts @@ -1,3 +1,62 @@ -// Auto-generated stub — replace with real implementation -export {}; -export const startServer: (...args: unknown[]) => { port?: number; stop: (closeActiveConnections: boolean) => void } = () => ({ stop() {} }); +import { Hono } from 'hono' +import { cors } from 'hono/cors' +import type { Server as BunServer } from 'bun' +import { EventBus } from './eventBus.js' +import { SessionManager } from './sessionManager.js' +import { errorHandler } from './errors.js' +import { createHealthRoutes } from './routes/health.js' +import { createInfoRoutes } from './routes/info.js' +import { createSessionRoutes } from './routes/session.js' +import { createEventRoutes } from './routes/event.js' +import { createPermissionRoutes } from './routes/permission.js' +import { createQuestionRoutes } from './routes/question.js' +import { createMessageRoutes } from './routes/message.js' +import { createProviderRoutes } from './routes/provider.js' +import { createFindRoutes } from './routes/find.js' +import type { ServerConfig } from './types.js' + +export function startServer( + config: ServerConfig, + sessionManager: SessionManager, +): BunServer & { port?: number } { + const eventBus = new EventBus() + eventBus.startHeartbeat() + + const app = new Hono() + + app.onError(errorHandler) + + // Auth is intentionally disabled for now so multiple local serve instances + // can be started and consumed without token plumbing. + + app.use( + '*', + cors({ + maxAge: 86400, + origin: (input) => { + if (!input) return + if (input.startsWith('http://localhost:')) return input + if (input.startsWith('http://127.0.0.1:')) return input + return + }, + }), + ) + + app.route('/', createHealthRoutes(sessionManager)) + app.route('/', createInfoRoutes(sessionManager)) + app.route('/', createSessionRoutes(sessionManager)) + app.route('/', createEventRoutes(eventBus)) + app.route('/', createPermissionRoutes(sessionManager)) + app.route('/', createQuestionRoutes(sessionManager)) + app.route('/', createMessageRoutes(sessionManager)) + app.route('/', createProviderRoutes(sessionManager)) + app.route('/', createFindRoutes()) + + const server = Bun.serve({ + hostname: config.host, + port: config.port, + fetch: app.fetch, + }) + + return server +} diff --git a/src/server/serverBanner.ts b/src/server/serverBanner.ts index b91b6b484..4902e41f7 100644 --- a/src/server/serverBanner.ts +++ b/src/server/serverBanner.ts @@ -1,3 +1,18 @@ -// Auto-generated stub — replace with real implementation -export {}; -export const printBanner: (...args: unknown[]) => void = () => {}; +import type { ServerConfig } from './types.js' + +export function printBanner( + config: ServerConfig, + authToken: string | undefined, + actualPort: number, +): void { + const url = config.unix + ? `unix:${config.unix}` + : `http://${config.host}:${actualPort}` + + process.stderr.write(`\ncsc server listening on ${url}\n`) + + // Auth token display intentionally disabled while serve auth is disabled. + + process.stderr.write(`Max sessions: ${config.maxSessions ?? 32}\n`) + process.stderr.write(`Workspace: ${config.workspace || process.cwd()}\n\n`) +} diff --git a/src/server/serverLog.ts b/src/server/serverLog.ts index e89c00070..0d38cdc76 100644 --- a/src/server/serverLog.ts +++ b/src/server/serverLog.ts @@ -1,3 +1,14 @@ -// Auto-generated stub — replace with real implementation -export {}; -export const createServerLogger: () => Record = () => ({}); +export function createServerLogger() { + return { + info(message: string, meta?: Record) { + process.stderr.write( + `[serve:info] ${message}${meta ? ' ' + JSON.stringify(meta) : ''}\n`, + ) + }, + error(message: string, meta?: Record) { + process.stderr.write( + `[serve:error] ${message}${meta ? ' ' + JSON.stringify(meta) : ''}\n`, + ) + }, + } +} diff --git a/src/server/sessionHandle.ts b/src/server/sessionHandle.ts new file mode 100644 index 000000000..23ac38a3e --- /dev/null +++ b/src/server/sessionHandle.ts @@ -0,0 +1,629 @@ +import { type ChildProcess, spawn } from 'child_process' +import { createInterface } from 'readline' +import { jsonParse, jsonStringify } from '../utils/slowOperations.js' +import { logError } from '../utils/log.js' +import type { EventBus } from './eventBus.js' +import type { SessionState } from './types.js' + +type StdoutMessage = { + type: string + [key: string]: unknown +} + +type MessageListener = (msg: StdoutMessage) => void + +export type InitData = { + commands?: Array<{ name: string; description: string; argumentHint?: string }> + agents?: Array<{ name: string; description?: string; model?: string }> + models?: unknown + account?: { + email?: string + organization?: string + subscriptionType?: string + tokenSource?: string + apiKeySource?: string + apiProvider?: string + } + output_style?: string + available_output_styles?: string[] + pid?: number +} + +export type PendingPermission = { + requestId: string + sessionId: string + toolName: string + toolUseId: string + input: Record + title: string + description: string +} + +export type PendingQuestion = { + requestId: string + sessionId: string + mcpServerName: string + message: string + mode: string + requestedSchema: Record +} + +export type SessionHandleOptions = { + sessionId: string + cwd: string + model?: string + permissionMode?: string + systemPrompt?: string + resumeSessionId?: string + eventBus: EventBus + execPath: string + scriptArgs: string[] + verbose?: boolean +} + +export class SessionHandle { + readonly sessionId: string + readonly cwd: string + private child: ChildProcess | null = null + private _status: SessionState = 'starting' + private _model?: string + private _permissionMode?: string + private _title?: string + private _costUsd = 0 + private _inputTokens = 0 + private _outputTokens = 0 + private createdAt = Date.now() + private lastActiveAt = Date.now() + private eventBus: EventBus + private pendingPermissions = new Map() + private pendingQuestions = new Map() + private promptResolve: + | ((value: { done: boolean; error?: string }) => void) + | null = null + private promptReject: ((reason: unknown) => void) | null = null + private verbose: boolean + private lastStderr: string[] = [] + private listeners = new Set() + private _initData: InitData | null = null + private initRequestId: string | null = null + private initResolve: ((data: InitData) => void) | null = null + private _prompting = false + + get status(): SessionState { + return this._status + } + get model(): string | undefined { + return this._model + } + get permissionMode(): string | undefined { + return this._permissionMode + } + get title(): string | undefined { + return this._title + } + get costUsd(): number { + return this._costUsd + } + get inputTokens(): number { + return this._inputTokens + } + get messageCount(): number { + return 0 + } + get usage() { + return { + input_tokens: this._inputTokens, + output_tokens: this._outputTokens, + } + } + get initData(): InitData | null { + return this._initData + } + get prompting(): boolean { + return this._prompting + } + + constructor(private opts: SessionHandleOptions) { + this.sessionId = opts.sessionId + this.cwd = opts.cwd + this.eventBus = opts.eventBus + this._model = opts.model + this._permissionMode = opts.permissionMode + this.verbose = opts.verbose ?? false + } + + onMessage(listener: MessageListener): () => void { + this.listeners.add(listener) + return () => { this.listeners.delete(listener) } + } + + async start(): Promise { + const args = [ + ...this.opts.scriptArgs, + '--print', + '--input-format', + 'stream-json', + '--output-format', + 'stream-json', + '--session-id', + this.sessionId, + ...(this.opts.model ? ['--model', this.opts.model] : []), + ...(this.opts.permissionMode + ? ['--permission-mode', this.opts.permissionMode] + : []), + ...(this.opts.resumeSessionId + ? ['--resume', this.opts.resumeSessionId] + : []), + ...(this.opts.verbose ? ['--verbose'] : []), + ] + + const env: NodeJS.ProcessEnv = { + ...process.env, + CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC: '1', + } + + if (this.opts.systemPrompt) { + env.CLAUDE_CODE_SYSTEM_PROMPT = this.opts.systemPrompt + } + + this.child = spawn(this.opts.execPath, args, { + cwd: this.opts.cwd, + stdio: ['pipe', 'pipe', 'pipe'], + env, + windowsHide: true, + }) + + this.setupStdout() + this.setupStderr() + + this.child.on('close', (code, signal) => { + if (this._status !== 'stopped') { + this._status = 'stopped' + this.eventBus.publishSessionEvent(this.sessionId, 'deleted', { + status: 'stopped', + exit_code: code, + signal: signal ?? null, + }) + } + if (this.initResolve) { + this.initResolve = null + } + if (this.promptReject) { + this.promptReject(new Error(`Process exited with code ${code}`)) + } + }) + + this.child.on('error', err => { + logError(err) + this._status = 'stopped' + if (this.initResolve) { + this.initResolve = null + } + if (this.promptReject) { + this.promptReject(err) + } + }) + + this.initRequestId = crypto.randomUUID() + const initPromise = new Promise((resolve, reject) => { + this.initResolve = resolve + const timeout = setTimeout(() => { + this.initResolve = null + reject(new Error('Init handshake timed out (30s)')) + }, 30000) + const originalResolve = resolve + this.initResolve = (data: InitData) => { + clearTimeout(timeout) + originalResolve(data) + } + }) + + this.sendInitialize() + + try { + this._initData = await initPromise + } catch (err) { + this._status = 'stopped' + throw err + } + + this._status = 'running' + this.lastActiveAt = Date.now() + this.eventBus.publishSessionEvent(this.sessionId, 'ready', { + status: 'running', + model: this._model, + }) + } + + private sendInitialize(): void { + const msg = jsonStringify({ + type: 'control_request', + request_id: this.initRequestId, + request: { + subtype: 'initialize', + }, + }) + this.writeStdin(msg) + } + + private setupStdout(): void { + if (!this.child?.stdout) return + const rl = createInterface({ input: this.child.stdout }) + rl.on('line', line => { + if (!line.trim()) return + let msg: StdoutMessage + try { + msg = jsonParse(line) as StdoutMessage + } catch { + return + } + this.routeMessage(msg) + }) + } + + private setupStderr(): void { + if (!this.child?.stderr) return + const rl = createInterface({ input: this.child.stderr }) + rl.on('line', line => { + if (this.verbose) { + process.stderr.write(`[serve:${this.sessionId}] ${line}\n`) + } + if (this.lastStderr.length >= 20) { + this.lastStderr.shift() + } + this.lastStderr.push(line) + }) + } + + private emitMessage(msg: StdoutMessage): void { + for (const listener of this.listeners) { + try { + listener(msg) + } catch {} + } + } + + private routeMessage(msg: StdoutMessage): void { + if ( + this._status === 'starting' && + msg.type === 'control_response' && + this.initResolve + ) { + const response = msg.response as Record | undefined + if (response?.subtype === 'success' && response?.request_id === this.initRequestId) { + const initData = (response.response ?? {}) as InitData + this._initData = initData + if (initData.models) { + const models = initData.models as Record + const vals = Object.values(models) + if (vals.length > 0 && !this._model) this._model = vals[0] + } + this.initResolve(initData) + this.initResolve = null + return + } + } + + if (msg.type === 'control_response') { + const response = msg.response as Record | undefined + const requestId = response?.request_id as string | undefined + if (requestId && this.pendingControlResponses.has(requestId)) { + const pending = this.pendingControlResponses.get(requestId)! + this.pendingControlResponses.delete(requestId) + clearTimeout(pending.timeout) + if (response?.subtype === 'error') { + pending.reject(new Error((response.error as string) ?? 'Unknown error')) + } else { + pending.resolve((response?.response ?? {}) as Record) + } + return + } + } + + this.emitMessage(msg) + + switch (msg.type) { + case 'system': { + this.eventBus.publishSessionEvent(this.sessionId, 'message', msg) + break + } + case 'assistant': { + this.lastActiveAt = Date.now() + this._status = 'running' + this.eventBus.publishSessionEvent(this.sessionId, 'message', msg) + break + } + case 'result': { + this.lastActiveAt = Date.now() + this._status = 'running' + const cost = msg.cost_usd as number | undefined + const usage = msg.usage as + | { input_tokens?: number; output_tokens?: number } + | undefined + if (cost) this._costUsd += cost + if (usage?.input_tokens) this._inputTokens += usage.input_tokens + if (usage?.output_tokens) this._outputTokens += usage.output_tokens + this.eventBus.publishSessionEvent(this.sessionId, 'result', msg) + if (this.promptResolve) { + this.promptResolve({ done: true }) + this.promptResolve = null + this.promptReject = null + } + break + } + case 'control_request': { + const request = msg.request as Record | undefined + const requestId = msg.request_id as string + if (request?.subtype === 'can_use_tool') { + const perm: PendingPermission = { + requestId, + sessionId: this.sessionId, + toolName: (request.tool_name as string) ?? 'Unknown', + toolUseId: (request.tool_use_id as string) ?? '', + input: (request.input as Record) ?? {}, + title: `${request.tool_name}: ${JSON.stringify(request.input).slice(0, 80)}`, + description: `Execute ${request.tool_name}`, + } + this.pendingPermissions.set(requestId, perm) + this.eventBus.publishSessionEvent( + this.sessionId, + 'control_request', + { request_id: requestId, request }, + ) + } else if (request?.subtype === 'elicitation') { + const question: PendingQuestion = { + requestId, + sessionId: this.sessionId, + mcpServerName: (request.mcp_server_name as string) ?? '', + message: (request.message as string) ?? '', + mode: (request.mode as string) ?? 'form', + requestedSchema: + (request.requested_schema as Record) ?? {}, + } + this.pendingQuestions.set(requestId, question) + this.eventBus.publishSessionEvent( + this.sessionId, + 'control_request', + { request_id: requestId, request }, + ) + } + break + } + default: { + this.eventBus.publishSessionEvent(this.sessionId, 'message', msg) + break + } + } + } + + private _outputTokens = 0 + private pendingControlResponses = new Map< + string, + { + resolve: (response: Record) => void + reject: (error: Error) => void + timeout: ReturnType + } + >() + + setTitle(title: string): void { + this._title = title + } + + async sendControlRequest( + request: Record, + timeoutMs = 10000, + ): Promise> { + const requestId = crypto.randomUUID() + this.writeStdin( + jsonStringify({ + type: 'control_request', + request_id: requestId, + request, + }), + ) + return this.waitForControlResponse(requestId, timeoutMs) + } + + async setModel(model: string): Promise { + this._model = model + await this.sendControlRequest({ subtype: 'set_model', model }) + } + + async setPermissionMode(mode: string): Promise { + this._permissionMode = mode + await this.sendControlRequest({ subtype: 'set_permission_mode', mode }) + } + + async getMcpStatus(): Promise { + const response = await this.sendControlRequest({ subtype: 'mcp_status' }) + return (response.mcpServers as unknown[]) ?? [] + } + + private waitForControlResponse( + requestId: string, + timeoutMs = 10000, + ): Promise> { + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + this.pendingControlResponses.delete(requestId) + reject(new Error(`Control response timed out for ${requestId}`)) + }, timeoutMs) + this.pendingControlResponses.set(requestId, { resolve, reject, timeout }) + }) + } + + async prompt(content: string): Promise<{ done: boolean; error?: string }> { + if (this._status !== 'running') { + throw new Error( + `Session ${this.sessionId} is not running (status: ${this._status})`, + ) + } + if (this._prompting) { + throw new Error( + `Session ${this.sessionId} is already processing a prompt`, + ) + } + this._prompting = true + this.lastActiveAt = Date.now() + + const userMsg = jsonStringify({ + type: 'user', + message: { role: 'user', content }, + parent_tool_use_id: null, + session_id: this.sessionId, + }) + + this.writeStdin(userMsg) + + return new Promise((resolve, reject) => { + this.promptResolve = (value) => { + this._prompting = false + resolve(value) + } + this.promptReject = (reason) => { + this._prompting = false + reject(reason) + } + }) + } + + async abort(): Promise { + if (!this.child || this.child.killed) return + + const interrupt = jsonStringify({ + type: 'control_request', + request_id: crypto.randomUUID(), + request: { subtype: 'interrupt' }, + }) + this.writeStdin(interrupt) + + await new Promise(resolve => { + const timeout = setTimeout(() => { + this.kill() + resolve() + }, 5000) + const check = setInterval(() => { + if (!this.child || this.child.killed) { + clearTimeout(timeout) + clearInterval(check) + resolve() + } + }, 200) + }) + } + + replyPermission( + requestId: string, + behavior: 'allow' | 'deny', + opts?: { updatedInput?: Record; message?: string }, + ): void { + const response = { + type: 'control_response', + response: { + subtype: 'success', + request_id: requestId, + response: + behavior === 'allow' + ? { behavior: 'allow', updatedInput: opts?.updatedInput } + : { behavior: 'deny', message: opts?.message ?? 'Denied' }, + }, + } + this.writeStdin(jsonStringify(response)) + this.pendingPermissions.delete(requestId) + } + + replyQuestion( + requestId: string, + action: 'accept' | 'decline', + content?: Record, + ): void { + const response = { + type: 'control_response', + response: { + subtype: 'success', + request_id: requestId, + response: + action === 'accept' + ? { action: 'accept', content } + : { action: 'decline' }, + }, + } + this.writeStdin(jsonStringify(response)) + this.pendingQuestions.delete(requestId) + } + + kill(): void { + if (this.child && !this.child.killed) { + if (process.platform === 'win32') { + this.child.kill() + } else { + this.child.kill('SIGTERM') + } + } + this._status = 'stopped' + this.rejectAllPendingControlResponses(new Error('Session killed')) + } + + forceKill(): void { + if (this.child && !this.child.killed) { + if (process.platform === 'win32') { + this.child.kill() + } else { + this.child.kill('SIGKILL') + } + } + this._status = 'stopped' + this.rejectAllPendingControlResponses(new Error('Session force killed')) + } + + getPendingPermissions(): PendingPermission[] { + return [...this.pendingPermissions.values()] + } + + getPendingPermission(requestId: string): PendingPermission | undefined { + return this.pendingPermissions.get(requestId) + } + + getPendingQuestions(): PendingQuestion[] { + return [...this.pendingQuestions.values()] + } + + getPendingQuestion(requestId: string): PendingQuestion | undefined { + return this.pendingQuestions.get(requestId) + } + + getLastStderr(): string[] { + return [...this.lastStderr] + } + + getInfo() { + return { + session_id: this.sessionId, + status: this._status, + cwd: this.cwd, + title: this._title, + model: this._model, + permission_mode: this._permissionMode, + created_at: this.createdAt, + last_active_at: this.lastActiveAt, + cost_usd: this._costUsd, + input_tokens: this._inputTokens, + output_tokens: this._outputTokens, + } + } + + private writeStdin(data: string): void { + if (this.child?.stdin && !this.child.stdin.destroyed) { + this.child.stdin.write(data + '\n') + } + } + + private rejectAllPendingControlResponses(error: Error): void { + for (const [id, pending] of this.pendingControlResponses) { + clearTimeout(pending.timeout) + pending.reject(error) + } + this.pendingControlResponses.clear() + } +} diff --git a/src/server/sessionManager.ts b/src/server/sessionManager.ts index 5684d78f8..1e48718fc 100644 --- a/src/server/sessionManager.ts +++ b/src/server/sessionManager.ts @@ -1,3 +1,274 @@ -// Auto-generated stub — replace with real implementation -export {}; -export const SessionManager: new (...args: unknown[]) => { destroyAll(): Promise; [key: string]: unknown } = class { async destroyAll() {} } as never; +import { readFile, writeFile, mkdir } from 'fs/promises' +import { existsSync } from 'fs' +import { join } from 'path' +import { getClaudeConfigHomeDir } from '../utils/envUtils.js' +import { logError } from '../utils/log.js' +import type { EventBus } from './eventBus.js' +import { SessionHandle } from './sessionHandle.js' +import type { SessionIndex, SessionIndexEntry, SessionState } from './types.js' + +const INDEX_FILE = 'server-sessions.json' + +export class SessionManager { + private sessions = new Map() + private eventBus: EventBus + private maxSessions: number + private idleTimeoutMs: number + private defaultWorkspace?: string + private idleCheckInterval: ReturnType | null = null + private indexDirty = false + + constructor(opts: { + eventBus: EventBus + maxSessions?: number + idleTimeoutMs?: number + workspace?: string + }) { + this.eventBus = opts.eventBus + this.maxSessions = opts.maxSessions ?? 32 + this.idleTimeoutMs = opts.idleTimeoutMs ?? 1800000 + this.defaultWorkspace = opts.workspace + } + + async init(): Promise { + await this.loadIndex() + this.startIdleCheck() + } + + private getIndexPath(): string { + return join(getClaudeConfigHomeDir(), INDEX_FILE) + } + + private async loadIndex(): Promise { + const path = this.getIndexPath() + if (!existsSync(path)) return + try { + const raw = await readFile(path, 'utf-8') + const index: SessionIndex = JSON.parse(raw) + for (const [, entry] of Object.entries(index)) { + entry + } + } catch (err) { + logError(err as Error) + } + } + + private async saveIndex(): Promise { + const index: SessionIndex = {} + for (const [key, handle] of this.sessions) { + const info = handle.getInfo() + index[key] = { + sessionId: handle.sessionId, + transcriptSessionId: handle.sessionId, + cwd: handle.cwd, + permissionMode: info.permission_mode, + createdAt: info.created_at, + lastActiveAt: info.last_active_at, + } + } + const dir = getClaudeConfigHomeDir() + if (!existsSync(dir)) { + await mkdir(dir, { recursive: true }) + } + try { + await writeFile(this.getIndexPath(), JSON.stringify(index, null, 2)) + } catch (err) { + logError(err as Error) + } + } + + private scheduleIndexSave(): void { + if (this.indexDirty) return + this.indexDirty = true + setTimeout(async () => { + this.indexDirty = false + await this.saveIndex() + }, 1000) + } + + private startIdleCheck(): void { + if (this.idleTimeoutMs <= 0) return + const checkInterval = Math.min(this.idleTimeoutMs / 2, 60000) + this.idleCheckInterval = setInterval(() => { + const now = Date.now() + for (const [id, handle] of this.sessions) { + const info = handle.getInfo() + if ( + info.status === 'running' && + now - info.last_active_at > this.idleTimeoutMs! + ) { + handle.kill() + this.eventBus.publishSessionEvent(id, 'deleted', { + status: 'stopped', + reason: 'idle_timeout', + }) + this.sessions.delete(id) + this.scheduleIndexSave() + } + } + }, checkInterval) + } + + getActiveCount(): number { + return this.sessions.size + } + + getSession(id: string): SessionHandle | undefined { + return this.sessions.get(id) + } + + getAllSessions(): SessionHandle[] { + return [...this.sessions.values()] + } + + getSessionStatuses(): Record< + string, + { status: SessionState; has_pending_permission: boolean } + > { + const result: Record< + string, + { status: SessionState; has_pending_permission: boolean } + > = {} + for (const [id, handle] of this.sessions) { + result[id] = { + status: handle.status, + has_pending_permission: handle.getPendingPermissions().length > 0, + } + } + return result + } + + async createSession(opts: { + cwd?: string + model?: string + permissionMode?: string + systemPrompt?: string + resumeSessionId?: string + execPath: string + scriptArgs: string[] + }): Promise { + if (this.sessions.size >= this.maxSessions) { + throw new Error( + `Maximum concurrent sessions reached (${this.maxSessions})`, + ) + } + + const sessionId = crypto.randomUUID() + const cwd = opts.cwd || this.defaultWorkspace || process.cwd() + + const handle = new SessionHandle({ + sessionId, + cwd, + model: opts.model, + permissionMode: opts.permissionMode, + systemPrompt: opts.systemPrompt, + resumeSessionId: opts.resumeSessionId, + eventBus: this.eventBus, + execPath: opts.execPath, + scriptArgs: opts.scriptArgs, + }) + + this.sessions.set(sessionId, handle) + this.eventBus.publishSessionEvent(sessionId, 'created', { + status: 'starting', + }) + this.scheduleIndexSave() + + try { + await handle.start() + } catch (err) { + this.sessions.delete(sessionId) + this.scheduleIndexSave() + throw err + } + + return handle + } + + async deleteSession(id: string): Promise { + const handle = this.sessions.get(id) + if (!handle) return false + handle.forceKill() + this.sessions.delete(id) + this.eventBus.publishSessionEvent(id, 'deleted', { status: 'stopped' }) + this.scheduleIndexSave() + return true + } + + async destroyAll(): Promise { + for (const [, handle] of this.sessions) { + handle.forceKill() + } + this.sessions.clear() + if (this.idleCheckInterval) { + clearInterval(this.idleCheckInterval) + } + await this.saveIndex() + } + + getAllPendingPermissions(): Array<{ + requestId: string + sessionId: string + toolName: string + toolUseId: string + input: Record + title: string + description: string + }> { + const result: Array<{ + requestId: string + sessionId: string + toolName: string + toolUseId: string + input: Record + title: string + description: string + }> = [] + for (const [, handle] of this.sessions) { + result.push(...handle.getPendingPermissions()) + } + return result + } + + getAllPendingQuestions(): Array<{ + requestId: string + sessionId: string + mcpServerName: string + message: string + mode: string + requestedSchema: Record + }> { + const result: Array<{ + requestId: string + sessionId: string + mcpServerName: string + message: string + mode: string + requestedSchema: Record + }> = [] + for (const [, handle] of this.sessions) { + result.push(...handle.getPendingQuestions()) + } + return result + } + + findPermissionAcrossSessions( + requestId: string, + ): { handle: SessionHandle; perm: import('./sessionHandle.js').PendingPermission } | null { + for (const [, handle] of this.sessions) { + const perm = handle.getPendingPermission(requestId) + if (perm) return { handle, perm } + } + return null + } + + findQuestionAcrossSessions( + requestId: string, + ): { handle: SessionHandle; question: import('./sessionHandle.js').PendingQuestion } | null { + for (const [, handle] of this.sessions) { + const question = handle.getPendingQuestion(requestId) + if (question) return { handle, question } + } + return null + } +} diff --git a/src/server/transcriptReader.ts b/src/server/transcriptReader.ts new file mode 100644 index 000000000..6fc63aa6b --- /dev/null +++ b/src/server/transcriptReader.ts @@ -0,0 +1,305 @@ +import { readFile, readdir, stat } from 'fs/promises' +import { existsSync } from 'fs' +import { join } from 'path' +import { + getProjectsDir, + getProjectDir, + findProjectDir, +} from '../utils/sessionStoragePortable.js' +import { safeParseJSON } from '../utils/json.js' + +type TranscriptEntry = { + type: string + uuid?: string + parentUuid?: string + sessionId?: string + timestamp?: string + role?: string + subtype?: string + message?: { + role?: string + content?: unknown + } + content?: unknown + [key: string]: unknown +} + +export type SessionMessage = { + uuid: string + type: string + role: string + content: unknown + timestamp: number + parent_uuid: string | null + usage?: { input_tokens: number; output_tokens: number } +} + +export type TodoItem = { + id: string + content: string + status: string + priority: string +} + +const MESSAGE_TYPES = new Set(['user', 'assistant']) +const SKIP_TYPES = new Set([ + 'summary', + 'tag', + 'mode', + 'agent-setting', + 'pr-link', + 'customTitle', + 'file-history-snapshot', + 'attribution-snapshot', + 'context-collapse-commit', + 'context-collapse-snapshot', +]) + +const pathCache = new Map() + +async function resolveTranscriptPath( + sessionId: string, + cwd?: string, +): Promise { + const cacheKey = `${sessionId}:${cwd ?? ''}` + const cached = pathCache.get(cacheKey) + if (cached !== undefined) return cached + + const result = await resolveTranscriptPathUncached(sessionId, cwd) + pathCache.set(cacheKey, result) + return result +} + +async function resolveTranscriptPathUncached( + sessionId: string, + cwd?: string, +): Promise { + if (cwd) { + const projectDir = getProjectDir(cwd) + const direct = join(projectDir, `${sessionId}.jsonl`) + if (existsSync(direct)) return direct + } + + const found = await findProjectDir(cwd ?? process.cwd()) + if (found) { + const direct = join(found, `${sessionId}.jsonl`) + if (existsSync(direct)) return direct + } + + const projectsDir = getProjectsDir() + const target = `${sessionId}.jsonl` + try { + const dirs = await readdir(projectsDir) + for (const dir of dirs) { + const candidate = join(projectsDir, dir, target) + if (existsSync(candidate)) return candidate + } + } catch {} + + return null +} + +function parseEntry(line: string): TranscriptEntry | null { + const parsed = safeParseJSON(line, false) + if (!parsed || typeof parsed !== 'object') return null + return parsed as TranscriptEntry +} + +export async function readSessionMessages(opts: { + sessionId: string + cwd?: string + limit?: number + before?: string + includeSystem?: boolean +}): Promise<{ + messages: SessionMessage[] + nextCursor?: string +}> { + const path = await resolveTranscriptPath(opts.sessionId, opts.cwd) + if (!path || !existsSync(path)) { + return { messages: [] } + } + + const raw = await readFile(path, 'utf-8') + const lines = raw.split('\n').filter(Boolean) + + let lastCompactBoundaryIndex = -1 + for (let i = lines.length - 1; i >= 0; i--) { + const entry = parseEntry(lines[i]) + if ( + entry?.type === 'system' && + entry?.subtype === 'compact_boundary' && + !entry?.compactMetadata?.preservedSegment + ) { + lastCompactBoundaryIndex = i + break + } + } + + const startLine = lastCompactBoundaryIndex >= 0 ? lastCompactBoundaryIndex + 1 : 0 + + const allMessages: SessionMessage[] = [] + let cursorIndex = -1 + + for (let i = startLine; i < lines.length; i++) { + const entry = parseEntry(lines[i]) + if (!entry) continue + if (!entry.type || !entry.uuid) continue + if (SKIP_TYPES.has(entry.type)) continue + if (entry.type === 'attribution-snapshot') continue + + const includeThis = + MESSAGE_TYPES.has(entry.type) || + (opts.includeSystem && entry.type === 'system') + + if (!includeThis) continue + + if (entry.uuid === opts.before) { + cursorIndex = allMessages.length + } + + const role = entry.role ?? entry.message?.role ?? entry.type + const content = entry.message?.content ?? entry.content ?? '' + const timestamp = entry.timestamp + ? new Date(entry.timestamp).getTime() + : 0 + + allMessages.push({ + uuid: entry.uuid, + type: entry.type, + role: role ?? entry.type, + content, + timestamp, + parent_uuid: entry.parentUuid ?? null, + usage: entry.usage as + | { input_tokens: number; output_tokens: number } + | undefined, + }) + } + + const startIndex = cursorIndex >= 0 ? cursorIndex + 1 : 0 + const sliced = allMessages.slice(startIndex) + const limited = opts.limit ? sliced.slice(0, opts.limit) : sliced + + const nextCursor = + opts.limit && sliced.length > opts.limit + ? sliced[opts.limit]?.uuid + : undefined + + return { messages: limited, nextCursor } +} + +export async function readSessionTodos(opts: { + sessionId: string + cwd?: string +}): Promise { + const path = await resolveTranscriptPath(opts.sessionId, opts.cwd) + if (!path || !existsSync(path)) return [] + + const raw = await readFile(path, 'utf-8') + const lines = raw.split('\n').filter(Boolean) + + let latestTodos: TodoItem[] = [] + + for (const line of lines) { + const entry = parseEntry(line) + if (!entry) continue + + if (entry.type === 'assistant' && entry.message?.content) { + const content = entry.message.content + if (!Array.isArray(content)) continue + + for (const block of content) { + if (!block || typeof block !== 'object') continue + const b = block as Record + if (b.type === 'tool_use' && (b.name === 'TodoWrite' || b.name === 'TaskCreateTool')) { + const input = b.input as + | { todos?: Array<{ id: string; content: string; status: string; priority: string }> } + | undefined + if (input?.todos) { + latestTodos = input.todos + } + } + } + } + } + + return latestTodos +} + +export async function readSessionDiff(opts: { + sessionId: string + cwd?: string + messageID?: string +}): Promise<{ + diffs: Array<{ + file: string + status: string + additions: number + deletions: number + patch: string + }> +}> { + const { execSync } = await import('child_process') + const cwd = opts.cwd ?? process.cwd() + + try { + const diffOutput = execSync('git diff HEAD', { + encoding: 'utf-8', + timeout: 10000, + cwd, + }) + + if (!diffOutput.trim()) return { diffs: [] } + + const files = new Map< + string, + { additions: number; deletions: number; patch: string } + >() + let currentFile = '' + let currentPatch: string[] = [] + + for (const line of diffOutput.split('\n')) { + const match = /^diff --git a\/(.+?) b\/(.+?)$/.exec(line) + if (match) { + if (currentFile && currentPatch.length > 0) { + files.set(currentFile, { + additions: currentPatch.filter(l => l.startsWith('+') && !l.startsWith('+++')).length, + deletions: currentPatch.filter(l => l.startsWith('-') && !l.startsWith('---')).length, + patch: currentPatch.join('\n'), + }) + } + currentFile = match[2] + currentPatch = [line] + } else { + currentPatch.push(line) + } + } + if (currentFile && currentPatch.length > 0) { + files.set(currentFile, { + additions: currentPatch.filter(l => l.startsWith('+') && !l.startsWith('+++')).length, + deletions: currentPatch.filter(l => l.startsWith('-') && !l.startsWith('---')).length, + patch: currentPatch.join('\n'), + }) + } + + return { + diffs: [...files.entries()].map(([file, info]) => ({ + file, + status: + info.additions > 0 && info.deletions > 0 + ? 'modified' + : info.additions > 0 + ? 'added' + : 'removed', + ...info, + })), + } + } catch { + return { diffs: [] } + } +} + +export function clearPathCache(): void { + pathCache.clear() +} diff --git a/src/server/types.ts b/src/server/types.ts index 7f876a020..b305bc8fa 100644 --- a/src/server/types.ts +++ b/src/server/types.ts @@ -13,7 +13,7 @@ export const connectResponseSchema = lazySchema(() => export type ServerConfig = { port: number host: string - authToken: string + authToken?: string unix?: string /** Idle timeout for detached sessions (ms). 0 = never expire. */ idleTimeoutMs?: number