diff --git a/.env.example b/.env.example index a28d1a43..2eec4b42 100644 --- a/.env.example +++ b/.env.example @@ -1,3 +1,50 @@ -ANTHROPIC_API_KEY=sk-ant-your-key-here -SERVER_PORT=3456 -SERVER_HOST=127.0.0.1 +# ============================================================ +# MiniMax(直连 Anthropic 兼容接口) +# ============================================================ +# ANTHROPIC_AUTH_TOKEN=your_token_here +# ANTHROPIC_BASE_URL=https://api.minimaxi.com/anthropic +# ANTHROPIC_MODEL=MiniMax-M2.7-highspeed +# ANTHROPIC_DEFAULT_SONNET_MODEL=MiniMax-M2.7-highspeed +# ANTHROPIC_DEFAULT_HAIKU_MODEL=MiniMax-M2.7-highspeed +# ANTHROPIC_DEFAULT_OPUS_MODEL=MiniMax-M2.7-highspeed +# API_TIMEOUT_MS=3000000 + +# ============================================================ +# OpenAI(通过 LiteLLM 代理) +# 先启动: litellm --config litellm_config.yaml --port 4000 +# ============================================================ +# ANTHROPIC_AUTH_TOKEN=sk-anything +# ANTHROPIC_BASE_URL=http://localhost:4000 +# ANTHROPIC_MODEL=gpt-4o +# ANTHROPIC_DEFAULT_SONNET_MODEL=gpt-4o +# ANTHROPIC_DEFAULT_HAIKU_MODEL=gpt-4o +# ANTHROPIC_DEFAULT_OPUS_MODEL=gpt-4o +# API_TIMEOUT_MS=3000000 + +# ============================================================ +# DeepSeek(通过 LiteLLM 代理) +# 先启动: litellm --config litellm_config.yaml --port 4000 +# ============================================================ +# ANTHROPIC_AUTH_TOKEN=sk-anything +# ANTHROPIC_BASE_URL=http://localhost:4000 +# ANTHROPIC_MODEL=deepseek-chat +# ANTHROPIC_DEFAULT_SONNET_MODEL=deepseek-chat +# ANTHROPIC_DEFAULT_HAIKU_MODEL=deepseek-chat +# ANTHROPIC_DEFAULT_OPUS_MODEL=deepseek-chat +# API_TIMEOUT_MS=3000000 + +# ============================================================ +# OpenRouter(直连 Anthropic 兼容接口) +# ============================================================ +# ANTHROPIC_AUTH_TOKEN=sk-or-v1-xxx +# ANTHROPIC_BASE_URL=https://openrouter.ai/api/v1 +# ANTHROPIC_MODEL=openai/gpt-4o +# ANTHROPIC_DEFAULT_SONNET_MODEL=openai/gpt-4o +# ANTHROPIC_DEFAULT_HAIKU_MODEL=openai/gpt-4o-mini +# ANTHROPIC_DEFAULT_OPUS_MODEL=openai/gpt-4o + +# ============================================================ +# 通用设置(建议始终开启) +# ============================================================ +DISABLE_TELEMETRY=1 +CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC=1 \ No newline at end of file diff --git a/src/server/__tests__/cron-scheduler.test.ts b/src/server/__tests__/cron-scheduler.test.ts new file mode 100644 index 00000000..f5733639 --- /dev/null +++ b/src/server/__tests__/cron-scheduler.test.ts @@ -0,0 +1,575 @@ +/** + * Tests for CronScheduler — cron matching, task execution, log storage, and API endpoints + */ + +import { describe, it, expect, beforeEach, afterEach, mock, spyOn } from 'bun:test' +import * as fs from 'fs/promises' +import * as path from 'path' +import * as os from 'os' +import { + cronMatches, + fieldMatches, + CronScheduler, + type TaskRun, +} from '../services/cronScheduler.js' +import { CronService, type CronTask } from '../services/cronService.js' + +// ─── Test helpers ─────────────────────────────────────────────────────────── + +let tmpDir: string +const originalConfigDir = process.env.CLAUDE_CONFIG_DIR + +async function createTmpDir(): Promise { + const dir = path.join( + os.tmpdir(), + `claude-test-${Date.now()}-${Math.random().toString(36).slice(2)}`, + ) + await fs.mkdir(dir, { recursive: true }) + return dir +} + +async function cleanupTmpDir(dir: string): Promise { + try { + await fs.rm(dir, { recursive: true, force: true }) + } catch { + // ignore + } +} + +// ─── fieldMatches tests ──────────────────────────────────────────────────── + +describe('fieldMatches', () => { + it('should match wildcard', () => { + expect(fieldMatches('*', 0)).toBe(true) + expect(fieldMatches('*', 59)).toBe(true) + expect(fieldMatches('*', 23)).toBe(true) + }) + + it('should match exact number', () => { + expect(fieldMatches('5', 5)).toBe(true) + expect(fieldMatches('5', 6)).toBe(false) + expect(fieldMatches('0', 0)).toBe(true) + expect(fieldMatches('30', 30)).toBe(true) + }) + + it('should match comma-separated list', () => { + expect(fieldMatches('1,3,5', 1)).toBe(true) + expect(fieldMatches('1,3,5', 3)).toBe(true) + expect(fieldMatches('1,3,5', 5)).toBe(true) + expect(fieldMatches('1,3,5', 2)).toBe(false) + expect(fieldMatches('1,3,5', 4)).toBe(false) + }) + + it('should match range', () => { + expect(fieldMatches('1-5', 1)).toBe(true) + expect(fieldMatches('1-5', 3)).toBe(true) + expect(fieldMatches('1-5', 5)).toBe(true) + expect(fieldMatches('1-5', 0)).toBe(false) + expect(fieldMatches('1-5', 6)).toBe(false) + }) + + it('should match step from wildcard', () => { + expect(fieldMatches('*/2', 0)).toBe(true) + expect(fieldMatches('*/2', 2)).toBe(true) + expect(fieldMatches('*/2', 4)).toBe(true) + expect(fieldMatches('*/2', 1)).toBe(false) + expect(fieldMatches('*/2', 3)).toBe(false) + expect(fieldMatches('*/15', 0)).toBe(true) + expect(fieldMatches('*/15', 15)).toBe(true) + expect(fieldMatches('*/15', 30)).toBe(true) + expect(fieldMatches('*/15', 7)).toBe(false) + }) + + it('should match step within range', () => { + expect(fieldMatches('1-10/3', 1)).toBe(true) + expect(fieldMatches('1-10/3', 4)).toBe(true) + expect(fieldMatches('1-10/3', 7)).toBe(true) + expect(fieldMatches('1-10/3', 10)).toBe(true) + expect(fieldMatches('1-10/3', 2)).toBe(false) + expect(fieldMatches('1-10/3', 11)).toBe(false) + expect(fieldMatches('1-10/3', 0)).toBe(false) + }) + + it('should handle combined comma and range', () => { + expect(fieldMatches('1-3,7,10-12', 2)).toBe(true) + expect(fieldMatches('1-3,7,10-12', 7)).toBe(true) + expect(fieldMatches('1-3,7,10-12', 11)).toBe(true) + expect(fieldMatches('1-3,7,10-12', 5)).toBe(false) + }) +}) + +// ─── cronMatches tests ───────────────────────────────────────────────────── + +describe('cronMatches', () => { + it('should match every-minute expression', () => { + const date = new Date(2026, 3, 5, 14, 30, 0) // April 5, 2026 14:30 (Sunday) + expect(cronMatches('* * * * *', date)).toBe(true) + }) + + it('should match daily at 9:00', () => { + const match = new Date(2026, 3, 5, 9, 0, 0) + const noMatch = new Date(2026, 3, 5, 9, 1, 0) + expect(cronMatches('0 9 * * *', match)).toBe(true) + expect(cronMatches('0 9 * * *', noMatch)).toBe(false) + }) + + it('should match every 2 hours at minute 0', () => { + expect(cronMatches('0 */2 * * *', new Date(2026, 0, 1, 0, 0))).toBe(true) + expect(cronMatches('0 */2 * * *', new Date(2026, 0, 1, 2, 0))).toBe(true) + expect(cronMatches('0 */2 * * *', new Date(2026, 0, 1, 4, 0))).toBe(true) + expect(cronMatches('0 */2 * * *', new Date(2026, 0, 1, 1, 0))).toBe(false) + expect(cronMatches('0 */2 * * *', new Date(2026, 0, 1, 3, 0))).toBe(false) + }) + + it('should match weekdays at 14:30', () => { + // April 6, 2026 is a Monday (dow = 1) + const monday = new Date(2026, 3, 6, 14, 30, 0) + // April 5, 2026 is a Sunday (dow = 0) + const sunday = new Date(2026, 3, 5, 14, 30, 0) + expect(cronMatches('30 14 * * 1-5', monday)).toBe(true) + expect(cronMatches('30 14 * * 1-5', sunday)).toBe(false) + }) + + it('should match specific month and day', () => { + // January 15 at midnight + const jan15 = new Date(2026, 0, 15, 0, 0) + const feb15 = new Date(2026, 1, 15, 0, 0) + expect(cronMatches('0 0 15 1 *', jan15)).toBe(true) + expect(cronMatches('0 0 15 1 *', feb15)).toBe(false) + }) + + it('should reject invalid cron expressions', () => { + const date = new Date() + expect(cronMatches('* * *', date)).toBe(false) // only 3 fields + expect(cronMatches('', date)).toBe(false) + expect(cronMatches('* * * * * *', date)).toBe(false) // 6 fields + }) + + it('should match day-of-week with Sunday as 0', () => { + // Sunday = 0 + const sunday = new Date(2026, 3, 5, 10, 0) // April 5, 2026 is Sunday + expect(cronMatches('0 10 * * 0', sunday)).toBe(true) + expect(cronMatches('0 10 * * 6', sunday)).toBe(false) + }) +}) + +// ─── CronScheduler execution tests ──────────────────────────────────────── + +describe('CronScheduler', () => { + let cronService: CronService + let scheduler: CronScheduler + + beforeEach(async () => { + tmpDir = await createTmpDir() + process.env.CLAUDE_CONFIG_DIR = tmpDir + cronService = new CronService() + scheduler = new CronScheduler(cronService) + }) + + afterEach(async () => { + scheduler.stop() + if (originalConfigDir) { + process.env.CLAUDE_CONFIG_DIR = originalConfigDir + } else { + delete process.env.CLAUDE_CONFIG_DIR + } + await cleanupTmpDir(tmpDir) + }) + + it('should start and stop without errors', () => { + scheduler.start() + scheduler.stop() + // Starting again after stop should also work + scheduler.start() + scheduler.stop() + }) + + it('should not start twice', () => { + scheduler.start() + // Second start should be a no-op (no error) + scheduler.start() + scheduler.stop() + }) + + it('should return empty runs when no tasks have executed', async () => { + const runs = await scheduler.getRecentRuns() + expect(runs).toEqual([]) + }) + + it('should return empty runs for a non-existent task ID', async () => { + const runs = await scheduler.getTaskRuns('nonexistent') + expect(runs).toEqual([]) + }) + + it('should persist a task run to the log file', async () => { + // Create a task that runs "echo hello" — we'll invoke executeTask directly + // with a mock-like approach: create a task then check the log file + const task = await cronService.createTask({ + cron: '* * * * *', + prompt: 'echo test', + name: 'Test Task', + recurring: true, + }) + + // We can't easily mock Bun.spawn in bun:test, so we'll check the log + // file was created by reading it after execution attempt. + // The CLI subprocess will likely fail (not a real CLI available in tests), + // but the run should still be logged with 'failed' status. + try { + await scheduler.executeTask(task) + } catch { + // Expected — CLI binary may not be available in test environment + } + + const logPath = path.join(tmpDir, 'scheduled_tasks_log.json') + const logExists = await fs + .stat(logPath) + .then(() => true) + .catch(() => false) + expect(logExists).toBe(true) + + const logContent = JSON.parse(await fs.readFile(logPath, 'utf-8')) as { + runs: TaskRun[] + } + expect(logContent.runs.length).toBeGreaterThanOrEqual(1) + expect(logContent.runs[0].taskId).toBe(task.id) + expect(logContent.runs[0].taskName).toBe('Test Task') + expect(logContent.runs[0].prompt).toBe('echo test') + }) + + it('should disable non-recurring task after execution', async () => { + const task = await cronService.createTask({ + cron: '* * * * *', + prompt: 'one-shot task', + recurring: false, + }) + + try { + await scheduler.executeTask(task) + } catch { + // CLI may not be available + } + + // After execution, the task should be disabled + const tasks = await cronService.listTasks() + const updated = tasks.find((t) => t.id === task.id) + expect(updated?.enabled).toBe(false) + }) + + it('should NOT disable recurring task after execution', async () => { + const task = await cronService.createTask({ + cron: '* * * * *', + prompt: 'recurring task', + recurring: true, + }) + + try { + await scheduler.executeTask(task) + } catch { + // CLI may not be available + } + + const tasks = await cronService.listTasks() + const updated = tasks.find((t) => t.id === task.id) + // enabled should not have been set to false + expect(updated?.enabled).not.toBe(false) + }) + + it('should update lastFiredAt after execution', async () => { + const task = await cronService.createTask({ + cron: '* * * * *', + prompt: 'fire test', + recurring: true, + }) + + const beforeExec = new Date().toISOString() + + try { + await scheduler.executeTask(task) + } catch { + // CLI may not be available + } + + const tasks = await cronService.listTasks() + const updated = tasks.find((t) => t.id === task.id) + expect(updated?.lastFiredAt).toBeDefined() + // lastFiredAt should be a valid ISO timestamp at or after beforeExec + expect(new Date(updated!.lastFiredAt!).getTime()).toBeGreaterThanOrEqual( + new Date(beforeExec).getTime() - 1000, // allow 1s tolerance + ) + }) + + it('should skip disabled tasks during tick', async () => { + // Create a task matching every minute but disabled + const task = await cronService.createTask({ + cron: '* * * * *', + prompt: 'should not run', + enabled: false, + recurring: true, + }) + + await scheduler.tick() + + // No runs should be logged + const runs = await scheduler.getTaskRuns(task.id) + expect(runs).toHaveLength(0) + }) + + it('getTaskRuns should return runs sorted newest first', async () => { + const task = await cronService.createTask({ + cron: '* * * * *', + prompt: 'multi run', + recurring: true, + }) + + // Execute twice + try { + await scheduler.executeTask(task) + } catch { + /* ignore */ + } + try { + await scheduler.executeTask(task) + } catch { + /* ignore */ + } + + const runs = await scheduler.getTaskRuns(task.id) + expect(runs.length).toBeGreaterThanOrEqual(2) + // Should be sorted newest first + if (runs.length >= 2) { + expect( + new Date(runs[0].startedAt).getTime(), + ).toBeGreaterThanOrEqual(new Date(runs[1].startedAt).getTime()) + } + }) + + it('getRecentRuns should respect limit parameter', async () => { + const task = await cronService.createTask({ + cron: '* * * * *', + prompt: 'limit test', + recurring: true, + }) + + // Execute 3 times + for (let i = 0; i < 3; i++) { + try { + await scheduler.executeTask(task) + } catch { + /* ignore */ + } + } + + const runs = await scheduler.getRecentRuns(2) + expect(runs.length).toBeLessThanOrEqual(2) + }) +}) + +// ─── Execution log trimming ──────────────────────────────────────────────── + +describe('Execution log trimming', () => { + let cronService: CronService + let scheduler: CronScheduler + + beforeEach(async () => { + tmpDir = await createTmpDir() + process.env.CLAUDE_CONFIG_DIR = tmpDir + cronService = new CronService() + scheduler = new CronScheduler(cronService) + }) + + afterEach(async () => { + scheduler.stop() + if (originalConfigDir) { + process.env.CLAUDE_CONFIG_DIR = originalConfigDir + } else { + delete process.env.CLAUDE_CONFIG_DIR + } + await cleanupTmpDir(tmpDir) + }) + + it('should keep log entries within the max limit', async () => { + // Pre-populate the log file with 105 entries for a single task + const logPath = path.join(tmpDir, 'scheduled_tasks_log.json') + const runs: TaskRun[] = [] + for (let i = 0; i < 105; i++) { + runs.push({ + id: `run-${i}`, + taskId: 'task-1', + taskName: 'Test', + startedAt: new Date(Date.now() - (105 - i) * 1000).toISOString(), + completedAt: new Date(Date.now() - (105 - i) * 1000 + 100).toISOString(), + status: 'completed', + prompt: 'test', + exitCode: 0, + durationMs: 100, + }) + } + await fs.writeFile(logPath, JSON.stringify({ runs }, null, 2), 'utf-8') + + // Now execute one more task run — this triggers a trim + const task = await cronService.createTask({ + cron: '* * * * *', + prompt: 'trigger trim', + recurring: true, + }) + + try { + await scheduler.executeTask(task) + } catch { + /* ignore */ + } + + // Read back the log + const logContent = JSON.parse(await fs.readFile(logPath, 'utf-8')) as { + runs: TaskRun[] + } + const task1Runs = logContent.runs.filter((r) => r.taskId === 'task-1') + // Should have been trimmed to at most 100 + expect(task1Runs.length).toBeLessThanOrEqual(100) + }) +}) + +// ─── Scheduled Tasks API with runs endpoints ────────────────────────────── + +describe('Scheduled Tasks API — runs endpoints', () => { + let handleScheduledTasksApi: ( + req: Request, + url: URL, + segments: string[], + ) => Promise + + beforeEach(async () => { + tmpDir = await createTmpDir() + process.env.CLAUDE_CONFIG_DIR = tmpDir + + const mod = await import('../api/scheduled-tasks.js') + handleScheduledTasksApi = mod.handleScheduledTasksApi + }) + + afterEach(async () => { + if (originalConfigDir) { + process.env.CLAUDE_CONFIG_DIR = originalConfigDir + } else { + delete process.env.CLAUDE_CONFIG_DIR + } + await cleanupTmpDir(tmpDir) + }) + + it('GET /api/scheduled-tasks/runs should return empty runs', async () => { + const req = new Request('http://localhost/api/scheduled-tasks/runs', { + method: 'GET', + }) + const url = new URL(req.url) + const resp = await handleScheduledTasksApi(req, url, [ + 'api', + 'scheduled-tasks', + 'runs', + ]) + const body = (await resp.json()) as { runs: unknown[] } + expect(resp.status).toBe(200) + expect(body.runs).toEqual([]) + }) + + it('GET /api/scheduled-tasks/:id/runs should return empty runs for a task', async () => { + const req = new Request( + 'http://localhost/api/scheduled-tasks/abc123/runs', + { method: 'GET' }, + ) + const url = new URL(req.url) + const resp = await handleScheduledTasksApi(req, url, [ + 'api', + 'scheduled-tasks', + 'abc123', + 'runs', + ]) + const body = (await resp.json()) as { runs: unknown[] } + expect(resp.status).toBe(200) + expect(body.runs).toEqual([]) + }) + + it('GET /api/scheduled-tasks/runs should return runs from log', async () => { + // Write some runs to the log file + const logPath = path.join(tmpDir, 'scheduled_tasks_log.json') + const runs: TaskRun[] = [ + { + id: 'run-1', + taskId: 'task-a', + taskName: 'Task A', + startedAt: new Date().toISOString(), + completedAt: new Date().toISOString(), + status: 'completed', + prompt: 'test prompt', + exitCode: 0, + durationMs: 500, + }, + { + id: 'run-2', + taskId: 'task-b', + taskName: 'Task B', + startedAt: new Date().toISOString(), + completedAt: new Date().toISOString(), + status: 'failed', + prompt: 'another prompt', + error: 'some error', + exitCode: 1, + durationMs: 200, + }, + ] + await fs.writeFile(logPath, JSON.stringify({ runs }, null, 2), 'utf-8') + + const req = new Request('http://localhost/api/scheduled-tasks/runs', { + method: 'GET', + }) + const url = new URL(req.url) + const resp = await handleScheduledTasksApi(req, url, [ + 'api', + 'scheduled-tasks', + 'runs', + ]) + const body = (await resp.json()) as { runs: TaskRun[] } + expect(resp.status).toBe(200) + expect(body.runs).toHaveLength(2) + }) + + it('GET /api/scheduled-tasks/:id/runs should filter by task ID', async () => { + const logPath = path.join(tmpDir, 'scheduled_tasks_log.json') + const runs: TaskRun[] = [ + { + id: 'run-1', + taskId: 'task-a', + taskName: 'Task A', + startedAt: new Date().toISOString(), + status: 'completed', + prompt: 'prompt a', + exitCode: 0, + }, + { + id: 'run-2', + taskId: 'task-b', + taskName: 'Task B', + startedAt: new Date().toISOString(), + status: 'completed', + prompt: 'prompt b', + exitCode: 0, + }, + ] + await fs.writeFile(logPath, JSON.stringify({ runs }, null, 2), 'utf-8') + + const req = new Request( + 'http://localhost/api/scheduled-tasks/task-a/runs', + { method: 'GET' }, + ) + const url = new URL(req.url) + const resp = await handleScheduledTasksApi(req, url, [ + 'api', + 'scheduled-tasks', + 'task-a', + 'runs', + ]) + const body = (await resp.json()) as { runs: TaskRun[] } + expect(resp.status).toBe(200) + expect(body.runs).toHaveLength(1) + expect(body.runs[0].taskId).toBe('task-a') + }) +}) diff --git a/src/server/__tests__/real-llm-test.ts b/src/server/__tests__/real-llm-test.ts new file mode 100644 index 00000000..0a8391f3 --- /dev/null +++ b/src/server/__tests__/real-llm-test.ts @@ -0,0 +1,526 @@ +/** + * Real LLM Integration Test + * + * 真实调用 MiniMax API,验证完整的 WebSocket 对话流: + * Server → CLI subprocess → MiniMax API → streaming response → WebSocket → client + * + * 使用 .env 中的 MiniMax 配置。 + */ + +const SERVER_PORT = 19876 +const BASE_URL = `http://127.0.0.1:${SERVER_PORT}` +const WS_URL = `ws://127.0.0.1:${SERVER_PORT}` + +// Generate a valid UUID for session ID (CLI requires UUID format) +function uuid(): string { + return crypto.randomUUID() +} + +async function sleep(ms: number) { + return new Promise((resolve) => setTimeout(resolve, ms)) +} + +// ─── Test helpers ────────────────────────────────────────────────────────── + +type ServerMsg = { + type: string + [key: string]: any +} + +function createWebSocket(sessionId: string): Promise<{ + ws: WebSocket + messages: ServerMsg[] + waitForType: (type: string, timeoutMs?: number) => Promise + waitForAny: (types: string[], timeoutMs?: number) => Promise + close: () => void +}> { + return new Promise((resolve, reject) => { + const messages: ServerMsg[] = [] + const waiters: Array<{ + types: string[] + resolve: (msg: ServerMsg) => void + reject: (err: Error) => void + }> = [] + + const ws = new WebSocket(`${WS_URL}/ws/${sessionId}`) + + ws.onmessage = (event) => { + try { + const msg = JSON.parse(event.data as string) as ServerMsg + messages.push(msg) + // Check waiters + for (let i = waiters.length - 1; i >= 0; i--) { + if (waiters[i].types.includes(msg.type)) { + waiters[i].resolve(msg) + waiters.splice(i, 1) + } + } + } catch (e) { + console.error('Failed to parse WS message:', event.data) + } + } + + ws.onerror = (event) => { + reject(new Error(`WebSocket error`)) + } + + ws.onopen = () => { + resolve({ + ws, + messages, + waitForType(type: string, timeoutMs = 60000) { + // Check existing messages first + const existing = messages.find((m) => m.type === type) + if (existing) return Promise.resolve(existing) + + return new Promise((res, rej) => { + const timer = setTimeout(() => { + rej( + new Error( + `Timeout waiting for message type "${type}" after ${timeoutMs}ms. Got: ${messages.map((m) => m.type).join(', ')}` + ) + ) + }, timeoutMs) + waiters.push({ + types: [type], + resolve: (msg) => { + clearTimeout(timer) + res(msg) + }, + reject: rej, + }) + }) + }, + waitForAny(types: string[], timeoutMs = 60000) { + const existing = messages.find((m) => types.includes(m.type)) + if (existing) return Promise.resolve(existing) + + return new Promise((res, rej) => { + const timer = setTimeout(() => { + rej( + new Error( + `Timeout waiting for any of [${types.join(', ')}] after ${timeoutMs}ms. Got: ${messages.map((m) => m.type).join(', ')}` + ) + ) + }, timeoutMs) + waiters.push({ + types, + resolve: (msg) => { + clearTimeout(timer) + res(msg) + }, + reject: rej, + }) + }) + }, + close() { + ws.close() + }, + }) + } + }) +} + +// ─── Tests ───────────────────────────────────────────────────────────────── + +let server: ReturnType | null = null + +async function startTestServer() { + const { startServer } = await import('../index.js') + server = startServer(SERVER_PORT, '127.0.0.1') + await sleep(500) // Let server start + console.log(`\n✅ Server started on port ${SERVER_PORT}`) +} + +async function stopTestServer() { + if (server) { + server.stop(true) + server = null + await sleep(200) + console.log('✅ Server stopped') + } +} + +// ── Test 1: REST API health check ────────────────────────────────────── + +async function testHealthCheck() { + console.log('\n── Test 1: Health Check ──') + const res = await fetch(`${BASE_URL}/health`) + const body = await res.json() + if (res.status !== 200 || body.status !== 'ok') { + throw new Error(`Health check failed: ${JSON.stringify(body)}`) + } + console.log('✅ Health check passed') +} + +// ── Test 2: REST API sessions ────────────────────────────────────────── + +async function testSessionsApi() { + console.log('\n── Test 2: Sessions API ──') + const res = await fetch(`${BASE_URL}/api/sessions`) + if (res.status !== 200) { + throw new Error(`Sessions API failed: ${res.status}`) + } + const body = await res.json() + console.log(`✅ Sessions API returned ${body.sessions?.length ?? 0} sessions`) +} + +// ── Test 3: REST API settings ────────────────────────────────────────── + +async function testSettingsApi() { + console.log('\n── Test 3: Settings API ──') + const res = await fetch(`${BASE_URL}/api/settings`) + if (res.status !== 200) { + throw new Error(`Settings API failed: ${res.status}`) + } + const body = await res.json() + console.log(`✅ Settings API returned:`, Object.keys(body.settings || body)) +} + +// ── Test 4: REST API models ──────────────────────────────────────────── + +async function testModelsApi() { + console.log('\n── Test 4: Models API ──') + const res = await fetch(`${BASE_URL}/api/models`) + if (res.status !== 200) { + throw new Error(`Models API failed: ${res.status}`) + } + const body = await res.json() + console.log(`✅ Models API returned:`, body) +} + +// ── Test 5: WebSocket connection ─────────────────────────────────────── + +async function testWebSocketConnect() { + console.log('\n── Test 5: WebSocket Connect ──') + const sessionId = uuid() + const client = await createWebSocket(sessionId) + const connMsg = await client.waitForType('connected', 5000) + if (connMsg.sessionId !== sessionId) { + throw new Error(`Session ID mismatch: ${connMsg.sessionId} !== ${sessionId}`) + } + client.close() + await sleep(200) + console.log('✅ WebSocket connected and received session ID') +} + +// ── Test 6: WebSocket ping/pong ──────────────────────────────────────── + +async function testWebSocketPing() { + console.log('\n── Test 6: WebSocket Ping/Pong ──') + const sessionId = uuid() + const client = await createWebSocket(sessionId) + await client.waitForType('connected', 5000) + + client.ws.send(JSON.stringify({ type: 'ping' })) + const pong = await client.waitForType('pong', 5000) + if (pong.type !== 'pong') { + throw new Error('Pong not received') + } + client.close() + await sleep(200) + console.log('✅ Ping/Pong works') +} + +// ── Test 7: Real LLM chat (the critical test!) ──────────────────────── + +async function testRealLLMChat() { + console.log('\n── Test 7: Real LLM Chat (MiniMax API) ──') + console.log(' This will spawn a CLI subprocess and call the real API...') + + const sessionId = uuid() + const client = await createWebSocket(sessionId) + await client.waitForType('connected', 5000) + console.log(` Session: ${sessionId}`) + + // Send a simple message + client.ws.send( + JSON.stringify({ + type: 'user_message', + content: 'Say "hello world" and nothing else. Keep it short.', + }) + ) + + console.log(' Message sent, waiting for response...') + + // We should get a status:thinking first + const statusMsg = await client.waitForType('status', 10000) + console.log(` Got status: ${statusMsg.state} ${statusMsg.verb || ''}`) + + // Wait for either content_delta (success) or error + const responseMsg = await client.waitForAny( + ['content_delta', 'content_start', 'error', 'message_complete'], + 120000 // 2 minutes for LLM response + ) + + console.log(` Got response type: ${responseMsg.type}`) + + if (responseMsg.type === 'error') { + console.log(` ❌ Error: ${responseMsg.message} (code: ${responseMsg.code})`) + throw new Error(`LLM returned error: ${responseMsg.message}`) + } + + // Collect all messages until message_complete + let fullText = '' + if (responseMsg.type === 'content_delta' && responseMsg.text) { + fullText += responseMsg.text + } + + // Wait for message_complete + try { + const complete = await client.waitForType('message_complete', 120000) + console.log(` Usage: input=${complete.usage?.input_tokens}, output=${complete.usage?.output_tokens}`) + } catch { + console.log(' Warning: message_complete not received within timeout') + } + + // Gather all content_delta text + for (const msg of client.messages) { + if (msg.type === 'content_delta' && msg.text) { + if (!fullText.includes(msg.text)) { + fullText += msg.text + } + } + } + + console.log(` Response text: "${fullText.substring(0, 200)}"`) + + if (fullText.length === 0) { + // Check all messages for debugging + console.log(' All messages received:') + for (const msg of client.messages) { + console.log(` ${msg.type}: ${JSON.stringify(msg).substring(0, 150)}`) + } + throw new Error('No content received from LLM') + } + + client.close() + await sleep(500) + console.log('✅ Real LLM chat works! Got response from MiniMax API') +} + +// ── Test 8: Scheduled tasks CRUD ─────────────────────────────────────── + +async function testScheduledTasks() { + console.log('\n── Test 8: Scheduled Tasks CRUD ──') + + // Create + const createRes = await fetch(`${BASE_URL}/api/scheduled-tasks`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + name: 'Test Task', + prompt: 'Test prompt', + cron: '0 9 * * *', + }), + }) + if (createRes.status !== 201) { + const body = await createRes.text() + throw new Error(`Create scheduled task failed: ${createRes.status} ${body}`) + } + const { task } = await createRes.json() + console.log(` Created task: ${task.id}`) + + // List + const listRes = await fetch(`${BASE_URL}/api/scheduled-tasks`) + const listBody = await listRes.json() + const found = listBody.tasks?.find((t: any) => t.id === task.id) + if (!found) throw new Error('Created task not found in list') + console.log(` Listed ${listBody.tasks.length} tasks`) + + // Update + const updateRes = await fetch(`${BASE_URL}/api/scheduled-tasks/${task.id}`, { + method: 'PUT', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ name: 'Updated Task' }), + }) + if (updateRes.status !== 200) throw new Error(`Update failed: ${updateRes.status}`) + + // Delete + const deleteRes = await fetch(`${BASE_URL}/api/scheduled-tasks/${task.id}`, { + method: 'DELETE', + }) + if (deleteRes.status !== 200) throw new Error(`Delete failed: ${deleteRes.status}`) + + console.log('✅ Scheduled Tasks CRUD works') +} + +// ── Test 9: Settings read/write ──────────────────────────────────────── + +async function testSettingsReadWrite() { + console.log('\n── Test 9: Settings Read/Write ──') + + // Read current + const readRes = await fetch(`${BASE_URL}/api/settings`) + const original = await readRes.json() + + // Update via /api/settings/user + const updateRes = await fetch(`${BASE_URL}/api/settings/user`, { + method: 'PUT', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ testKey_integration: true }), + }) + if (updateRes.status !== 200) throw new Error(`Settings update failed: ${updateRes.status}`) + + // Read back + const readBack = await fetch(`${BASE_URL}/api/settings/user`) + const updated = await readBack.json() + + // Clean up: remove test key via overwrite + const cleanSettings = { ...updated } + delete cleanSettings.testKey_integration + await fetch(`${BASE_URL}/api/settings/user`, { + method: 'PUT', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(cleanSettings), + }) + + console.log('✅ Settings read/write works') +} + +// ── Test 10: Permission mode ────────────────────────────────────────── + +async function testPermissionMode() { + console.log('\n── Test 10: Permission Mode ──') + const res = await fetch(`${BASE_URL}/api/permissions/mode`) + if (res.status !== 200) throw new Error(`Permissions failed: ${res.status}`) + const body = await res.json() + console.log(` Current mode: ${body.mode || body.permissionMode || JSON.stringify(body)}`) + console.log('✅ Permission mode API works') +} + +// ── Test 11: Search API ──────────────────────────────────────────────── + +async function testSearchApi() { + console.log('\n── Test 11: Search API ──') + const res = await fetch(`${BASE_URL}/api/search/sessions`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ query: 'test' }), + }) + if (res.status !== 200) throw new Error(`Search failed: ${res.status}`) + const body = await res.json() + console.log(` Search results: ${body.results?.length ?? 0}`) + console.log('✅ Search API works') +} + +// ── Test 12: Agents API ──────────────────────────────────────────────── + +async function testAgentsApi() { + console.log('\n── Test 12: Agents API ──') + const res = await fetch(`${BASE_URL}/api/agents`) + if (res.status !== 200) throw new Error(`Agents failed: ${res.status}`) + const body = await res.json() + console.log(` Agents: ${body.agents?.length ?? 0}`) + console.log('✅ Agents API works') +} + +// ── Test 13: Teams API ───────────────────────────────────────────────── + +async function testTeamsApi() { + console.log('\n── Test 13: Teams API ──') + const res = await fetch(`${BASE_URL}/api/teams`) + if (res.status !== 200) throw new Error(`Teams failed: ${res.status}`) + const body = await res.json() + console.log(` Teams: ${body.teams?.length ?? 0}`) + console.log('✅ Teams API works') +} + +// ── Test 14: Tasks API ───────────────────────────────────────────────── + +async function testTasksApi() { + console.log('\n── Test 14: Tasks API ──') + const res = await fetch(`${BASE_URL}/api/tasks`) + if (res.status !== 200) throw new Error(`Tasks failed: ${res.status}`) + const body = await res.json() + console.log(` Tasks: ${body.tasks?.length ?? 0}`) + console.log('✅ Tasks API works') +} + +// ── Test 15: Status/Diagnostics API ──────────────────────────────────── + +async function testStatusApi() { + console.log('\n── Test 15: Status API ──') + const res = await fetch(`${BASE_URL}/api/status`) + if (res.status !== 200) throw new Error(`Status failed: ${res.status}`) + const body = await res.json() + console.log(` Status: ${body.status || JSON.stringify(body).substring(0, 100)}`) + console.log('✅ Status API works') +} + +// ─── Main ────────────────────────────────────────────────────────────────── + +async function main() { + console.log('╔══════════════════════════════════════════════════════════╗') + console.log('║ Real LLM Integration Test — MiniMax API via CLI ║') + console.log('╚══════════════════════════════════════════════════════════╝') + + const failures: string[] = [] + + try { + await startTestServer() + + // REST API tests (fast, run first) + const restTests = [ + testHealthCheck, + testSessionsApi, + testSettingsApi, + testModelsApi, + testScheduledTasks, + testSettingsReadWrite, + testPermissionMode, + testSearchApi, + testAgentsApi, + testTeamsApi, + testTasksApi, + testStatusApi, + ] + + for (const test of restTests) { + try { + await test() + } catch (err: any) { + console.log(`❌ ${test.name} FAILED: ${err.message}`) + failures.push(`${test.name}: ${err.message}`) + } + } + + // WebSocket tests + const wsTests = [testWebSocketConnect, testWebSocketPing] + + for (const test of wsTests) { + try { + await test() + } catch (err: any) { + console.log(`❌ ${test.name} FAILED: ${err.message}`) + failures.push(`${test.name}: ${err.message}`) + } + } + + // Real LLM test (slow, run last) + try { + await testRealLLMChat() + } catch (err: any) { + console.log(`❌ testRealLLMChat FAILED: ${err.message}`) + failures.push(`testRealLLMChat: ${err.message}`) + } + } finally { + await stopTestServer() + } + + // Summary + console.log('\n' + '═'.repeat(60)) + if (failures.length === 0) { + console.log('🎉 ALL TESTS PASSED!') + } else { + console.log(`⚠️ ${failures.length} TEST(S) FAILED:`) + for (const f of failures) { + console.log(` ❌ ${f}`) + } + } + console.log('═'.repeat(60)) + + process.exit(failures.length > 0 ? 1 : 0) +} + +main() diff --git a/src/server/__tests__/team-watcher.test.ts b/src/server/__tests__/team-watcher.test.ts new file mode 100644 index 00000000..1a90e38e --- /dev/null +++ b/src/server/__tests__/team-watcher.test.ts @@ -0,0 +1,400 @@ +/** + * Unit tests for TeamWatcher — real-time team status push via WebSocket + */ + +import { describe, it, expect, beforeEach, afterEach, mock } from 'bun:test' +import * as fs from 'node:fs/promises' +import * as fsSyn from 'node:fs' +import * as path from 'node:path' +import * as os from 'node:os' +import type { TeamMemberStatus } from '../ws/events.js' + +// ============================================================================ +// Test helpers +// ============================================================================ + +let tmpDir: string + +async function setupTmpConfigDir(): Promise { + tmpDir = path.join( + os.tmpdir(), + `claude-watcher-test-${Date.now()}-${Math.random().toString(36).slice(2)}`, + ) + await fs.mkdir(path.join(tmpDir, 'teams'), { 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 +} + +/** 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, + } +} + +// ============================================================================ +// Mock the WebSocket handler exports +// ============================================================================ + +// Track all messages sent via broadcast +let broadcastedMessages: Array<{ sessionId: string; message: unknown }> = [] +let mockActiveSessionIds: string[] = [] + +// We need to mock the handler module before importing TeamWatcher +// Use Bun's module mock +const mockSendToSession = mock((sessionId: string, message: unknown) => { + broadcastedMessages.push({ sessionId, message }) + return true +}) + +const mockGetActiveSessionIds = mock(() => { + return mockActiveSessionIds +}) + +// Mock the handler module +import { TeamWatcher } from '../services/teamWatcher.js' + +// Since TeamWatcher imports from handler.js at the module level, we need to +// test using the class directly and override the broadcast behavior. +// Instead, we test extractMemberStatuses directly and test the integration +// by verifying the check cycle behavior via a wrapper approach. + +// ============================================================================ +// TeamWatcher.extractMemberStatuses tests +// ============================================================================ + +describe('TeamWatcher.extractMemberStatuses', () => { + let watcher: TeamWatcher + + beforeEach(() => { + watcher = new TeamWatcher() + }) + + it('should extract member statuses from a valid config', () => { + const config = makeTeamConfig() + const statuses = watcher.extractMemberStatuses(config) + + expect(statuses).toHaveLength(2) + expect(statuses[0]).toEqual({ + agentId: 'agent-lead', + role: 'lead', + status: 'running', + currentTask: undefined, + }) + expect(statuses[1]).toEqual({ + agentId: 'agent-worker', + role: 'worker', + status: 'idle', + currentTask: undefined, + }) + }) + + it('should return running status when isActive is undefined', () => { + const config = makeTeamConfig() + delete (config.members[0] as Record).isActive + const statuses = watcher.extractMemberStatuses(config) + + expect(statuses[0]!.status).toBe('running') + }) + + it('should return idle status when isActive is false', () => { + const config = makeTeamConfig() + const statuses = watcher.extractMemberStatuses(config) + + expect(statuses[1]!.status).toBe('idle') + }) + + it('should use agentType as role', () => { + const config = makeTeamConfig() + const statuses = watcher.extractMemberStatuses(config) + + expect(statuses[0]!.role).toBe('lead') + expect(statuses[1]!.role).toBe('worker') + }) + + it('should fall back to name when agentType is missing', () => { + const config = makeTeamConfig() + delete (config.members[0] as Record).agentType + const statuses = watcher.extractMemberStatuses(config) + + expect(statuses[0]!.role).toBe('Lead Agent') + }) + + it('should fall back to "member" when both agentType and name are missing', () => { + const config = makeTeamConfig() + delete (config.members[0] as Record).agentType + delete (config.members[0] as Record).name + const statuses = watcher.extractMemberStatuses(config) + + expect(statuses[0]!.role).toBe('member') + }) + + it('should return empty array when config has no members', () => { + const config = { name: 'empty-team' } + const statuses = watcher.extractMemberStatuses(config) + expect(statuses).toEqual([]) + }) + + it('should return empty array when members is not an array', () => { + const config = { name: 'bad-team', members: 'not-an-array' } + const statuses = watcher.extractMemberStatuses(config) + expect(statuses).toEqual([]) + }) + + it('should include currentTask when present in config', () => { + const config = makeTeamConfig() + ;(config.members[0] as Record).currentTask = 'Implementing feature X' + const statuses = watcher.extractMemberStatuses(config) + + expect(statuses[0]!.currentTask).toBe('Implementing feature X') + }) +}) + +// ============================================================================ +// TeamWatcher polling integration tests +// ============================================================================ + +describe('TeamWatcher polling', () => { + let watcher: TeamWatcher + + beforeEach(async () => { + await setupTmpConfigDir() + watcher = new TeamWatcher() + }) + + afterEach(async () => { + watcher.stop() + watcher.reset() + await cleanupTmpDir() + }) + + it('should detect new team creation via checkNow()', async () => { + // First poll: no teams + watcher.checkNow() + + // Create a team + await writeTeamConfig('new-team', makeTeamConfig({ name: 'new-team' })) + + // The watcher internally calls broadcast which calls sendToSession. + // Since sendToSession depends on active sessions, we test that the + // internal snapshot state is updated correctly. + // After checkNow, the watcher should have recorded the team. + watcher.checkNow() + + // Now modify the team config and check again -- this proves the previous + // checkNow() recorded the snapshot (otherwise it would emit team_created again) + const updatedConfig = makeTeamConfig({ name: 'new-team', description: 'updated' }) + await writeTeamConfig('new-team', updatedConfig) + watcher.checkNow() + + // If we got here without errors, the snapshot logic is working + }) + + it('should detect team config changes', async () => { + // Create initial team + await writeTeamConfig('change-team', makeTeamConfig({ name: 'change-team' })) + + // First poll picks up the team + watcher.checkNow() + + // Modify the config + const updatedConfig = makeTeamConfig({ + name: 'change-team', + description: 'updated description', + }) + await writeTeamConfig('change-team', updatedConfig) + + // Second poll should detect the change + watcher.checkNow() + // No error means the diff detection worked + }) + + it('should detect team deletion', async () => { + // Create a team + await writeTeamConfig('doomed-team', makeTeamConfig({ name: 'doomed-team' })) + + // First poll picks it up + watcher.checkNow() + + // Delete the team directory + await fs.rm(path.join(tmpDir, 'teams', 'doomed-team'), { recursive: true, force: true }) + + // Next poll should detect deletion + watcher.checkNow() + // If no error, deletion detection worked + }) + + it('should handle missing teams directory gracefully', async () => { + // Remove the entire teams directory + await fs.rm(path.join(tmpDir, 'teams'), { recursive: true, force: true }) + + // Should not throw + watcher.checkNow() + }) + + it('should handle malformed config.json gracefully', async () => { + // Create a team with invalid JSON + const teamDir = path.join(tmpDir, 'teams', 'bad-json') + await fs.mkdir(teamDir, { recursive: true }) + await fs.writeFile(path.join(teamDir, 'config.json'), 'not valid json', 'utf-8') + + // Should not throw + watcher.checkNow() + }) + + it('should skip directories without config.json', async () => { + // Create a directory with no config.json + const teamDir = path.join(tmpDir, 'teams', 'no-config') + await fs.mkdir(teamDir, { recursive: true }) + + // Should not throw + watcher.checkNow() + }) + + it('should track multiple teams independently', async () => { + await writeTeamConfig('team-a', makeTeamConfig({ name: 'team-a' })) + await writeTeamConfig('team-b', makeTeamConfig({ name: 'team-b' })) + + // Pick up both teams + watcher.checkNow() + + // Modify only team-a + await writeTeamConfig('team-a', makeTeamConfig({ name: 'team-a', description: 'changed' })) + + // Should detect change in team-a but not team-b + watcher.checkNow() + + // Delete only team-b + await fs.rm(path.join(tmpDir, 'teams', 'team-b'), { recursive: true, force: true }) + + watcher.checkNow() + // No errors means independent tracking works + }) + + it('should start and stop polling without errors', async () => { + // Start with a short interval + watcher.start(50) + + // Let it run a couple of cycles + await new Promise((resolve) => setTimeout(resolve, 150)) + + // Stop + watcher.stop() + + // Starting again should work + watcher.start(50) + watcher.stop() + }) + + it('should not start duplicate intervals when start() called twice', async () => { + watcher.start(100) + watcher.start(100) // second call should be a no-op + + // Let it run briefly + await new Promise((resolve) => setTimeout(resolve, 50)) + + watcher.stop() + }) + + it('should handle teams directory appearing after initial check', async () => { + // Remove teams dir + await fs.rm(path.join(tmpDir, 'teams'), { recursive: true, force: true }) + + // First check -- no teams dir + watcher.checkNow() + + // Create teams dir and a team + await fs.mkdir(path.join(tmpDir, 'teams'), { recursive: true }) + await writeTeamConfig('late-team', makeTeamConfig({ name: 'late-team' })) + + // Second check should pick it up + watcher.checkNow() + }) + + it('reset() should clear internal state', async () => { + await writeTeamConfig('reset-team', makeTeamConfig({ name: 'reset-team' })) + + // Pick up the team + watcher.checkNow() + + // Reset and check again -- should treat it as new + watcher.reset() + watcher.checkNow() + // No error means reset worked + }) +}) + +// ============================================================================ +// Broadcast integration tests +// ============================================================================ + +describe('TeamWatcher broadcast', () => { + it('should call sendToSession for each active session', async () => { + // This test verifies the broadcast logic by importing the real module + // and checking that getActiveSessionIds/sendToSession are called. + // Since the handler module manages real WebSocket state, we verify + // that when there are no active sessions, broadcast is a no-op. + + await setupTmpConfigDir() + const watcher = new TeamWatcher() + + await writeTeamConfig('broadcast-team', makeTeamConfig({ name: 'broadcast-team' })) + + // With no active WebSocket sessions, checkNow should still succeed + // (broadcast sends to zero sessions) + watcher.checkNow() + + watcher.stop() + watcher.reset() + await cleanupTmpDir() + }) +}) diff --git a/src/server/api/scheduled-tasks.ts b/src/server/api/scheduled-tasks.ts index 13de0f2f..da7ea778 100644 --- a/src/server/api/scheduled-tasks.ts +++ b/src/server/api/scheduled-tasks.ts @@ -1,13 +1,16 @@ /** * Scheduled Tasks REST API * - * GET /api/scheduled-tasks — 获取任务列表 - * POST /api/scheduled-tasks — 创建任务 - * PUT /api/scheduled-tasks/:id — 更新任务 - * DELETE /api/scheduled-tasks/:id — 删除任务 + * GET /api/scheduled-tasks — 获取任务列表 + * POST /api/scheduled-tasks — 创建任务 + * GET /api/scheduled-tasks/runs — 获取所有任务的最近执行记录 + * GET /api/scheduled-tasks/:id/runs — 获取指定任务的执行记录 + * PUT /api/scheduled-tasks/:id — 更新任务 + * DELETE /api/scheduled-tasks/:id — 删除任务 */ import { CronService } from '../services/cronService.js' +import { cronScheduler } from '../services/cronScheduler.js' import { ApiError, errorResponse } from '../middleware/errorHandler.js' const cronService = new CronService() @@ -19,7 +22,22 @@ export async function handleScheduledTasksApi( ): Promise { try { const method = req.method - const taskId = segments[2] // /api/scheduled-tasks/:id + const taskId = segments[2] // /api/scheduled-tasks/:id or "runs" + const subResource = segments[3] // /api/scheduled-tasks/:id/runs + + // ── GET /api/scheduled-tasks/runs ──────────────────────────────────── + if (method === 'GET' && taskId === 'runs') { + const url = new URL(req.url) + const limit = parseInt(url.searchParams.get('limit') || '50', 10) + const runs = await cronScheduler.getRecentRuns(limit) + return Response.json({ runs }) + } + + // ── GET /api/scheduled-tasks/:id/runs ──────────────────────────────── + if (method === 'GET' && taskId && subResource === 'runs') { + const runs = await cronScheduler.getTaskRuns(taskId) + return Response.json({ runs }) + } // ── GET /api/scheduled-tasks ────────────────────────────────────────── if (method === 'GET' && !taskId) { @@ -35,6 +53,7 @@ export async function handleScheduledTasksApi( description: body.description as string | undefined, cron: body.cron as string, prompt: body.prompt as string, + enabled: body.enabled !== undefined ? (body.enabled as boolean) : undefined, recurring: body.recurring as boolean | undefined, permanent: body.permanent as boolean | undefined, permissionMode: body.permissionMode as string | undefined, @@ -46,21 +65,21 @@ export async function handleScheduledTasksApi( } // ── PUT /api/scheduled-tasks/:id ────────────────────────────────────── - if (method === 'PUT' && taskId) { + if (method === 'PUT' && taskId && !subResource) { const body = await parseJsonBody(req) const task = await cronService.updateTask(taskId, body) return Response.json({ task }) } // ── DELETE /api/scheduled-tasks/:id ─────────────────────────────────── - if (method === 'DELETE' && taskId) { + if (method === 'DELETE' && taskId && !subResource) { await cronService.deleteTask(taskId) return Response.json({ ok: true }) } throw new ApiError( 405, - `Method ${method} not allowed on /api/scheduled-tasks${taskId ? `/${taskId}` : ''}`, + `Method ${method} not allowed on /api/scheduled-tasks${taskId ? `/${taskId}` : ''}${subResource ? `/${subResource}` : ''}`, 'METHOD_NOT_ALLOWED', ) } catch (error) { diff --git a/src/server/index.ts b/src/server/index.ts index 34b1e439..9bbc6acb 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -9,6 +9,8 @@ import { handleApiRequest } from './router.js' import { handleWebSocket, type WebSocketData } from './ws/handler.js' import { corsHeaders } from './middleware/cors.js' import { requireAuth } from './middleware/auth.js' +import { teamWatcher } from './services/teamWatcher.js' +import { cronScheduler } from './services/cronScheduler.js' const PORT = parseInt(process.env.SERVER_PORT || '3456', 10) const HOST = process.env.SERVER_HOST || '127.0.0.1' @@ -78,6 +80,12 @@ export function startServer(port = PORT, host = HOST) { websocket: handleWebSocket, }) + // Start watching ~/.claude/teams/ for real-time WebSocket push + teamWatcher.start() + + // Start the cron scheduler to execute scheduled tasks + cronScheduler.start() + console.log(`[Server] Claude Code API server running at http://${host}:${port}`) return server } diff --git a/src/server/services/cronScheduler.ts b/src/server/services/cronScheduler.ts new file mode 100644 index 00000000..e1ff63d4 --- /dev/null +++ b/src/server/services/cronScheduler.ts @@ -0,0 +1,430 @@ +/** + * CronScheduler — Execution engine for scheduled tasks + * + * Periodically checks all scheduled tasks and executes those whose cron + * expression matches the current time. Tasks are run by spawning a CLI + * subprocess with the task's prompt. Execution history is persisted to + * ~/.claude/scheduled_tasks_log.json. + */ + +import * as fs from 'fs/promises' +import * as path from 'path' +import * as os from 'os' +import * as crypto from 'crypto' +import { CronService, type CronTask } from './cronService.js' + +// ─── Types ───────────────────────────────────────────────────────────────────── + +export type TaskRun = { + id: string // random ID + taskId: string // references CronTask.id + taskName: string + startedAt: string // ISO timestamp + completedAt?: string + status: 'running' | 'completed' | 'failed' | 'timeout' + prompt: string + output?: string // captured stdout summary + error?: string + exitCode?: number + durationMs?: number +} + +// ─── Cron expression matching ────────────────────────────────────────────────── + +/** + * Check whether a single cron field matches a given numeric value. + * + * Supported syntax per field: + * * — any value + * 5 — exact match + * 1,3,5 — list + * 1-5 — inclusive range + * *​/2 — step from 0 + * 1-10/3 — step within a range + */ +export function fieldMatches(field: string, value: number): boolean { + if (field === '*') return true + + // Comma-separated list — each element can be a range or step + const parts = field.split(',') + return parts.some((part) => singleFieldMatches(part.trim(), value)) +} + +function singleFieldMatches(part: string, value: number): boolean { + // Step: */n or range/n + if (part.includes('/')) { + const [rangePart, stepStr] = part.split('/') + const step = parseInt(stepStr, 10) + if (isNaN(step) || step <= 0) return false + + if (rangePart === '*') { + return value % step === 0 + } + // range/step e.g. 1-10/3 + if (rangePart.includes('-')) { + const [startStr, endStr] = rangePart.split('-') + const start = parseInt(startStr, 10) + const end = parseInt(endStr, 10) + if (value < start || value > end) return false + return (value - start) % step === 0 + } + // single/step e.g. 5/2 — treat as start with step + const start = parseInt(rangePart, 10) + if (value < start) return false + return (value - start) % step === 0 + } + + // Range: a-b + if (part.includes('-')) { + const [startStr, endStr] = part.split('-') + const start = parseInt(startStr, 10) + const end = parseInt(endStr, 10) + return value >= start && value <= end + } + + // Exact number + return parseInt(part, 10) === value +} + +/** + * Check whether a standard 5-field cron expression matches the given date. + * Fields: minute hour day-of-month month day-of-week + */ +export function cronMatches(cronExpr: string, date: Date): boolean { + const fields = cronExpr.trim().split(/\s+/) + if (fields.length !== 5) return false + + const [minute, hour, dayOfMonth, month, dayOfWeek] = fields + return ( + fieldMatches(minute, date.getMinutes()) && + fieldMatches(hour, date.getHours()) && + fieldMatches(dayOfMonth, date.getDate()) && + fieldMatches(month, date.getMonth() + 1) && + fieldMatches(dayOfWeek, date.getDay()) + ) +} + +// ─── Log file I/O ────────────────────────────────────────────────────────────── + +type RunsFile = { runs: TaskRun[] } + +function getLogFilePath(): string { + const configDir = + process.env.CLAUDE_CONFIG_DIR || path.join(os.homedir(), '.claude') + return path.join(configDir, 'scheduled_tasks_log.json') +} + +async function readRunsFile(): Promise { + try { + const raw = await fs.readFile(getLogFilePath(), 'utf-8') + const parsed = JSON.parse(raw) as RunsFile + if (!Array.isArray(parsed.runs)) return { runs: [] } + return parsed + } catch (err: unknown) { + if ((err as NodeJS.ErrnoException).code === 'ENOENT') { + return { runs: [] } + } + throw err + } +} + +async function writeRunsFile(data: RunsFile): Promise { + const filePath = getLogFilePath() + const dir = path.dirname(filePath) + await fs.mkdir(dir, { recursive: true }) + + const tmpFile = `${filePath}.tmp.${Date.now()}` + try { + await fs.writeFile(tmpFile, JSON.stringify(data, null, 2) + '\n', 'utf-8') + await fs.rename(tmpFile, filePath) + } catch (err) { + await fs.unlink(tmpFile).catch(() => {}) + throw err + } +} + +/** Append a run to the log and trim to keep at most MAX_RUNS_PER_TASK per task. */ +async function appendRun(run: TaskRun): Promise { + const data = await readRunsFile() + data.runs.push(run) + trimRuns(data) + await writeRunsFile(data) +} + +/** Update an existing run in the log (matched by run.id). */ +async function updateRun(run: TaskRun): Promise { + const data = await readRunsFile() + const idx = data.runs.findIndex((r) => r.id === run.id) + if (idx !== -1) { + data.runs[idx] = run + } else { + data.runs.push(run) + } + trimRuns(data) + await writeRunsFile(data) +} + +const MAX_RUNS_PER_TASK = 100 + +/** Keep only the latest MAX_RUNS_PER_TASK entries per task. */ +function trimRuns(data: RunsFile): void { + const countByTask = new Map() + // Count from the end (newest first) and mark for removal + const keep = new Array(data.runs.length).fill(false) + for (let i = data.runs.length - 1; i >= 0; i--) { + const taskId = data.runs[i].taskId + const count = countByTask.get(taskId) || 0 + if (count < MAX_RUNS_PER_TASK) { + keep[i] = true + countByTask.set(taskId, count + 1) + } + } + data.runs = data.runs.filter((_, i) => keep[i]) +} + +// ─── Scheduler ───────────────────────────────────────────────────────────────── + +const TASK_TIMEOUT_MS = 10 * 60 * 1000 // 10 minutes + +export class CronScheduler { + private intervalId: Timer | null = null + private runningTasks = new Map< + string, + { proc: ReturnType; startedAt: number; runId: string } + >() + private cronService: CronService + + constructor(cronService?: CronService) { + this.cronService = cronService || new CronService() + } + + /** Start the scheduler (called on server boot). */ + start(): void { + if (this.intervalId) return // already running + console.log('[CronScheduler] Starting — checking every 60 s') + this.intervalId = setInterval(() => this.tick(), 60_000) + // Immediate first check + this.tick() + } + + /** Stop the scheduler and kill any running task processes. */ + stop(): void { + if (this.intervalId) { + clearInterval(this.intervalId) + this.intervalId = null + } + for (const [taskId, entry] of this.runningTasks) { + try { + entry.proc.kill() + } catch { + // process may have already exited + } + this.runningTasks.delete(taskId) + } + console.log('[CronScheduler] Stopped') + } + + /** One tick of the scheduler — evaluate all tasks against the current time. */ + async tick(): Promise { + try { + const tasks = await this.cronService.listTasks() + const now = new Date() + + for (const task of tasks) { + // Skip disabled tasks + if (task.enabled === false) continue + + // Skip if already running + if (this.runningTasks.has(task.id)) continue + + if (cronMatches(task.cron, now)) { + // Fire and forget — don't await; we want all matching tasks to start + this.executeTask(task).catch((err) => { + console.error( + `[CronScheduler] Unhandled error executing task ${task.id}:`, + err, + ) + }) + } + } + } catch (err) { + console.error('[CronScheduler] Error during tick:', err) + } + } + + /** Execute a single task by spawning a CLI subprocess. */ + async executeTask(task: CronTask): Promise { + const runId = crypto.randomBytes(6).toString('hex') + const startedAt = new Date().toISOString() + + const run: TaskRun = { + id: runId, + taskId: task.id, + taskName: task.name || task.prompt.slice(0, 60), + startedAt, + status: 'running', + prompt: task.prompt, + } + + // Persist the "running" state + await appendRun(run) + + // Resolve CLI entry point relative to this file + const cliPath = path.resolve(import.meta.dir, '../../entrypoints/cli.tsx') + + const inputPayload = JSON.stringify({ + type: 'user', + content: task.prompt, + }) + '\n' + + const proc = Bun.spawn( + [ + 'bun', + cliPath, + '--print', + '--input-format', + 'stream-json', + '--output-format', + 'stream-json', + ], + { + stdin: 'pipe', + stdout: 'pipe', + stderr: 'pipe', + cwd: task.folderPath || os.homedir(), + }, + ) + + this.runningTasks.set(task.id, { proc, startedAt: Date.now(), runId }) + + // Write prompt to stdin then close it + try { + proc.stdin.write(inputPayload) + proc.stdin.end() + } catch { + // If writing fails, the process may have already exited + } + + // Set up a timeout + const timeoutId = setTimeout(() => { + if (this.runningTasks.has(task.id)) { + try { + proc.kill() + } catch { + // ignore + } + } + }, TASK_TIMEOUT_MS) + + try { + // Collect stdout + const stdoutChunks: string[] = [] + if (proc.stdout) { + const reader = proc.stdout.getReader() + const decoder = new TextDecoder() + try { + while (true) { + const { done, value } = await reader.read() + if (done) break + stdoutChunks.push(decoder.decode(value, { stream: true })) + } + } catch { + // stream may be interrupted on kill + } + } + + // Wait for exit + const exitCode = await proc.exited + + clearTimeout(timeoutId) + this.runningTasks.delete(task.id) + + const completedAt = new Date().toISOString() + const output = stdoutChunks.join('') + const durationMs = + new Date(completedAt).getTime() - new Date(startedAt).getTime() + + // Determine if this was a timeout + const wasTimeout = durationMs >= TASK_TIMEOUT_MS + + const completedRun: TaskRun = { + ...run, + completedAt, + status: wasTimeout ? 'timeout' : exitCode === 0 ? 'completed' : 'failed', + output: output.slice(0, 10_000), // cap stored output + exitCode, + durationMs, + } + + // Collect stderr for error field + if (exitCode !== 0 && proc.stderr) { + try { + const stderrText = await new Response(proc.stderr).text() + completedRun.error = stderrText.slice(0, 5_000) + } catch { + // ignore + } + } + + await updateRun(completedRun) + + // Update lastFiredAt on the task + await this.cronService.updateLastFired(task.id, startedAt) + + // If non-recurring, disable after first run + if (!task.recurring) { + await this.cronService.updateTask(task.id, { enabled: false }).catch(() => { + // Task may have been deleted + }) + } + + return completedRun + } catch (err) { + clearTimeout(timeoutId) + this.runningTasks.delete(task.id) + + const completedAt = new Date().toISOString() + const failedRun: TaskRun = { + ...run, + completedAt, + status: 'failed', + error: (err as Error).message, + durationMs: + new Date(completedAt).getTime() - new Date(startedAt).getTime(), + } + + await updateRun(failedRun) + await this.cronService.updateLastFired(task.id, startedAt) + + return failedRun + } + } + + // ─── Query helpers ───────────────────────────────────────────────────────── + + /** Get execution history for a specific task. */ + async getTaskRuns(taskId: string): Promise { + const data = await readRunsFile() + return data.runs + .filter((r) => r.taskId === taskId) + .sort( + (a, b) => + new Date(b.startedAt).getTime() - new Date(a.startedAt).getTime(), + ) + } + + /** Get recent runs across all tasks. */ + async getRecentRuns(limit = 50): Promise { + const data = await readRunsFile() + return data.runs + .sort( + (a, b) => + new Date(b.startedAt).getTime() - new Date(a.startedAt).getTime(), + ) + .slice(0, limit) + } +} + +// ─── Singleton export ────────────────────────────────────────────────────────── + +export const cronScheduler = new CronScheduler() diff --git a/src/server/services/cronService.ts b/src/server/services/cronService.ts index d6c51508..09204409 100644 --- a/src/server/services/cronService.ts +++ b/src/server/services/cronService.ts @@ -18,7 +18,8 @@ export type CronTask = { cron: string // 5-field cron expression prompt: string createdAt: number // epoch ms - lastFiredAt?: number + lastFiredAt?: string // ISO timestamp of last execution + enabled?: boolean // allow disabling without deleting (default true) recurring?: boolean permanent?: boolean permissionMode?: string @@ -94,6 +95,17 @@ export class CronService { await this.writeTasksFile(data) } + /** 更新任务的最后执行时间 */ + async updateLastFired(taskId: string, timestamp: string): Promise { + const data = await this.readTasksFile() + const index = data.tasks.findIndex((t) => t.id === taskId) + if (index === -1) { + return // Task may have been deleted; silently ignore + } + data.tasks[index].lastFiredAt = timestamp + await this.writeTasksFile(data) + } + // --------------------------------------------------------------------------- // 内部: 文件读写 // --------------------------------------------------------------------------- diff --git a/src/server/services/teamWatcher.ts b/src/server/services/teamWatcher.ts new file mode 100644 index 00000000..bf0f3ea0 --- /dev/null +++ b/src/server/services/teamWatcher.ts @@ -0,0 +1,162 @@ +/** + * TeamWatcher -- monitors ~/.claude/teams/ for changes and pushes + * real-time updates to all connected WebSocket clients. + * + * Uses polling (setInterval) rather than fs.watch for cross-platform reliability. + * Detects three kinds of events: + * - team_created : a new team directory with config.json appears + * - team_update : an existing team's config.json content changes + * - team_deleted : a previously-seen team directory disappears + */ + +import * as fs from 'fs' +import * as path from 'path' +import * as os from 'os' +import { sendToSession, getActiveSessionIds } from '../ws/handler.js' +import type { ServerMessage, TeamMemberStatus } from '../ws/events.js' + +// ─── Helpers ────────────────────────────────────────────────────────────── + +function getTeamsDir(): string { + const configDir = process.env.CLAUDE_CONFIG_DIR || path.join(os.homedir(), '.claude') + return path.join(configDir, 'teams') +} + +// ─── TeamWatcher ────────────────────────────────────────────────────────── + +export class TeamWatcher { + private intervalId: ReturnType | null = null + private lastSnapshots = new Map() // teamName -> raw JSON content + + /** Start polling for team changes. */ + start(intervalMs = 3000): void { + if (this.intervalId) return // already running + // Run an initial check immediately, then start the interval + this.check() + this.intervalId = setInterval(() => this.check(), intervalMs) + } + + /** Stop polling. */ + stop(): void { + if (this.intervalId) { + clearInterval(this.intervalId) + this.intervalId = null + } + } + + /** Visible for testing -- force a single poll cycle. */ + checkNow(): void { + this.check() + } + + /** Clear internal snapshot state (useful in tests). */ + reset(): void { + this.lastSnapshots.clear() + } + + // ── Core polling logic ───────────────────────────────────────────────── + + private check(): void { + const teamsDir = getTeamsDir() + + let entries: fs.Dirent[] + try { + entries = fs.readdirSync(teamsDir, { withFileTypes: true }) + } catch { + // teams directory doesn't exist yet -- nothing to watch + // If we previously knew about teams, they are now all "deleted" + for (const [name] of this.lastSnapshots) { + this.broadcast({ type: 'team_deleted', teamName: name }) + } + this.lastSnapshots.clear() + return + } + + const currentTeamNames = new Set() + + for (const entry of entries) { + if (!entry.isDirectory()) continue + const teamName = entry.name + currentTeamNames.add(teamName) + + const configPath = path.join(teamsDir, teamName, 'config.json') + let content: string + try { + content = fs.readFileSync(configPath, 'utf-8') + } catch { + // config.json not readable (missing / permissions) -- skip + continue + } + + const lastContent = this.lastSnapshots.get(teamName) + + if (lastContent === undefined) { + // New team detected + this.lastSnapshots.set(teamName, content) + this.broadcast({ type: 'team_created', teamName }) + } else if (content !== lastContent) { + // Team config changed -- extract member statuses and broadcast + this.lastSnapshots.set(teamName, content) + try { + const config = JSON.parse(content) + const members = this.extractMemberStatuses(config) + this.broadcast({ type: 'team_update', teamName, members }) + } catch { + // JSON parse failed -- broadcast with empty members + this.broadcast({ type: 'team_update', teamName, members: [] }) + } + } + // else: content unchanged, nothing to do + } + + // Check for deleted teams (were in lastSnapshots but no longer on disk) + for (const [name] of this.lastSnapshots) { + if (!currentTeamNames.has(name)) { + this.lastSnapshots.delete(name) + this.broadcast({ type: 'team_deleted', teamName: name }) + } + } + } + + // ── Member status extraction ─────────────────────────────────────────── + + /** + * Parse the TeamFile config and derive a TeamMemberStatus for each member. + * + * The raw config has: + * members: [{ agentId, name, agentType, isActive, sessionId, ... }] + * + * We map `isActive` to the status enum and use `agentType` / `name` as role. + */ + extractMemberStatuses(config: Record): TeamMemberStatus[] { + const members = config.members + if (!Array.isArray(members)) return [] + + return members.map((m: Record) => { + const status = this.deriveStatus(m.isActive as boolean | undefined) + return { + agentId: (m.agentId as string) || '', + role: (m.agentType as string) || (m.name as string) || 'member', + status, + currentTask: (m.currentTask as string) || undefined, + } + }) + } + + private deriveStatus(isActive: boolean | undefined): TeamMemberStatus['status'] { + if (isActive === false) return 'idle' + // isActive === true or undefined => running + return 'running' + } + + // ── Broadcasting ─────────────────────────────────────────────────────── + + private broadcast(message: ServerMessage): void { + const sessionIds = getActiveSessionIds() + for (const id of sessionIds) { + sendToSession(id, message) + } + } +} + +export const teamWatcher = new TeamWatcher() diff --git a/src/server/ws/events.ts b/src/server/ws/events.ts index b85c5f12..77473348 100644 --- a/src/server/ws/events.ts +++ b/src/server/ws/events.ts @@ -36,7 +36,12 @@ export type ServerMessage = | { type: 'thinking'; text: string } | { type: 'status'; state: ChatState; verb?: string; elapsed?: number; tokens?: number } | { type: 'error'; message: string; code: string; retryable?: boolean } + | { type: 'system_notification'; subtype: string; message?: string; data?: unknown } | { type: 'pong' } + | { type: 'team_update'; teamName: string; members: TeamMemberStatus[] } + | { type: 'team_created'; teamName: string } + | { type: 'team_deleted'; teamName: string } + | { type: 'task_update'; taskId: string; status: string; progress?: string } export type TokenUsage = { input_tokens: number @@ -47,6 +52,13 @@ export type TokenUsage = { export type ChatState = 'idle' | 'thinking' | 'tool_executing' | 'streaming' | 'permission_pending' +export type TeamMemberStatus = { + agentId: string + role: string + status: 'running' | 'idle' | 'completed' | 'error' + currentTask?: string +} + // ============================================================================ // Internal types // ============================================================================ diff --git a/src/server/ws/handler.ts b/src/server/ws/handler.ts index 60b2af41..09395981 100644 --- a/src/server/ws/handler.ts +++ b/src/server/ws/handler.ts @@ -99,9 +99,9 @@ async function handleUserMessage( // 注册 CLI stdout → WebSocket 转发 conversationService.onOutput(sessionId, (cliMsg) => { - const serverMsg = translateCliMessage(cliMsg) - if (serverMsg) { - sendMessage(ws, serverMsg) + const serverMsgs = translateCliMessage(cliMsg) + for (const msg of serverMsgs) { + sendMessage(ws, msg) } }) } catch (err) { @@ -146,60 +146,256 @@ function handleStopGeneration(ws: ServerWebSocket) { // ============================================================================ /** - * 将 CLI stdout 的 stream-json 消息转换为 WebSocket ServerMessage。 + * Track the type of the currently active content block by stream index. + * Used to determine whether a content_block_stop corresponds to a tool_use or text block. + */ +const activeBlockTypes = new Map() + +/** + * 将 CLI stdout 的 stream-json 消息转换为 WebSocket ServerMessage 数组。 * * CLI 输出格式参考 src/bridge/sessionRunner.ts 中的 stream-json 协议。 + * 返回数组以支持单条 CLI 消息产生多条 ServerMessage(例如多个内容块)。 */ -function translateCliMessage(cliMsg: any): ServerMessage | null { +function translateCliMessage(cliMsg: any): ServerMessage[] { 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 } + // 检查是否有认证错误 + if (cliMsg.error) { + return [{ + type: 'error', + message: cliMsg.message?.content?.[0]?.text || cliMsg.error, + code: cliMsg.error, + }] + } + + // 助手消息 - 提取所有内容块,每个块产生独立的 ServerMessage + if (cliMsg.message?.content && Array.isArray(cliMsg.message.content)) { + const messages: ServerMessage[] = [] + + for (const block of cliMsg.message.content) { + if (block.type === 'thinking' && block.thinking) { + // Bug #5: 处理 thinking 块 + messages.push({ type: 'thinking', text: block.thinking }) + } else if (block.type === 'text' && block.text) { + messages.push({ type: 'content_start', blockType: 'text' }) + messages.push({ type: 'content_delta', text: block.text }) + } else if (block.type === 'tool_use') { + // Bug #2: 不再 return,而是 push 到数组中 + // Bug #3: 发送 tool input + // Bug #4: 发送 tool_use_complete + messages.push({ + type: 'content_start', + blockType: 'tool_use', + toolName: block.name, + toolUseId: block.id, + }) + if (block.input !== undefined) { + messages.push({ + type: 'content_delta', + toolInput: JSON.stringify(block.input), + }) } + messages.push({ + type: 'tool_use_complete', + toolName: block.name, + toolUseId: block.id, + input: block.input, + }) + } + } + + return messages + } + return [] + } + + case 'user': { + // Bug #1: 处理 tool_result 消息 + // CLI 发送 type:'user' 消息,其中 content 包含 tool_result 块 + const messages: ServerMessage[] = [] + + if (cliMsg.message?.content && Array.isArray(cliMsg.message.content)) { + for (const block of cliMsg.message.content) { + if (block.type === 'tool_result') { + messages.push({ + type: 'tool_result', + toolUseId: block.tool_use_id, + content: block.content, + isError: !!block.is_error, + }) } } } - return null + + return messages + } + + case 'stream_event': { + // Bug #6: 处理增量流式事件 + const event = cliMsg.event + if (!event) return [] + + switch (event.type) { + case 'message_start': { + return [{ type: 'status', state: 'streaming' }] + } + + case 'content_block_start': { + const contentBlock = event.content_block + if (!contentBlock) return [] + + // Track the block type by index for content_block_stop + const index = event.index ?? 0 + activeBlockTypes.set(index, contentBlock.type === 'tool_use' ? 'tool_use' : 'text') + + if (contentBlock.type === 'tool_use') { + return [{ + type: 'content_start', + blockType: 'tool_use', + toolName: contentBlock.name, + toolUseId: contentBlock.id, + }] + } + return [{ type: 'content_start', blockType: 'text' }] + } + + case 'content_block_delta': { + const delta = event.delta + if (!delta) return [] + + if (delta.type === 'text_delta' && delta.text) { + return [{ type: 'content_delta', text: delta.text }] + } + if (delta.type === 'input_json_delta' && delta.partial_json) { + return [{ type: 'content_delta', toolInput: delta.partial_json }] + } + if (delta.type === 'thinking_delta' && delta.thinking) { + return [{ type: 'thinking', text: delta.thinking }] + } + return [] + } + + case 'content_block_stop': { + const index = event.index ?? 0 + const blockType = activeBlockTypes.get(index) + activeBlockTypes.delete(index) + + // For tool_use blocks, emit tool_use_complete + // Note: toolName and toolUseId may not be available at stop time + // from stream events alone; the UI should track them from content_block_start + if (blockType === 'tool_use') { + return [{ + type: 'tool_use_complete', + toolName: '', + toolUseId: '', + input: null, + }] + } + return [] + } + + case 'message_stop': { + // message_stop is handled by the 'result' message + return [] + } + + case 'message_delta': { + // message_delta may contain stop_reason or usage updates + return [] + } + + default: + return [] + } } case 'control_request': { // 权限请求 — CLI 需要用户授权才能执行工具 if (cliMsg.request?.subtype === 'can_use_tool') { - return { + return [{ type: 'permission_request', requestId: cliMsg.request_id, toolName: cliMsg.request.tool_name || 'Unknown', input: cliMsg.request.input || {}, description: cliMsg.request.description, - } + }] } - return null + return [] } 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, - }, - } + // 对话结果(成功或错误) + const usage = { + input_tokens: cliMsg.usage?.input_tokens || 0, + output_tokens: cliMsg.usage?.output_tokens || 0, } - return null + + if (cliMsg.is_error) { + // 错误和完成消息都发送 + return [ + { + type: 'error', + message: cliMsg.result || 'Unknown error', + code: 'CLI_ERROR', + }, + { type: 'message_complete', usage }, + ] + } + + return [{ type: 'message_complete', usage }] } - case 'system': - return { type: 'status', state: 'idle' } + case 'system': { + // 区分不同的 system 子类型 + const subtype = cliMsg.subtype + if (subtype === 'init') { + // CLI 初始化完成 — 发送模型信息 + return [{ type: 'status', state: 'idle', verb: `Model: ${cliMsg.model || 'unknown'}` }] + } + if (subtype === 'hook_started' || subtype === 'hook_response') { + // Hook 执行中 — 不转发给前端 + return [] + } + // Bug #7: 处理 task/team system 消息 + if (subtype === 'task_notification') { + return [{ + type: 'system_notification', + subtype: 'task_notification', + message: cliMsg.message || cliMsg.title, + data: cliMsg, + }] + } + if (subtype === 'task_started') { + return [{ + type: 'status', + state: 'tool_executing', + verb: cliMsg.message || 'Task started', + }] + } + if (subtype === 'task_progress') { + return [{ + type: 'status', + state: 'tool_executing', + verb: cliMsg.message || 'Task in progress', + }] + } + if (subtype === 'session_state_changed') { + return [{ + type: 'system_notification', + subtype: 'session_state_changed', + message: cliMsg.message, + data: cliMsg, + }] + } + // 其他 system 消息 + return [] + } default: - return null + // 未知类型 — 调试输出但不转发 + console.log(`[WS] Unknown CLI message type: ${cliMsg.type}`, JSON.stringify(cliMsg).substring(0, 200)) + return [] } }