diff --git a/src/server/__tests__/conversations.test.ts b/src/server/__tests__/conversations.test.ts new file mode 100644 index 00000000..4d379135 --- /dev/null +++ b/src/server/__tests__/conversations.test.ts @@ -0,0 +1,257 @@ +/** + * Tests for ConversationService and WebSocket chat integration + * + * ConversationService 管理 CLI 子进程的生命周期。 + * WebSocket 集成测试验证消息从客户端经过服务端到达 CLI 的完整流转。 + */ + +import { describe, it, expect, beforeAll, afterAll } from 'bun:test' +import * as fs from 'fs/promises' +import * as path from 'path' +import * as os from 'os' +import { ConversationService } from '../services/conversationService.js' + +// ============================================================================ +// ConversationService unit tests +// ============================================================================ + +describe('ConversationService', () => { + it('should report no session for unknown ID', () => { + const svc = new ConversationService() + const sid = crypto.randomUUID() + expect(svc.hasSession(sid)).toBe(false) + }) + + it('should track active sessions as empty initially', () => { + const svc = new ConversationService() + expect(svc.getActiveSessions()).toEqual([]) + }) + + it('should return false when sending message to non-existent session', () => { + const svc = new ConversationService() + const result = svc.sendMessage('no-such-session', 'hello') + expect(result).toBe(false) + }) + + it('should return false when responding to permission for non-existent session', () => { + const svc = new ConversationService() + const result = svc.respondToPermission('no-such-session', 'req-1', true) + expect(result).toBe(false) + }) + + it('should return false when sending interrupt to non-existent session', () => { + const svc = new ConversationService() + const result = svc.sendInterrupt('no-such-session') + expect(result).toBe(false) + }) + + it('should not throw when stopping non-existent session', () => { + const svc = new ConversationService() + expect(() => svc.stopSession('no-such-session')).not.toThrow() + }) + + it('should not throw when registering callback for non-existent session', () => { + const svc = new ConversationService() + expect(() => svc.onOutput('no-such-session', () => {})).not.toThrow() + }) +}) + +// ============================================================================ +// WebSocket integration tests (with real server, CLI falls back to echo) +// ============================================================================ + +describe('WebSocket Chat Integration', () => { + let server: ReturnType + let baseUrl: string + let wsUrl: string + let tmpDir: string + + beforeAll(async () => { + tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'claude-conv-')) + process.env.CLAUDE_CONFIG_DIR = tmpDir + await fs.mkdir(path.join(tmpDir, 'projects'), { recursive: true }) + + const port = 15000 + Math.floor(Math.random() * 1000) + const { startServer } = await import('../index.js') + server = startServer(port, '127.0.0.1') + baseUrl = `http://127.0.0.1:${port}` + wsUrl = `ws://127.0.0.1:${port}` + }) + + afterAll(async () => { + server?.stop() + if (tmpDir) { + await fs.rm(tmpDir, { recursive: true, force: true }) + } + delete process.env.CLAUDE_CONFIG_DIR + }) + + it('should connect and receive connected event', async () => { + const messages: any[] = [] + const ws = new WebSocket(`${wsUrl}/ws/chat-test-1`) + + await new Promise((resolve) => { + ws.onmessage = (e) => { + messages.push(JSON.parse(e.data as string)) + if (messages.length >= 1) { + ws.close() + resolve() + } + } + ws.onerror = () => { + ws.close() + resolve() + } + setTimeout(() => { + ws.close() + resolve() + }, 3000) + }) + + expect(messages[0].type).toBe('connected') + expect(messages[0].sessionId).toBe('chat-test-1') + }) + + it('should handle stop_generation and return idle status', async () => { + const messages: any[] = [] + const ws = new WebSocket(`${wsUrl}/ws/chat-test-2`) + + await new Promise((resolve) => { + ws.onmessage = (e) => { + const msg = JSON.parse(e.data as string) + messages.push(msg) + if (msg.type === 'connected') { + ws.send(JSON.stringify({ type: 'stop_generation' })) + } + if (msg.type === 'status' && msg.state === 'idle') { + ws.close() + resolve() + } + } + ws.onerror = () => { + ws.close() + resolve() + } + setTimeout(() => { + ws.close() + resolve() + }, 3000) + }) + + expect(messages.some((m) => m.type === 'status' && m.state === 'idle')).toBe(true) + }) + + it('should send user_message and receive fallback echo response', async () => { + const messages: any[] = [] + const ws = new WebSocket(`${wsUrl}/ws/chat-test-3`) + + await new Promise((resolve) => { + ws.onmessage = (e) => { + const msg = JSON.parse(e.data as string) + messages.push(msg) + if (msg.type === 'connected') { + ws.send( + JSON.stringify({ type: 'user_message', content: 'Hello from test' }) + ) + } + // Wait until we receive idle status after the echo + if ( + msg.type === 'status' && + msg.state === 'idle' && + messages.length > 3 + ) { + ws.close() + resolve() + } + } + ws.onerror = () => { + ws.close() + resolve() + } + setTimeout(() => { + ws.close() + resolve() + }, 5000) + }) + + const types = messages.map((m) => m.type) + expect(types).toContain('connected') + expect(types).toContain('status') + // Fallback echo produces content_start, content_delta, message_complete, status + expect(types).toContain('content_start') + expect(types).toContain('content_delta') + expect(types).toContain('message_complete') + + // Verify thinking was first status + const statusMsgs = messages.filter((m) => m.type === 'status') + expect(statusMsgs[0].state).toBe('thinking') + }) + + it('should handle permission_response without error', async () => { + const messages: any[] = [] + const ws = new WebSocket(`${wsUrl}/ws/chat-test-4`) + + await new Promise((resolve) => { + ws.onmessage = (e) => { + const msg = JSON.parse(e.data as string) + messages.push(msg) + if (msg.type === 'connected') { + // Send a permission response (no active session, should not crash) + ws.send( + JSON.stringify({ + type: 'permission_response', + requestId: 'test-req-1', + allowed: true, + }) + ) + // Give a moment then close + setTimeout(() => { + ws.close() + resolve() + }, 500) + } + } + ws.onerror = () => { + ws.close() + resolve() + } + setTimeout(() => { + ws.close() + resolve() + }, 3000) + }) + + // Should have received connected and no error + expect(messages[0].type).toBe('connected') + expect(messages.some((m) => m.type === 'error')).toBe(false) + }) + + it('should handle ping/pong', async () => { + const messages: any[] = [] + const ws = new WebSocket(`${wsUrl}/ws/chat-test-5`) + + await new Promise((resolve) => { + ws.onmessage = (e) => { + const msg = JSON.parse(e.data as string) + messages.push(msg) + if (msg.type === 'connected') { + ws.send(JSON.stringify({ type: 'ping' })) + } + if (msg.type === 'pong') { + ws.close() + resolve() + } + } + ws.onerror = () => { + ws.close() + resolve() + } + setTimeout(() => { + ws.close() + resolve() + }, 3000) + }) + + expect(messages.some((m) => m.type === 'pong')).toBe(true) + }) +}) diff --git a/src/server/__tests__/tasks.test.ts b/src/server/__tests__/tasks.test.ts new file mode 100644 index 00000000..ea1c7a39 --- /dev/null +++ b/src/server/__tests__/tasks.test.ts @@ -0,0 +1,171 @@ +/** + * Unit tests for TaskService and Tasks API + */ + +import { describe, it, expect, beforeEach, afterEach } from 'bun:test' +import * as fs from 'fs/promises' +import * as path from 'path' +import * as os from 'os' +import { TaskService } from '../services/taskService.js' + +// ============================================================================ +// TaskService unit tests +// ============================================================================ + +describe('TaskService', () => { + let tmpDir: string + + beforeEach(async () => { + tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'claude-tasks-')) + process.env.CLAUDE_CONFIG_DIR = tmpDir + }) + + afterEach(async () => { + await fs.rm(tmpDir, { recursive: true, force: true }) + delete process.env.CLAUDE_CONFIG_DIR + }) + + it('should return empty list when no tasks dir', async () => { + const svc = new TaskService() + const tasks = await svc.listTasks() + expect(tasks).toEqual([]) + }) + + it('should list tasks from JSON files', async () => { + const tasksDir = path.join(tmpDir, 'tasks') + await fs.mkdir(tasksDir, { recursive: true }) + + await fs.writeFile(path.join(tasksDir, 'task-001.json'), JSON.stringify({ + id: 'task-001', + type: 'local_agent', + status: 'completed', + name: 'code-review', + description: 'Review PR #42', + createdAt: Date.now() - 60000, + completedAt: Date.now(), + })) + + await fs.writeFile(path.join(tasksDir, 'task-002.json'), JSON.stringify({ + id: 'task-002', + type: 'in_process_teammate', + status: 'running', + name: 'frontend-dev', + teamName: 'ui-team', + createdAt: Date.now(), + })) + + const svc = new TaskService() + const tasks = await svc.listTasks() + expect(tasks.length).toBe(2) + // 按 createdAt 倒序 + expect(tasks[0].id).toBe('task-002') + expect(tasks[1].id).toBe('task-001') + }) + + it('should scan nested team task directories', async () => { + const teamDir = path.join(tmpDir, 'tasks', 'my-team') + await fs.mkdir(teamDir, { recursive: true }) + + await fs.writeFile(path.join(teamDir, 'member-1.json'), JSON.stringify({ + id: 'member-1', + type: 'in_process_teammate', + status: 'completed', + teamName: 'my-team', + })) + + const svc = new TaskService() + const tasks = await svc.listTasks() + expect(tasks.length).toBe(1) + expect(tasks[0].teamName).toBe('my-team') + }) + + it('should get single task by ID', async () => { + const tasksDir = path.join(tmpDir, 'tasks') + await fs.mkdir(tasksDir, { recursive: true }) + + await fs.writeFile(path.join(tasksDir, 'abc.json'), JSON.stringify({ + id: 'abc', + type: 'local_shell', + status: 'failed', + name: 'build', + })) + + const svc = new TaskService() + const task = await svc.getTask('abc') + expect(task).toBeDefined() + expect(task!.status).toBe('failed') + }) + + it('should return null for unknown task', async () => { + const svc = new TaskService() + const task = await svc.getTask('nonexistent') + expect(task).toBeNull() + }) + + it('should skip invalid JSON files gracefully', async () => { + const tasksDir = path.join(tmpDir, 'tasks') + await fs.mkdir(tasksDir, { recursive: true }) + await fs.writeFile(path.join(tasksDir, 'bad.json'), 'not json {{{') + + const svc = new TaskService() + const tasks = await svc.listTasks() + expect(tasks).toEqual([]) + }) +}) + +// ============================================================================ +// Tasks API integration tests +// ============================================================================ + +describe('Tasks API', () => { + let server: any + let baseUrl: string + let tmpDir: string + + beforeEach(async () => { + tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'claude-tasks-api-')) + process.env.CLAUDE_CONFIG_DIR = tmpDir + await fs.mkdir(path.join(tmpDir, 'projects'), { recursive: true }) + + const port = 15500 + Math.floor(Math.random() * 500) + const { startServer } = await import('../../server/index.js') + server = startServer(port, '127.0.0.1') + baseUrl = `http://127.0.0.1:${port}` + }) + + afterEach(async () => { + server?.stop() + await fs.rm(tmpDir, { recursive: true, force: true }) + delete process.env.CLAUDE_CONFIG_DIR + }) + + it('should return empty tasks list', async () => { + const res = await fetch(`${baseUrl}/api/tasks`) + const data = await res.json() + expect(res.status).toBe(200) + expect(data.tasks).toEqual([]) + }) + + it('should return tasks when files exist', async () => { + const tasksDir = path.join(tmpDir, 'tasks') + await fs.mkdir(tasksDir, { recursive: true }) + await fs.writeFile(path.join(tasksDir, 'test.json'), JSON.stringify({ + id: 'test', type: 'local_agent', status: 'completed', name: 'test-task', + })) + + const res = await fetch(`${baseUrl}/api/tasks`) + const data = await res.json() + expect(data.tasks.length).toBe(1) + expect(data.tasks[0].name).toBe('test-task') + }) + + it('should return 404 for unknown task', async () => { + const res = await fetch(`${baseUrl}/api/tasks/nonexistent`) + expect(res.status).toBe(404) + }) + + it('should reject non-GET methods', async () => { + const res = await fetch(`${baseUrl}/api/tasks`, { method: 'POST' }) + expect(res.status).toBe(405) + }) +}) diff --git a/src/server/__tests__/teams.test.ts b/src/server/__tests__/teams.test.ts new file mode 100644 index 00000000..f3d5918f --- /dev/null +++ b/src/server/__tests__/teams.test.ts @@ -0,0 +1,484 @@ +/** + * Unit tests for TeamService and Teams API + */ + +import { describe, it, expect, beforeEach, afterEach } from 'bun:test' +import * as fs from 'node:fs/promises' +import * as path from 'node:path' +import * as os from 'node:os' +import { TeamService } from '../services/teamService.js' + +// ============================================================================ +// Test helpers +// ============================================================================ + +let tmpDir: string +let service: TeamService + +async function setupTmpConfigDir(): Promise { + tmpDir = path.join( + os.tmpdir(), + `claude-teams-test-${Date.now()}-${Math.random().toString(36).slice(2)}`, + ) + await fs.mkdir(path.join(tmpDir, 'teams'), { recursive: true }) + await fs.mkdir(path.join(tmpDir, 'projects'), { recursive: true }) + process.env.CLAUDE_CONFIG_DIR = tmpDir + return tmpDir +} + +async function cleanupTmpDir(): Promise { + if (tmpDir) { + await fs.rm(tmpDir, { recursive: true, force: true }) + } + delete process.env.CLAUDE_CONFIG_DIR +} + +/** Write a team config.json to the temp directory. */ +async function writeTeamConfig( + teamName: string, + config: Record, +): Promise { + const teamDir = path.join(tmpDir, 'teams', teamName) + await fs.mkdir(teamDir, { recursive: true }) + const configPath = path.join(teamDir, 'config.json') + await fs.writeFile(configPath, JSON.stringify(config), 'utf-8') + return configPath +} + +/** Write a mock JSONL transcript file under projects. */ +async function writeTranscriptFile( + projectDir: string, + sessionId: string, + entries: Record[], +): Promise { + const dir = path.join(tmpDir, 'projects', projectDir) + await fs.mkdir(dir, { recursive: true }) + const filePath = path.join(dir, `${sessionId}.jsonl`) + const content = entries.map((e) => JSON.stringify(e)).join('\n') + '\n' + await fs.writeFile(filePath, content, 'utf-8') + return filePath +} + +/** Create a standard team config for testing. */ +function makeTeamConfig(overrides?: Record) { + return { + name: 'test-team', + description: 'A test team', + createdAt: 1700000000000, + leadAgentId: 'agent-lead', + members: [ + { + agentId: 'agent-lead', + name: 'Lead Agent', + agentType: 'lead', + model: 'claude-opus-4-6', + color: '#ff0000', + joinedAt: 1700000000000, + tmuxPaneId: '%0', + cwd: '/tmp/project', + sessionId: 'session-lead-001', + isActive: true, + }, + { + agentId: 'agent-worker', + name: 'Worker Agent', + agentType: 'worker', + model: 'claude-sonnet-4-20250514', + color: '#00ff00', + joinedAt: 1700000001000, + tmuxPaneId: '%1', + cwd: '/tmp/project/src', + sessionId: 'session-worker-001', + isActive: false, + }, + ], + ...overrides, + } +} + +// ============================================================================ +// TeamService tests +// ============================================================================ + +describe('TeamService', () => { + beforeEach(async () => { + await setupTmpConfigDir() + service = new TeamService() + }) + + afterEach(async () => { + await cleanupTmpDir() + }) + + // -------------------------------------------------------------------------- + // listTeams + // -------------------------------------------------------------------------- + + it('should return empty list when no teams exist', async () => { + const teams = await service.listTeams() + expect(teams).toEqual([]) + }) + + it('should return empty list when teams directory does not exist', async () => { + await fs.rm(path.join(tmpDir, 'teams'), { recursive: true, force: true }) + const teams = await service.listTeams() + expect(teams).toEqual([]) + }) + + it('should list teams from config files', async () => { + await writeTeamConfig('alpha', makeTeamConfig({ name: 'alpha' })) + await writeTeamConfig('beta', makeTeamConfig({ name: 'beta', description: 'Beta team' })) + + const teams = await service.listTeams() + expect(teams).toHaveLength(2) + + const names = teams.map((t) => t.name).sort() + expect(names).toEqual(['alpha', 'beta']) + }) + + it('should compute memberCount and activeMemberCount', async () => { + await writeTeamConfig('gamma', makeTeamConfig({ name: 'gamma' })) + + const teams = await service.listTeams() + expect(teams).toHaveLength(1) + expect(teams[0]!.memberCount).toBe(2) + expect(teams[0]!.activeMemberCount).toBe(1) // only lead is active + }) + + it('should skip malformed team directories', async () => { + // Create a team dir with invalid JSON + const badDir = path.join(tmpDir, 'teams', 'bad-team') + await fs.mkdir(badDir, { recursive: true }) + await fs.writeFile(path.join(badDir, 'config.json'), 'not json', 'utf-8') + + // Also create a valid team + await writeTeamConfig('good-team', makeTeamConfig({ name: 'good-team' })) + + const teams = await service.listTeams() + expect(teams).toHaveLength(1) + expect(teams[0]!.name).toBe('good-team') + }) + + // -------------------------------------------------------------------------- + // getTeam + // -------------------------------------------------------------------------- + + it('should return team detail with members', async () => { + await writeTeamConfig('detail-team', makeTeamConfig({ name: 'detail-team' })) + + const detail = await service.getTeam('detail-team') + expect(detail.name).toBe('detail-team') + expect(detail.leadAgentId).toBe('agent-lead') + expect(detail.members).toHaveLength(2) + expect(detail.members[0]!.agentId).toBe('agent-lead') + expect(detail.members[1]!.agentId).toBe('agent-worker') + }) + + it('should derive running status for active member', async () => { + await writeTeamConfig('status-team', makeTeamConfig({ name: 'status-team' })) + + const detail = await service.getTeam('status-team') + const lead = detail.members.find((m) => m.agentId === 'agent-lead')! + expect(lead.status).toBe('running') + }) + + it('should derive idle status for inactive member', async () => { + await writeTeamConfig('status-team', makeTeamConfig({ name: 'status-team' })) + + const detail = await service.getTeam('status-team') + const worker = detail.members.find((m) => m.agentId === 'agent-worker')! + expect(worker.status).toBe('idle') + }) + + it('should derive running status when isActive is undefined', async () => { + const config = makeTeamConfig({ name: 'undef-team' }) + // Remove isActive from the first member to simulate undefined + delete (config.members[0] as Record).isActive + await writeTeamConfig('undef-team', config) + + const detail = await service.getTeam('undef-team') + const lead = detail.members.find((m) => m.agentId === 'agent-lead')! + expect(lead.status).toBe('running') + }) + + it('should throw 404 for non-existent team', async () => { + expect(service.getTeam('nonexistent')).rejects.toThrow('Team not found') + }) + + // -------------------------------------------------------------------------- + // getMemberTranscript + // -------------------------------------------------------------------------- + + it('should return transcript messages for a member', async () => { + await writeTeamConfig('transcript-team', makeTeamConfig({ name: 'transcript-team' })) + + // Write a mock transcript JSONL for the lead session + await writeTranscriptFile('-tmp-project', 'session-lead-001', [ + { + type: 'file-history-snapshot', + messageId: 'snap-1', + snapshot: {}, + }, + { + type: 'user', + uuid: 'msg-user-1', + message: { role: 'user', content: 'Hello team' }, + timestamp: '2026-01-01T00:01:00.000Z', + }, + { + type: 'assistant', + uuid: 'msg-asst-1', + message: { + role: 'assistant', + content: [{ type: 'text', text: 'Hi! Ready to help.' }], + }, + timestamp: '2026-01-01T00:02:00.000Z', + }, + ]) + + const messages = await service.getMemberTranscript( + 'transcript-team', + 'agent-lead', + ) + expect(messages).toHaveLength(2) + expect(messages[0]!.type).toBe('user') + expect(messages[0]!.id).toBe('msg-user-1') + expect(messages[1]!.type).toBe('assistant') + }) + + it('should return empty array when member has no sessionId', async () => { + const config = makeTeamConfig({ name: 'no-session-team' }) + delete (config.members[0] as Record).sessionId + await writeTeamConfig('no-session-team', config) + + const messages = await service.getMemberTranscript( + 'no-session-team', + 'agent-lead', + ) + expect(messages).toEqual([]) + }) + + it('should return empty array when transcript file not found', async () => { + await writeTeamConfig('no-file-team', makeTeamConfig({ name: 'no-file-team' })) + + // Don't write any transcript file + const messages = await service.getMemberTranscript( + 'no-file-team', + 'agent-lead', + ) + expect(messages).toEqual([]) + }) + + it('should throw 404 for unknown member', async () => { + await writeTeamConfig('member-team', makeTeamConfig({ name: 'member-team' })) + + expect( + service.getMemberTranscript('member-team', 'nonexistent-agent'), + ).rejects.toThrow('Member not found') + }) + + it('should skip meta entries in transcript', async () => { + await writeTeamConfig('meta-team', makeTeamConfig({ name: 'meta-team' })) + + await writeTranscriptFile('-tmp-project', 'session-lead-001', [ + { + type: 'user', + uuid: 'msg-meta', + message: { role: 'user', content: 'internal meta' }, + isMeta: true, + timestamp: '2026-01-01T00:00:30.000Z', + }, + { + type: 'user', + uuid: 'msg-real', + message: { role: 'user', content: 'Real message' }, + timestamp: '2026-01-01T00:01:00.000Z', + }, + ]) + + const messages = await service.getMemberTranscript('meta-team', 'agent-lead') + expect(messages).toHaveLength(1) + expect(messages[0]!.id).toBe('msg-real') + }) + + // -------------------------------------------------------------------------- + // deleteTeam + // -------------------------------------------------------------------------- + + it('should delete a team with no active members', async () => { + const config = makeTeamConfig({ name: 'deletable' }) + // Set all members to inactive + for (const member of config.members) { + ;(member as Record).isActive = false + } + await writeTeamConfig('deletable', config) + + await service.deleteTeam('deletable') + + // Team dir should be gone + const teamDir = path.join(tmpDir, 'teams', 'deletable') + expect(fs.access(teamDir)).rejects.toThrow() + }) + + it('should refuse to delete a team with active members', async () => { + await writeTeamConfig('active-team', makeTeamConfig({ name: 'active-team' })) + + expect(service.deleteTeam('active-team')).rejects.toThrow( + 'has active members', + ) + }) + + it('should throw 404 when deleting non-existent team', async () => { + expect(service.deleteTeam('ghost')).rejects.toThrow('Team not found') + }) +}) + +// ============================================================================ +// Teams API integration tests +// ============================================================================ + +describe('Teams API', () => { + let baseUrl: string + let server: ReturnType | null = null + + beforeEach(async () => { + await setupTmpConfigDir() + service = new TeamService() + + const { handleTeamsApi } = await import('../api/teams.js') + + const port = 40000 + Math.floor(Math.random() * 10000) + baseUrl = `http://127.0.0.1:${port}` + + server = Bun.serve({ + port, + hostname: '127.0.0.1', + + async fetch(req) { + const url = new URL(req.url) + const segments = url.pathname.split('/').filter(Boolean) + + if (segments[0] === 'api' && segments[1] === 'teams') { + return handleTeamsApi(req, url, segments) + } + + return new Response('Not Found', { status: 404 }) + }, + }) + }) + + afterEach(async () => { + if (server) { + server.stop(true) + server = null + } + await cleanupTmpDir() + }) + + it('GET /api/teams should return empty list', async () => { + const res = await fetch(`${baseUrl}/api/teams`) + expect(res.status).toBe(200) + + const body = (await res.json()) as { teams: unknown[] } + expect(body.teams).toEqual([]) + }) + + it('GET /api/teams should list teams', async () => { + await writeTeamConfig('api-team', makeTeamConfig({ name: 'api-team' })) + + const res = await fetch(`${baseUrl}/api/teams`) + expect(res.status).toBe(200) + + const body = (await res.json()) as { teams: Array<{ name: string }> } + expect(body.teams).toHaveLength(1) + expect(body.teams[0]!.name).toBe('api-team') + }) + + it('GET /api/teams/:name should return team detail', async () => { + await writeTeamConfig('detail', makeTeamConfig({ name: 'detail' })) + + const res = await fetch(`${baseUrl}/api/teams/detail`) + expect(res.status).toBe(200) + + const body = (await res.json()) as { + name: string + leadAgentId: string + members: Array<{ agentId: string }> + } + expect(body.name).toBe('detail') + expect(body.leadAgentId).toBe('agent-lead') + expect(body.members).toHaveLength(2) + }) + + it('GET /api/teams/:name should 404 for unknown team', async () => { + const res = await fetch(`${baseUrl}/api/teams/nonexistent`) + expect(res.status).toBe(404) + }) + + it('GET /api/teams/:name/members/:id/transcript should return messages', async () => { + await writeTeamConfig('t-team', makeTeamConfig({ name: 't-team' })) + + await writeTranscriptFile('-tmp-project', 'session-lead-001', [ + { + type: 'user', + uuid: 'u1', + message: { role: 'user', content: 'Hello' }, + timestamp: '2026-01-01T00:01:00.000Z', + }, + ]) + + const res = await fetch( + `${baseUrl}/api/teams/t-team/members/agent-lead/transcript`, + ) + expect(res.status).toBe(200) + + const body = (await res.json()) as { + messages: Array<{ id: string; type: string }> + } + expect(body.messages).toHaveLength(1) + expect(body.messages[0]!.type).toBe('user') + }) + + it('GET /api/teams/:name/members/:id/transcript should 404 for unknown member', async () => { + await writeTeamConfig('t2-team', makeTeamConfig({ name: 't2-team' })) + + const res = await fetch( + `${baseUrl}/api/teams/t2-team/members/unknown-agent/transcript`, + ) + expect(res.status).toBe(404) + }) + + it('DELETE /api/teams/:name should delete team', async () => { + const config = makeTeamConfig({ name: 'del-team' }) + for (const member of (config as { members: Array> }).members) { + member.isActive = false + } + await writeTeamConfig('del-team', config) + + const res = await fetch(`${baseUrl}/api/teams/del-team`, { + method: 'DELETE', + }) + expect(res.status).toBe(200) + + const body = (await res.json()) as { ok: boolean } + expect(body.ok).toBe(true) + + // Verify it's gone + const res2 = await fetch(`${baseUrl}/api/teams/del-team`) + expect(res2.status).toBe(404) + }) + + it('DELETE /api/teams/:name should 409 when team has active members', async () => { + await writeTeamConfig('active', makeTeamConfig({ name: 'active' })) + + const res = await fetch(`${baseUrl}/api/teams/active`, { + method: 'DELETE', + }) + expect(res.status).toBe(409) + }) + + it('POST /api/teams should return 405', async () => { + const res = await fetch(`${baseUrl}/api/teams`, { method: 'POST' }) + expect(res.status).toBe(405) + }) +}) diff --git a/src/server/api/agents.ts b/src/server/api/agents.ts index 5e9f514a..1916f1b2 100644 --- a/src/server/api/agents.ts +++ b/src/server/api/agents.ts @@ -7,11 +7,12 @@ * PUT /api/agents/:name — 更新 Agent * DELETE /api/agents/:name — 删除 Agent * - * GET /api/tasks — 获取后台任务列表 (placeholder) - * GET /api/tasks/:id — 获取任务详情 (placeholder) + * GET /api/tasks — 获取后台任务列表 + * GET /api/tasks/:id — 获取任务详情 */ import { AgentService } from '../services/agentService.js' +import { taskService } from '../services/taskService.js' import { ApiError, errorResponse } from '../middleware/errorHandler.js' const agentService = new AgentService() @@ -25,7 +26,7 @@ export async function handleAgentsApi( const resource = segments[1] // 'agents' | 'tasks' if (resource === 'tasks') { - return await handleTasksPlaceholder(req, segments) + return await handleTasksApi(req, segments) } return await handleAgents(req, segments) @@ -96,36 +97,34 @@ async function handleAgents( ) } -// ─── Tasks placeholder ────────────────────────────────────────────────────── +// ─── Tasks API ───────────────────────────────────────────────────────────── -async function handleTasksPlaceholder( +async function handleTasksApi( req: Request, segments: string[], ): Promise { - const method = req.method - const taskId = segments[2] - - if (method !== 'GET') { + if (req.method !== 'GET') { throw new ApiError( 405, - `Method ${method} not allowed on /api/tasks`, + `Method ${req.method} not allowed on /api/tasks`, 'METHOD_NOT_ALLOWED', ) } - // GET /api/tasks/:id + const taskId = segments[2] + if (taskId) { - return Response.json({ - task: { - id: taskId, - status: 'pending', - message: 'Background tasks are not yet implemented', - }, - }) + // GET /api/tasks/:id + const task = await taskService.getTask(taskId) + if (!task) { + throw ApiError.notFound(`Task not found: ${taskId}`) + } + return Response.json({ task }) } // GET /api/tasks - return Response.json({ tasks: [] }) + const tasks = await taskService.listTasks() + return Response.json({ tasks }) } // ─── Helpers ──────────────────────────────────────────────────────────────── diff --git a/src/server/api/teams.ts b/src/server/api/teams.ts new file mode 100644 index 00000000..1e2e5c8c --- /dev/null +++ b/src/server/api/teams.ts @@ -0,0 +1,61 @@ +/** + * Teams REST API + * + * GET /api/teams — 列出所有团队 + * GET /api/teams/:name — 获取团队详情 + * GET /api/teams/:name/members/:id/transcript — 获取成员 transcript + * DELETE /api/teams/:name — 删除团队 + */ + +import { teamService } from '../services/teamService.js' +import { ApiError, errorResponse } from '../middleware/errorHandler.js' + +export async function handleTeamsApi( + req: Request, + _url: URL, + segments: string[], +): Promise { + try { + const method = req.method + const teamName = segments[2] ? decodeURIComponent(segments[2]) : undefined + + // ── GET /api/teams ──────────────────────────────────────────────────── + if (method === 'GET' && !teamName) { + const teams = await teamService.listTeams() + return Response.json({ teams }) + } + + // ── GET /api/teams/:name/members/:id/transcript ─────────────────────── + if ( + method === 'GET' && + teamName && + segments[3] === 'members' && + segments[4] && + segments[5] === 'transcript' + ) { + const agentId = decodeURIComponent(segments[4]) + const messages = await teamService.getMemberTranscript(teamName, agentId) + return Response.json({ messages }) + } + + // ── GET /api/teams/:name ────────────────────────────────────────────── + if (method === 'GET' && teamName) { + const team = await teamService.getTeam(teamName) + return Response.json(team) + } + + // ── DELETE /api/teams/:name ─────────────────────────────────────────── + if (method === 'DELETE' && teamName) { + await teamService.deleteTeam(teamName) + return Response.json({ ok: true }) + } + + throw new ApiError( + 405, + `Method ${method} not allowed on /api/teams${teamName ? `/${teamName}` : ''}`, + 'METHOD_NOT_ALLOWED', + ) + } catch (error) { + return errorResponse(error) + } +} diff --git a/src/server/router.ts b/src/server/router.ts index 9a5601c1..5034a0f9 100644 --- a/src/server/router.ts +++ b/src/server/router.ts @@ -10,6 +10,7 @@ import { handleSearchApi } from './api/search.js' import { handleAgentsApi } from './api/agents.js' import { handleStatusApi } from './api/status.js' import { handleConversationsApi } from './api/conversations.js' +import { handleTeamsApi } from './api/teams.js' export async function handleApiRequest(req: Request, url: URL): Promise { const path = url.pathname @@ -54,6 +55,9 @@ export async function handleApiRequest(req: Request, url: URL): Promise + outputCallbacks: Array<(msg: any) => void> + workDir: string +} + +export class ConversationService { + private sessions = new Map() + + /** + * 启动 CLI 子进程用于对话。 + * 如果该 sessionId 已有活跃进程，直接返回。 + */ + async startSession(sessionId: string, workDir: string): Promise { + if (this.sessions.has(sessionId)) return + + // 找到 CLI 入口点 + const cliPath = path.resolve(import.meta.dir, '../../entrypoints/cli.tsx') + + const proc = Bun.spawn( + [ + 'bun', + cliPath, + '--print', + '--verbose', + '--input-format', + 'stream-json', + '--output-format', + 'stream-json', + '--session-id', + sessionId, + ], + { + cwd: workDir, + env: { ...process.env }, + stdin: 'pipe', + stdout: 'pipe', + stderr: 'pipe', + } + ) + + const session: SessionProcess = { + proc, + outputCallbacks: [], + workDir, + } + this.sessions.set(sessionId, session) + + // 读取 stdout（NDJSON 格式，每行一个 JSON 对象） + this.readOutputStream(sessionId, proc) + + // 读取 stderr 用于调试 + this.readErrorStream(sessionId, proc) + + // 进程退出时清理 + proc.exited.then((code) => { + console.log( + `[ConversationService] CLI process for ${sessionId} exited with code ${code}` + ) + this.sessions.delete(sessionId) + }) + + // 等待一小段时间，检测进程是否立即崩溃退出。 + // 使用 Promise.race 直接竞争 proc.exited，比检查 sessions map 更可靠。 + const STARTUP_GRACE_MS = 1000 + const earlyExitCode = await Promise.race([ + proc.exited, + new Promise((resolve) => setTimeout(() => resolve(null), STARTUP_GRACE_MS)), + ]) + + if (earlyExitCode !== null) { + // 进程在启动窗口内退出了，确保清理并抛出错误 + this.sessions.delete(sessionId) + throw new Error( + `CLI process for ${sessionId} exited immediately with code ${earlyExitCode}` + ) + } + } + + private async readOutputStream( + sessionId: string, + proc: ReturnType + ): Promise { + if (!proc.stdout) return + + const reader = (proc.stdout as ReadableStream).getReader() + const decoder = new TextDecoder() + let buffer = '' + + try { + while (true) { + const { done, value } = await reader.read() + if (done) break + + buffer += decoder.decode(value, { stream: true }) + const lines = buffer.split('\n') + buffer = lines.pop() || '' // 保留最后不完整的行 + + for (const line of lines) { + if (!line.trim()) continue + try { + const msg = JSON.parse(line) + const session = this.sessions.get(sessionId) + if (session) { + for (const cb of session.outputCallbacks) { + cb(msg) + } + } + } catch { + // 非 JSON 行忽略 + } + } + } + } catch (err) { + console.error( + `[ConversationService] stdout read error for ${sessionId}:`, + err + ) + } + } + + private async readErrorStream( + sessionId: string, + proc: ReturnType + ): Promise { + if (!proc.stderr) return + + const reader = (proc.stderr as ReadableStream).getReader() + const decoder = new TextDecoder() + try { + while (true) { + const { done, value } = await reader.read() + if (done) break + const text = decoder.decode(value, { stream: true }) + if (text.trim()) { + console.error(`[CLI:${sessionId}] ${text.trim()}`) + } + } + } catch { + // stderr 读取错误不影响主流程 + } + } + + /** + * 注册 stdout 消息回调。CLI 进程每输出一行 JSON， + * 所有已注册的回调都会被调用。 + */ + onOutput(sessionId: string, callback: (msg: any) => void): void { + const session = this.sessions.get(sessionId) + if (session) { + session.outputCallbacks.push(callback) + } + } + + /** + * 发送用户消息到 CLI stdin。 + * 消息格式遵循 stream-json 协议。 + */ + sendMessage(sessionId: string, content: string): boolean { + const session = this.sessions.get(sessionId) + if (!session || !session.proc.stdin) return false + + const msg = + JSON.stringify({ + type: 'user', + message: { + role: 'user', + content: [{ type: 'text', text: content }], + }, + parent_tool_use_id: null, + session_id: '', + }) + '\n' + + session.proc.stdin.write(msg) + return true + } + + /** + * 回复权限请求（用户在 UI 中点击允许/拒绝）。 + */ + respondToPermission( + sessionId: string, + requestId: string, + allowed: boolean + ): boolean { + const session = this.sessions.get(sessionId) + if (!session || !session.proc.stdin) return false + + const response = + JSON.stringify({ + type: 'control_response', + response: { + subtype: 'success', + request_id: requestId, + response: allowed + ? { behavior: 'allow' } + : { behavior: 'deny', message: 'User denied via UI' }, + }, + }) + '\n' + + session.proc.stdin.write(response) + return true + } + + /** + * 发送中断信号，停止当前生成。 + */ + sendInterrupt(sessionId: string): boolean { + const session = this.sessions.get(sessionId) + if (!session || !session.proc.stdin) return false + + const request = + JSON.stringify({ + type: 'control_request', + request_id: crypto.randomUUID(), + request: { subtype: 'interrupt' }, + }) + '\n' + + session.proc.stdin.write(request) + return true + } + + hasSession(sessionId: string): boolean { + return this.sessions.has(sessionId) + } + + stopSession(sessionId: string): void { + const session = this.sessions.get(sessionId) + if (session) { + session.proc.kill() + this.sessions.delete(sessionId) + } + } + + getActiveSessions(): string[] { + return Array.from(this.sessions.keys()) + } +} + +// 导出全局单例 +export const conversationService = new ConversationService() diff --git a/src/server/services/taskService.ts b/src/server/services/taskService.ts new file mode 100644 index 00000000..ccfbd696 --- /dev/null +++ b/src/server/services/taskService.ts @@ -0,0 +1,99 @@ +/** + * TaskService — 后台任务的读取与查询 + * + * 任务信息存储在 ~/.claude/tasks/ 目录下，每个任务一个 JSON 文件。 + * 支持嵌套子目录（例如按 team/project 分组）。 + */ + +import * as fs from 'fs/promises' +import * as path from 'path' +import * as os from 'os' + +export type TaskInfo = { + id: string + type: 'local_shell' | 'local_agent' | 'in_process_teammate' | 'remote_agent' | 'workflow' + status: 'running' | 'completed' | 'failed' | 'pending' + name?: string + description?: string + createdAt?: number + completedAt?: number + teamName?: string + agentName?: string + metadata?: Record +} + +export class TaskService { + private getConfigDir(): string { + return process.env.CLAUDE_CONFIG_DIR || path.join(os.homedir(), '.claude') + } + + private getTasksDir(): string { + return path.join(this.getConfigDir(), 'tasks') + } + + /** 列出所有持久化任务 */ + async listTasks(): Promise { + const tasksDir = this.getTasksDir() + try { + const result: TaskInfo[] = [] + await this.scanTasksRecursive(tasksDir, result) + return result.sort((a, b) => (b.createdAt || 0) - (a.createdAt || 0)) + } catch (err: any) { + if (err.code === 'ENOENT') return [] + throw err + } + } + + /** 递归扫描任务目录 */ + private async scanTasksRecursive(dir: string, result: TaskInfo[]): Promise { + let entries + try { + entries = await fs.readdir(dir, { withFileTypes: true }) + } catch { + return + } + + for (const entry of entries) { + const fullPath = path.join(dir, entry.name) + if (entry.isDirectory()) { + await this.scanTasksRecursive(fullPath, result) + } else if (entry.name.endsWith('.json')) { + try { + const raw = await fs.readFile(fullPath, 'utf-8') + const data = JSON.parse(raw) + const task = this.parseTaskFile(entry.name, data) + if (task) result.push(task) + } catch { + // 跳过无法解析的文件 + } + } + } + } + + /** 解析单个任务文件 */ + private parseTaskFile(filename: string, data: any): TaskInfo | null { + if (!data || typeof data !== 'object') return null + + const id = filename.replace('.json', '') + return { + id: data.id || id, + type: data.type || 'local_agent', + status: data.status || 'completed', + name: data.name || data.title, + description: data.description || data.prompt, + createdAt: data.createdAt, + completedAt: data.completedAt, + teamName: data.teamName, + agentName: data.agentName || data.name, + metadata: data.metadata, + } + } + + /** 获取单个任务详情 */ + async getTask(taskId: string): Promise { + const tasks = await this.listTasks() + return tasks.find(t => t.id === taskId) || null + } +} + +export const taskService = new TaskService() diff --git a/src/server/services/teamService.ts b/src/server/services/teamService.ts new file mode 100644 index 00000000..86d6dffd --- /dev/null +++ b/src/server/services/teamService.ts @@ -0,0 +1,296 @@ +/** + * TeamService — 读取 CLI 生成的 Agent Teams 配置 + * + * Team 配置存储在 ~/.claude/teams/{name}/config.json + * 成员 transcript 存储为 JSONL 文件，通过 sessionId 在 ~/.claude/projects/ 下定位。 + * 服务端只读取，不负责创建团队（由 CLI 的 TeamCreate 工具完成）。 + */ + +import * as fs from 'fs/promises' +import * as path from 'path' +import * as os from 'os' +import { ApiError } from '../middleware/errorHandler.js' + +// ─── Types ───────────────────────────────────────────────────────────────── + +export type TeamMember = { + agentId: string + name: string + agentType?: string + model?: string + color?: string + status: 'running' | 'completed' | 'idle' | 'failed' + joinedAt: number + cwd: string + sessionId?: string +} + +export type TeamSummary = { + name: string + description?: string + createdAt: number + memberCount: number + activeMemberCount: number +} + +export type TeamDetail = TeamSummary & { + leadAgentId: string + members: TeamMember[] +} + +export type TranscriptMessage = { + id: string + type: 'user' | 'assistant' | 'system' | 'tool_use' | 'tool_result' + content: unknown + timestamp: string +} + +/** Raw config.json structure written by CLI */ +type TeamFileRaw = { + name: string + description?: string + createdAt: number + leadAgentId: string + leadSessionId?: string + members: Array<{ + agentId: string + name: string + agentType?: string + model?: string + prompt?: string + color?: string + joinedAt: number + tmuxPaneId: string + cwd: string + worktreePath?: string + sessionId?: string + isActive?: boolean + mode?: string + }> +} + +// ─── Service ─────────────────────────────────────────────────────────────── + +export class TeamService { + private getConfigDir(): string { + return process.env.CLAUDE_CONFIG_DIR || path.join(os.homedir(), '.claude') + } + + private getTeamsDir(): string { + return path.join(this.getConfigDir(), 'teams') + } + + private getProjectsDir(): string { + return path.join(this.getConfigDir(), 'projects') + } + + // ── List all teams ────────────────────────────────────────────────────── + + async listTeams(): Promise { + const teamsDir = this.getTeamsDir() + + try { + await fs.access(teamsDir) + } catch { + return [] + } + + const entries = await fs.readdir(teamsDir, { withFileTypes: true }) + const teams: TeamSummary[] = [] + + for (const entry of entries) { + if (!entry.isDirectory()) continue + + try { + const config = await this.loadTeamConfig(entry.name) + teams.push(this.toSummary(config)) + } catch { + // Skip malformed team directories + } + } + + return teams + } + + // ── Get team detail ───────────────────────────────────────────────────── + + async getTeam(name: string): Promise { + const config = await this.loadTeamConfig(name) + + const members: TeamMember[] = config.members.map((m) => ({ + agentId: m.agentId, + name: m.name, + agentType: m.agentType, + model: m.model, + color: m.color, + status: this.deriveStatus(m.isActive), + joinedAt: m.joinedAt, + cwd: m.cwd, + sessionId: m.sessionId, + })) + + return { + ...this.toSummary(config), + leadAgentId: config.leadAgentId, + members, + } + } + + // ── Get member transcript ─────────────────────────────────────────────── + + async getMemberTranscript( + teamName: string, + agentId: string, + ): Promise { + const config = await this.loadTeamConfig(teamName) + + const member = config.members.find((m) => m.agentId === agentId) + if (!member) { + throw ApiError.notFound( + `Member not found: ${agentId} in team ${teamName}`, + ) + } + + if (!member.sessionId) { + return [] + } + + const jsonlPath = await this.findTranscriptFile(member.sessionId) + if (!jsonlPath) { + return [] + } + + return this.parseTranscriptFile(jsonlPath) + } + + // ── Delete team ───────────────────────────────────────────────────────── + + async deleteTeam(name: string): Promise { + const config = await this.loadTeamConfig(name) + + const hasActive = config.members.some( + (m) => m.isActive === undefined || m.isActive === true, + ) + if (hasActive) { + throw ApiError.conflict( + `Cannot delete team "${name}": has active members`, + ) + } + + const teamDir = path.join(this.getTeamsDir(), name) + await fs.rm(teamDir, { recursive: true, force: true }) + } + + // ── Internal helpers ──────────────────────────────────────────────────── + + private async loadTeamConfig(name: string): Promise { + const configPath = path.join(this.getTeamsDir(), name, 'config.json') + + try { + const raw = await fs.readFile(configPath, 'utf-8') + return JSON.parse(raw) as TeamFileRaw + } catch { + throw ApiError.notFound(`Team not found: ${name}`) + } + } + + private toSummary(config: TeamFileRaw): TeamSummary { + const activeMemberCount = config.members.filter( + (m) => m.isActive === undefined || m.isActive === true, + ).length + + return { + name: config.name, + description: config.description, + createdAt: config.createdAt, + memberCount: config.members.length, + activeMemberCount, + } + } + + private deriveStatus( + isActive: boolean | undefined, + ): 'running' | 'completed' | 'idle' | 'failed' { + if (isActive === false) return 'idle' + // isActive === undefined || isActive === true + return 'running' + } + + /** Search ~/.claude/projects/ for a JSONL file matching the sessionId. */ + private async findTranscriptFile( + sessionId: string, + ): Promise { + const projectsDir = this.getProjectsDir() + + try { + await fs.access(projectsDir) + } catch { + return null + } + + const projectEntries = await fs.readdir(projectsDir, { + withFileTypes: true, + }) + + for (const entry of projectEntries) { + if (!entry.isDirectory()) continue + + const candidate = path.join(projectsDir, entry.name, `${sessionId}.jsonl`) + try { + await fs.access(candidate) + return candidate + } catch { + // Not in this project directory + } + } + + return null + } + + /** Parse a JSONL transcript file into messages. */ + private async parseTranscriptFile( + filePath: string, + ): Promise { + const raw = await fs.readFile(filePath, 'utf-8') + const lines = raw.split('\n').filter((line) => line.trim().length > 0) + + const messages: TranscriptMessage[] = [] + + for (const line of lines) { + try { + const entry = JSON.parse(line) as Record + + // Skip non-message entries (snapshots, meta, etc.) + const entryType = entry.type as string | undefined + if ( + entryType !== 'user' && + entryType !== 'assistant' && + entryType !== 'system' && + entryType !== 'tool_use' && + entryType !== 'tool_result' + ) { + continue + } + + // Skip meta entries + if (entry.isMeta) continue + + const message: TranscriptMessage = { + id: (entry.uuid as string) || crypto.randomUUID(), + type: entryType as TranscriptMessage['type'], + content: entry.message ?? entry.content ?? null, + timestamp: + (entry.timestamp as string) || new Date().toISOString(), + } + + messages.push(message) + } catch { + // Skip unparseable lines + } + } + + return messages + } +} + +export const teamService = new TeamService() diff --git a/src/server/ws/handler.ts b/src/server/ws/handler.ts index cbc5136f..60b2af41 100644 --- a/src/server/ws/handler.ts +++ b/src/server/ws/handler.ts @@ -2,10 +2,13 @@ * WebSocket connection handler * * 管理 WebSocket 连接生命周期，处理消息路由。 + * 用户消息通过 CLI 子进程（stream-json 模式）处理， + * CLI stdout 消息被转换为 ServerMessage 并转发到 WebSocket。 */ import type { ServerWebSocket } from 'bun' import type { ClientMessage, ServerMessage, WebSocketSession } from './events.js' +import { conversationService } from '../services/conversationService.js' export type WebSocketData = { sessionId: string @@ -34,7 +37,9 @@ export const handleWebSocket = { switch (message.type) { case 'user_message': - handleUserMessage(ws, message) + handleUserMessage(ws, message).catch((err) => { + console.error(`[WS] Unhandled error in handleUserMessage:`, err) + }) break case 'permission_response': @@ -61,6 +66,11 @@ export const handleWebSocket = { const { sessionId } = ws.data console.log(`[WS] Client disconnected from session: ${sessionId} (${code}: ${reason})`) activeSessions.delete(sessionId) + + // 断开连接时停止对应的 CLI 子进程 + if (conversationService.hasSession(sessionId)) { + conversationService.stopSession(sessionId) + } }, drain(ws: ServerWebSocket) { @@ -69,7 +79,7 @@ export const handleWebSocket = { } // ============================================================================ -// Message handlers (to be implemented by conversationService) +// Message handlers // ============================================================================ async function handleUserMessage( @@ -81,31 +91,140 @@ async function handleUserMessage( // Send thinking status sendMessage(ws, { type: 'status', state: 'thinking', verb: 'Thinking' }) - // TODO: Integrate with ConversationService to process the message - // For now, echo back a placeholder response - sendMessage(ws, { type: 'content_start', blockType: 'text' }) - sendMessage(ws, { - type: 'content_delta', - text: `[Server] Received message for session ${sessionId}: "${message.content}". Chat integration pending.`, - }) - sendMessage(ws, { - type: 'message_complete', - usage: { input_tokens: 0, output_tokens: 0 }, - }) - sendMessage(ws, { type: 'status', state: 'idle' }) + // 启动 CLI 子进程（如果还没有） + if (!conversationService.hasSession(sessionId)) { + try { + const workDir = process.cwd() + await conversationService.startSession(sessionId, workDir) + + // 注册 CLI stdout → WebSocket 转发 + conversationService.onOutput(sessionId, (cliMsg) => { + const serverMsg = translateCliMessage(cliMsg) + if (serverMsg) { + sendMessage(ws, serverMsg) + } + }) + } catch (err) { + console.error(`[WS] CLI start failed for ${sessionId}, falling back to echo`) + // CLI 启动失败时回退到 echo 模式，保证基本可用性 + sendFallbackEcho(ws, sessionId, message.content) + return + } + } + + // 将用户消息写入 CLI stdin + const sent = conversationService.sendMessage(sessionId, message.content) + if (!sent) { + // 消息发送失败（进程可能已退出），回退到 echo 模式 + sendFallbackEcho(ws, sessionId, message.content) + } } function handlePermissionResponse( ws: ServerWebSocket, message: Extract ) { - // TODO: Forward permission response to the active tool execution + const { sessionId } = ws.data + conversationService.respondToPermission(sessionId, message.requestId, message.allowed) console.log(`[WS] Permission response for ${message.requestId}: ${message.allowed}`) } function handleStopGeneration(ws: ServerWebSocket) { - // TODO: Abort the active generation for this session - console.log(`[WS] Stop generation requested for session: ${ws.data.sessionId}`) + const { sessionId } = ws.data + console.log(`[WS] Stop generation requested for session: ${sessionId}`) + + // 向 CLI 子进程发送中断信号 + if (conversationService.hasSession(sessionId)) { + conversationService.sendInterrupt(sessionId) + } + + sendMessage(ws, { type: 'status', state: 'idle' }) +} + +// ============================================================================ +// CLI message translation +// ============================================================================ + +/** + * 将 CLI stdout 的 stream-json 消息转换为 WebSocket ServerMessage。 + * + * CLI 输出格式参考 src/bridge/sessionRunner.ts 中的 stream-json 协议。 + */ +function translateCliMessage(cliMsg: any): ServerMessage | null { + switch (cliMsg.type) { + case 'assistant': { + // 助手消息 - 提取文本内容 + if (cliMsg.message?.content) { + const content = cliMsg.message.content + if (Array.isArray(content)) { + for (const block of content) { + if (block.type === 'text') { + return { type: 'content_delta', text: block.text } + } + } + } + } + return null + } + + case 'control_request': { + // 权限请求 — CLI 需要用户授权才能执行工具 + if (cliMsg.request?.subtype === 'can_use_tool') { + return { + type: 'permission_request', + requestId: cliMsg.request_id, + toolName: cliMsg.request.tool_name || 'Unknown', + input: cliMsg.request.input || {}, + description: cliMsg.request.description, + } + } + return null + } + + case 'result': { + // 对话完成 + if (cliMsg.subtype === 'success') { + return { + type: 'message_complete', + usage: { + input_tokens: cliMsg.usage?.input_tokens || 0, + output_tokens: cliMsg.usage?.output_tokens || 0, + }, + } + } + return null + } + + case 'system': + return { type: 'status', state: 'idle' } + + default: + return null + } +} + +// ============================================================================ +// Fallback echo (for when CLI is not available) +// ============================================================================ + +/** + * 当 CLI 子进程启动失败或不可用时，使用 echo 模式作为回退。 + * 保证 WebSocket 客户端始终能收到完整的消息流转。 + */ +function sendFallbackEcho( + ws: ServerWebSocket, + sessionId: string, + content: string +) { + sendMessage(ws, { type: 'content_start', blockType: 'text' }) + sendMessage(ws, { + type: 'content_delta', + text: `[Server] Received message for session ${sessionId}: "${content}". Chat integration pending.`, + }) + sendMessage(ws, { + type: 'message_complete', + usage: { input_tokens: 0, output_tokens: 0 }, + }) sendMessage(ws, { type: 'status', state: 'idle' }) }