diff --git a/src/config.ts b/src/config.ts index f52c7e6e..8f5a4082 100644 --- a/src/config.ts +++ b/src/config.ts @@ -50,6 +50,7 @@ export const config = { // 数据库配置 db: { sessionDbPath: process.env.SESSION_DB_PATH || './data/sessions.db', + pipelineDbPath: process.env.PIPELINE_DB_PATH || './data/pipelines.db', }, // 服务配置 diff --git a/src/feishu/event-handler.ts b/src/feishu/event-handler.ts index 782d7afd..55e6a67d 100644 --- a/src/feishu/event-handler.ts +++ b/src/feishu/event-handler.ts @@ -4,12 +4,19 @@ import { isUserAllowed, containsDangerousCommand } from '../utils/security.js'; import { sessionManager } from '../session/manager.js'; import { taskQueue } from '../session/queue.js'; import { claudeExecutor } from '../claude/executor.js'; -import { buildProgressCard, buildResultCard, buildStreamingCard, buildPipelineCard, buildStatusCard } from './message-builder.js'; -import { PipelineOrchestrator } from '../pipeline/orchestrator.js'; -import { PHASE_META, TOTAL_PHASES } from '../pipeline/types.js'; +import { buildProgressCard, buildResultCard, buildStreamingCard, buildStatusCard, buildCancelledCard, buildPipelineCard, buildPipelineConfirmCard } from './message-builder.js'; import { feishuClient } from './client.js'; import { config } from '../config.js'; import { setupWorkspace } from '../workspace/manager.js'; +import { ensureThread } from './thread-utils.js'; +import { pipelineStore } from '../pipeline/store.js'; +import { + createPendingPipeline, + startPipeline, + abortPipeline, + cancelPipeline, + retryPipeline, +} from '../pipeline/runner.js'; // ============================================================ // 使用飞书 SDK 的 EventDispatcher 处理事件 @@ -73,14 +80,95 @@ export function createCardActionHandler(): lark.CardActionHandler { encryptKey: config.feishu.encryptKey || undefined, verificationToken: config.feishu.verifyToken || undefined, }, async (data: Record) => { - logger.debug({ action: data }, 'Card action received'); - // TODO: 处理卡片按钮点击等交互 - return {}; + const action = data.action as { value?: Record } | undefined; + const actionType = action?.value?.action as string | undefined; + const pipelineId = action?.value?.pipelineId as string | undefined; + + // 提取操作者 user ID + const operatorId = (data.operator as { open_id?: string } | undefined)?.open_id; + + logger.info({ actionType, pipelineId, operatorId }, 'Card action received'); + + if (!actionType || !pipelineId) return {}; + + // 验证操作者身份：无法识别身份时拒绝操作（fail closed） + if (!operatorId) { + logger.warn({ pipelineId }, 'Card action rejected: no operator identity'); + return {}; + } + + // 只有管道创建者可以操作 + const record = pipelineStore.get(pipelineId); + if (record && record.userId !== operatorId) { + logger.warn({ pipelineId, operatorId, ownerId: record.userId }, 'Card action rejected: operator is not pipeline owner'); + return {}; + } + + switch (actionType) { + case 'pipeline_confirm': + return handlePipelineConfirm(pipelineId); + case 'pipeline_cancel': + return handlePipelineCancel(pipelineId); + case 'pipeline_abort': + return handlePipelineAbort(pipelineId); + case 'pipeline_retry': + return handlePipelineRetry(pipelineId); + default: + logger.warn({ actionType }, 'Unknown card action'); + return {}; + } }); return handler; } +async function handlePipelineConfirm(pipelineId: string): Promise> { + const record = pipelineStore.get(pipelineId); + if (!record) return {}; + + // 同步执行 CAS，确保只在转换成功后才返回进度卡片 + // 避免 CAS 失败时用户看到卡住的进度卡片 + if (!pipelineStore.tryStart(pipelineId)) { + // 已经被处理过（double-click 或并发取消） + return {}; + } + + // CAS 成功，在后台启动管道（startPipeline 会跳过自身的 tryStart） + startPipeline(pipelineId).catch((err) => { + logger.error({ err, pipelineId }, 'Failed to start pipeline'); + }); + + // 立即返回初始进度卡片 + return buildPipelineCard(record.prompt, 'plan', 1, 5, 0, undefined, undefined, pipelineId); +} + +async function handlePipelineCancel(pipelineId: string): Promise> { + const record = pipelineStore.get(pipelineId); + if (!record) return {}; + + cancelPipeline(pipelineId); + return buildCancelledCard(record.prompt); +} + +async function handlePipelineAbort(pipelineId: string): Promise> { + abortPipeline(pipelineId); + // 不立即替换卡片 — orchestrator 的 onPhaseChange 会在最终状态时更新 + return {}; +} + +async function handlePipelineRetry(pipelineId: string): Promise> { + const record = pipelineStore.get(pipelineId); + if (!record) return {}; + + const newId = await retryPipeline(pipelineId); + if (!newId) return {}; + + const newRecord = pipelineStore.get(newId); + if (!newRecord) return {}; + + return buildPipelineConfirmCard(newRecord.prompt, newId, newRecord.workingDir); +} + // ============================================================ // 队列驱动：确保同一 chat 的 query 串行执行 // ============================================================ @@ -344,7 +432,7 @@ async function handleSlashCommand( return true; } - // /dev - 自动开发管道（绕过 taskQueue，使用 acquireSession 并发保护） + // /dev - 自动开发管道 if (trimmed.startsWith('/dev ')) { const task = trimmed.slice('/dev '.length).trim(); if (!task) { @@ -394,62 +482,6 @@ async function handleSlashCommand( return false; } -/** - * 确保会话有话题，如果没有则创建一个 - * 返回 threadRootMessageId (用于后续 reply_in_thread)，失败返回 undefined - */ -async function ensureThread( - chatId: string, - userId: string, - messageId: string, - rootId?: string, -): Promise { - sessionManager.getOrCreate(chatId, userId); - - // 1. 用户在已有话题内发消息 — 直接复用该话题，无需发送问候 - if (rootId) { - // 更新 session 的话题信息，确保后续回复也发到这个话题 - sessionManager.setThread(chatId, userId, rootId, rootId); - return rootId; - } - - // 2. 用户在主聊天区发消息（无 rootId）— 新会话意图 - // 如果想继续旧话题，用户应在话题内回复；在主区发消息 = 新对话 - const greeting = '🤖 新会话已创建'; - const { messageId: botMsgId, threadId } = await feishuClient.replyInThread( - messageId, - greeting, - ); - - if (threadId && botMsgId) { - // 话题创建成功后才清空旧 conversationId，避免 replyInThread 失败时 - // 既没有新话题又丢失了续接旧对话的能力 - sessionManager.setConversationId(chatId, userId, ''); - sessionManager.setThread(chatId, userId, threadId, messageId); - return messageId; - } - - logger.warn({ chatId, userId }, 'Failed to create thread, falling back to main chat'); - return undefined; -} - -/** - * 并发保护：原子地尝试获取会话锁（CAS: idle → busy） - * @returns true 如果成功获取锁，false 如果已被占用 - */ -async function acquireSession( - chatId: string, - userId: string, - messageId: string, -): Promise { - // 原子 CAS：UPDATE ... WHERE status != 'busy'，单条 SQL 防止 TOCTOU 竞态 - if (!sessionManager.tryAcquire(chatId, userId)) { - await feishuClient.replyText(messageId, '⏳ 当前会话正在执行任务，请等待完成或使用 /stop 中断'); - return false; - } - return true; -} - /** * 执行 Claude Agent SDK 任务 * 支持 workspace 变更后自动 restart：第一次 query 触发 setup_workspace 后， @@ -671,7 +703,7 @@ async function sendResultCard( /** * 执行自动开发管道（/dev 命令触发） - * 绕过 taskQueue，使用 acquireSession 进行并发保护 + * 创建待确认管道，等待用户卡片确认后再开始执行 */ async function executePipelineTask( prompt: string, @@ -680,136 +712,15 @@ async function executePipelineTask( messageId: string, rootId?: string, ): Promise { - if (!await acquireSession(chatId, userId, messageId)) return; - const session = sessionManager.getOrCreate(chatId, userId); - const pipelineStartTime = Date.now(); - - // 确保话题存在 - const threadRootMsgId = await ensureThread(chatId, userId, messageId, rootId); - - // 发送管道初始卡片 - let progressMsgId: string | undefined; - const initialCard = buildPipelineCard(prompt, 'plan', 1, TOTAL_PHASES, 0); - if (threadRootMsgId) { - progressMsgId = await feishuClient.replyCardInThread(threadRootMsgId, initialCard); - } - if (!progressMsgId) { - progressMsgId = await feishuClient.sendCard(chatId, initialCard); - } - - // 获取历史摘要 - const summaries = sessionManager.getRecentSummaries(chatId, userId, 5); - let historySummaries: string | undefined; - if (summaries.length > 0) { - let combined = summaries.join('\n'); - if (combined.length > 3000) { - combined = combined.slice(-3000); - } - historySummaries = combined; - } - - try { - const orchestrator = new PipelineOrchestrator(); - - // 跟踪当前 phase 供 onStreamUpdate 使用 - let currentPipelinePhase: string = 'plan'; - let currentPhaseIndex = 1; - - const pipelineResult = await orchestrator.run( - prompt, - session.workingDir, - { - onPhaseChange: async (state) => { - currentPipelinePhase = state.phase; - currentPhaseIndex = PHASE_META[state.phase]?.index ?? currentPhaseIndex; - if (!progressMsgId) return; - const elapsed = Math.floor((Date.now() - pipelineStartTime) / 1000); - await feishuClient.updateCard( - progressMsgId, - buildPipelineCard( - prompt, - state.phase, - currentPhaseIndex, - TOTAL_PHASES, - elapsed, - state.totalCostUsd || undefined, - ), - ); - }, - onStreamUpdate: async (text: string) => { - if (!progressMsgId) return; - const elapsed = Math.floor((Date.now() - pipelineStartTime) / 1000); - // 使用 pipeline 卡片 + detail 区域展示流式输出，保留阶段进度 - const tail = text.length > 2000 ? '...\n' + text.slice(-2000) : text; - await feishuClient.updateCard( - progressMsgId, - buildPipelineCard(prompt, currentPipelinePhase, currentPhaseIndex, TOTAL_PHASES, elapsed, undefined, tail), - ); - }, - }, - historySummaries, - ); - - // 最终结果卡片 - const totalElapsed = Math.floor((Date.now() - pipelineStartTime) / 1000); - const costStr = pipelineResult.totalCostUsd - ? ` | 💰 $${pipelineResult.totalCostUsd.toFixed(4)}` - : ''; - - // 失败时用 failedAtPhase 定位实际失败的阶段 - const failedIndex = pipelineResult.state.failedAtPhase - ? PHASE_META[pipelineResult.state.failedAtPhase]?.index ?? TOTAL_PHASES - : TOTAL_PHASES; - - const finalCard = buildPipelineCard( - prompt, - pipelineResult.success ? 'done' : 'failed', - pipelineResult.success ? TOTAL_PHASES + 1 : failedIndex, - TOTAL_PHASES, - totalElapsed, - pipelineResult.totalCostUsd || undefined, - pipelineResult.summary.slice(0, 2500), - ); - - if (progressMsgId) { - await feishuClient.updateCard(progressMsgId, finalCard); - } else if (threadRootMsgId) { - await feishuClient.replyCardInThread(threadRootMsgId, finalCard); - } else { - await feishuClient.sendCard(chatId, finalCard); - } - - // 如果摘要太长，额外发送完整文本 - if (pipelineResult.summary.length > 2500) { - if (threadRootMsgId) { - await feishuClient.replyTextInThread(threadRootMsgId, pipelineResult.summary); - } else { - await feishuClient.sendText(chatId, pipelineResult.summary); - } - } - - // 保存摘要 - if (pipelineResult.summary.length > 100) { - try { - const date = new Date().toISOString().slice(0, 10); - const tail = pipelineResult.summary.slice(-500).trim(); - const summary = `[${date}] [pipeline] dir: ${session.workingDir} | ${tail}`; - sessionManager.saveSummary(chatId, userId, session.workingDir, summary); - } catch (err) { - logger.warn({ err }, 'Failed to save pipeline summary'); - } - } - } catch (err) { - logger.error({ err }, 'Error executing pipeline'); - await feishuClient.replyText(messageId, `❌ 管道执行出错: ${(err as Error).message}`); - } finally { - try { - sessionManager.setStatus(chatId, userId, 'idle'); - } catch (err) { - logger.error({ err, chatId, userId }, 'Failed to reset session status'); - } - } + await createPendingPipeline({ + chatId, + userId, + messageId, + rootId, + prompt, + workingDir: session.workingDir, + }); } /** diff --git a/src/feishu/message-builder.ts b/src/feishu/message-builder.ts index 3e093737..0d28cfe5 100644 --- a/src/feishu/message-builder.ts +++ b/src/feishu/message-builder.ts @@ -144,6 +144,7 @@ export function buildPipelineCard( elapsedSec: number, costUsd?: number, detail?: string, + pipelineId?: string, ): Record { const isDone = phase === 'done'; @@ -201,6 +202,43 @@ export function buildPipelineCard( }); } + // 交互按钮 + if (pipelineId) { + if (!isDone && !isFailed) { + // 执行中：添加「中止」按钮 + elements.push({ tag: 'hr' }); + elements.push({ + tag: 'action', + actions: [ + { + tag: 'button', + text: { tag: 'plain_text', content: '🛑 中止' }, + type: 'danger', + confirm: { + title: { tag: 'plain_text', content: '确认中止' }, + text: { tag: 'plain_text', content: '中止后当前阶段将运行至结束，但不会进入下一阶段。确定要中止吗？' }, + }, + value: { action: 'pipeline_abort', pipelineId }, + }, + ], + }); + } else if (isFailed) { + // 失败：添加「重试」按钮 + elements.push({ tag: 'hr' }); + elements.push({ + tag: 'action', + actions: [ + { + tag: 'button', + text: { tag: 'plain_text', content: '🔄 重试' }, + type: 'primary', + value: { action: 'pipeline_retry', pipelineId }, + }, + ], + }); + } + } + elements.push({ tag: 'hr' }); const costStr = costUsd ? ` | 💰 $${costUsd.toFixed(4)}` : ''; @@ -259,6 +297,130 @@ export function buildStatusCard( }; } +/** 构建管道确认卡片（/dev 命令后等待用户确认） */ +export function buildPipelineConfirmCard( + prompt: string, + pipelineId: string, + workingDir: string, +): Record { + return { + config: { wide_screen_mode: true }, + header: { + title: { tag: 'plain_text', content: '🤖 Claude Code - 自动开发管道' }, + template: 'blue', + }, + elements: [ + { + tag: 'div', + text: { + tag: 'lark_md', + content: `**任务:** ${escapeMarkdown(truncate(prompt, 300))}`, + }, + }, + { tag: 'hr' }, + { + tag: 'div', + text: { + tag: 'lark_md', + content: [ + `**工作目录:** ${workingDir}`, + `**预估查询数:** ~9 次 (最多 17 次，含重试)`, + `**流程:** 方案设计 → 方案审查 → 代码实现 → 代码审查 → 推送 PR`, + ].join('\n'), + }, + }, + { tag: 'hr' }, + { + tag: 'action', + actions: [ + { + tag: 'button', + text: { tag: 'plain_text', content: '✅ 确认执行' }, + type: 'primary', + value: { action: 'pipeline_confirm', pipelineId }, + }, + { + tag: 'button', + text: { tag: 'plain_text', content: '❌ 取消' }, + type: 'default', + value: { action: 'pipeline_cancel', pipelineId }, + }, + ], + }, + ], + }; +} + +/** 构建管道已取消卡片 */ +export function buildCancelledCard(prompt: string): Record { + return { + config: { wide_screen_mode: true }, + header: { + title: { tag: 'plain_text', content: '🤖 Claude Code - 已取消' }, + template: 'grey', + }, + elements: [ + { + tag: 'div', + text: { + tag: 'lark_md', + content: `**指令:** ${escapeMarkdown(truncate(prompt, 200))}`, + }, + }, + { tag: 'hr' }, + { + tag: 'note', + elements: [ + { tag: 'plain_text', content: '❌ 用户已取消' }, + ], + }, + ], + }; +} + +/** 构建管道中断卡片（服务重启导致中断） */ +export function buildInterruptedCard( + prompt: string, + pipelineId: string, +): Record { + return { + config: { wide_screen_mode: true }, + header: { + title: { tag: 'plain_text', content: '🤖 Claude Code - 管道中断' }, + template: 'orange', + }, + elements: [ + { + tag: 'div', + text: { + tag: 'lark_md', + content: `**指令:** ${escapeMarkdown(truncate(prompt, 200))}`, + }, + }, + { tag: 'hr' }, + { + tag: 'div', + text: { + tag: 'lark_md', + content: '⚠️ 服务重启，管道已中断', + }, + }, + { tag: 'hr' }, + { + tag: 'action', + actions: [ + { + tag: 'button', + text: { tag: 'plain_text', content: '🔄 重试' }, + type: 'primary', + value: { action: 'pipeline_retry', pipelineId }, + }, + ], + }, + ], + }; +} + // === 工具函数 === function escapeMarkdown(text: string): string { diff --git a/src/feishu/thread-utils.ts b/src/feishu/thread-utils.ts new file mode 100644 index 00000000..bc7efb26 --- /dev/null +++ b/src/feishu/thread-utils.ts @@ -0,0 +1,42 @@ +import { logger } from '../utils/logger.js'; +import { sessionManager } from '../session/manager.js'; +import { feishuClient } from './client.js'; + +/** + * 确保会话有话题，如果没有则创建一个 + * 返回 threadRootMessageId (用于后续 reply_in_thread)，失败返回 undefined + */ +export async function ensureThread( + chatId: string, + userId: string, + messageId: string, + rootId?: string, +): Promise { + sessionManager.getOrCreate(chatId, userId); + + // 1. 用户在已有话题内发消息 — 直接复用该话题，无需发送问候 + if (rootId) { + // 更新 session 的话题信息，确保后续回复也发到这个话题 + sessionManager.setThread(chatId, userId, rootId, rootId); + return rootId; + } + + // 2. 用户在主聊天区发消息（无 rootId）— 新会话意图 + // 如果想继续旧话题，用户应在话题内回复；在主区发消息 = 新对话 + const greeting = '🤖 新会话已创建'; + const { messageId: botMsgId, threadId } = await feishuClient.replyInThread( + messageId, + greeting, + ); + + if (threadId && botMsgId) { + // 话题创建成功后才清空旧 conversationId，避免 replyInThread 失败时 + // 既没有新话题又丢失了续接旧对话的能力 + sessionManager.setConversationId(chatId, userId, ''); + sessionManager.setThread(chatId, userId, threadId, messageId); + return messageId; + } + + logger.warn({ chatId, userId }, 'Failed to create thread, falling back to main chat'); + return undefined; +} diff --git a/src/index.ts b/src/index.ts index b6ef39a9..b09c39ff 100644 --- a/src/index.ts +++ b/src/index.ts @@ -4,6 +4,8 @@ import { startServer } from './server.js'; import { sessionManager } from './session/manager.js'; import { claudeExecutor } from './claude/executor.js'; import { cleanupTmpDirs, cleanupExpiredCaches } from './workspace/cache.js'; +import { pipelineStore } from './pipeline/store.js'; +import { recoverInterruptedPipelines } from './pipeline/runner.js'; function main(): void { logger.info('Starting Feishu Claude Code Bridge...'); @@ -29,17 +31,25 @@ function main(): void { // 启动 HTTP 服务 startServer(); - // 定时清理过期会话、Claude Code 进程和缓存 (每 30 分钟) + // 恢复被中断的管道（服务重启后通知用户） + recoverInterruptedPipelines().catch((err) => { + logger.error({ err }, 'Failed to recover interrupted pipelines'); + }); + + // 定时清理过期会话、Claude Code 进程、缓存和管道记录 (每 30 分钟) setInterval(() => { sessionManager.cleanup(); claudeExecutor.cleanup(); cleanupExpiredCaches(); + pipelineStore.cleanExpired(30); }, 30 * 60 * 1000); // 优雅退出 process.on('SIGINT', () => { logger.info('Received SIGINT, shutting down...'); claudeExecutor.killAll(); + pipelineStore.markRunningAsInterrupted(); + pipelineStore.close(); sessionManager.close(); process.exit(0); }); @@ -47,6 +57,8 @@ function main(): void { process.on('SIGTERM', () => { logger.info('Received SIGTERM, shutting down...'); claudeExecutor.killAll(); + pipelineStore.markRunningAsInterrupted(); + pipelineStore.close(); sessionManager.close(); process.exit(0); }); diff --git a/src/pipeline/__tests__/orchestrator.test.ts b/src/pipeline/__tests__/orchestrator.test.ts index 81ff654d..fa7373f4 100644 --- a/src/pipeline/__tests__/orchestrator.test.ts +++ b/src/pipeline/__tests__/orchestrator.test.ts @@ -584,6 +584,75 @@ describe('PipelineOrchestrator', () => { }); }); + // ============================================================ + // Abort 支持 + // ============================================================ + + describe('abort', () => { + it('should stop pipeline when abort is called before first phase', async () => { + const orchestrator = new PipelineOrchestrator(); + orchestrator.abort(); + + const result = await orchestrator.run('task', '/tmp', noopCallbacks); + + expect(result.success).toBe(false); + expect(result.state.phase).toBe('failed'); + expect(result.state.failureReason).toBe('用户手动中止'); + expect(mockExecute).not.toHaveBeenCalled(); + }); + + it('should stop pipeline between phases when abort is called', async () => { + // Plan executes, plan_review passes, then abort before implement runs + mockExecute.mockResolvedValueOnce(makeResult({ output: 'plan' })); + mockParallelReview.mockResolvedValueOnce(makeReviewResult()); + + const orch = new PipelineOrchestrator(); + + // Abort right when we see implement about to start + // onPhaseChange is called BEFORE the phase executes + const onPhaseChange = vi.fn().mockImplementation(async (state: { phase: string }) => { + if (state.phase === 'implement') { + orch.abort(); + } + }); + + // Need to provide a mock for implement in case abort timing doesn't prevent it + mockExecute.mockResolvedValueOnce(makeResult({ output: 'impl' })); + + const result = await orch.run('task', '/tmp', { onPhaseChange }); + + expect(result.success).toBe(false); + expect(result.state.phase).toBe('failed'); + expect(result.state.failureReason).toBe('用户手动中止'); + }); + + it('should expose current session key via getCurrentSessionKey()', async () => { + let capturedKey: string | undefined; + + mockExecute.mockImplementation(async (opts) => { + capturedKey = opts.sessionKey; + return makeResult({ output: 'plan' }); + }); + + mockParallelReview + .mockResolvedValueOnce(makeReviewResult()) + .mockResolvedValueOnce(makeReviewResult()); + + mockExecute + .mockResolvedValueOnce(makeResult({ output: 'plan' })) + .mockResolvedValueOnce(makeResult({ output: 'impl' })) + .mockResolvedValueOnce(makeResult({ output: 'pushed' })); + + const orchestrator = new PipelineOrchestrator(); + await orchestrator.run('task', '/tmp', noopCallbacks); + + // getCurrentSessionKey should return the last session key used + const key = orchestrator.getCurrentSessionKey(); + expect(key).toBeDefined(); + expect(key).toContain('pipeline-'); + }); + }); + // ============================================================ // MAX_ITERATIONS 循环保护 // ============================================================ diff --git a/src/pipeline/__tests__/runner.test.ts b/src/pipeline/__tests__/runner.test.ts new file mode 100644 index 00000000..e2be6234 --- /dev/null +++ b/src/pipeline/__tests__/runner.test.ts @@ -0,0 +1,171 @@ +import { describe, it, expect, vi, afterEach, afterAll, beforeAll } from 'vitest'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; +import type { PipelineStore } from '../store.js'; + +// Mock all external dependencies +vi.mock('../../utils/logger.js', () => ({ + logger: { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + }, +})); + +vi.mock('../../config.js', () => ({ + config: { + db: { sessionDbPath: '/tmp/test-runner.db', pipelineDbPath: '/tmp/test-runner-pipeline.db' }, + claude: { defaultWorkDir: '/tmp' }, + }, +})); + +vi.mock('../../feishu/client.js', () => ({ + feishuClient: { + sendCard: vi.fn().mockResolvedValue('card_msg_1'), + updateCard: vi.fn().mockResolvedValue(undefined), + replyInThread: vi.fn().mockResolvedValue({ messageId: 'bot_msg_1', threadId: 'thread_1' }), + replyCardInThread: vi.fn().mockResolvedValue('card_msg_2'), + replyTextInThread: vi.fn().mockResolvedValue(undefined), + sendText: vi.fn().mockResolvedValue(undefined), + replyText: vi.fn().mockResolvedValue(undefined), + }, +})); + +vi.mock('../../session/manager.js', () => ({ + sessionManager: { + getOrCreate: vi.fn().mockReturnValue({ + chatId: 'chat1', + userId: 'user1', + workingDir: '/tmp/work', + status: 'idle', + }), + get: vi.fn().mockReturnValue(null), + setThread: vi.fn(), + setConversationId: vi.fn(), + setStatus: vi.fn(), + tryAcquire: vi.fn().mockReturnValue(true), + getRecentSummaries: vi.fn().mockReturnValue([]), + saveSummary: vi.fn(), + setWorkingDir: vi.fn(), + }, +})); + +vi.mock('../../claude/executor.js', () => ({ + claudeExecutor: { + execute: vi.fn(), + killSession: vi.fn(), + }, +})); + +vi.mock('../reviewer.js', () => ({ + parallelReview: vi.fn(), +})); + +vi.mock('../../feishu/thread-utils.js', () => ({ + ensureThread: vi.fn().mockResolvedValue('root1'), +})); + +// Use a dynamic import pattern to handle the store mock properly +const _tempDir = mkdtempSync(join(tmpdir(), 'runner-test-')); +const _storeDbPath = join(_tempDir, 'test.db'); + +// Mock store with a factory that creates its own instance +vi.mock('../store.js', async (importOriginal) => { + const mod = await importOriginal(); + // Create a fresh temp directory for the store inside the factory + const tempDirInner = mkdtempSync(join(tmpdir(), 'runner-store-')); + const store = new mod.PipelineStore(join(tempDirInner, 'test.db')); + return { + ...mod, + pipelineStore: store, + // Store tempDir for cleanup + __tempDir: tempDirInner, + }; +}); + +// Import after mocks are set up +const { createPendingPipeline, cancelPipeline } = await import('../runner.js'); +const storeModule = await import('../store.js') as typeof import('../store.js') & { __tempDir: string }; +const { pipelineStore } = storeModule; +const { feishuClient } = await import('../../feishu/client.js'); + +describe('Pipeline Runner', () => { + afterEach(() => { + vi.clearAllMocks(); + }); + + afterAll(() => { + pipelineStore.close(); + rmSync(storeModule.__tempDir, { recursive: true, force: true }); + rmSync(_tempDir, { recursive: true, force: true }); + }); + + describe('createPendingPipeline', () => { + it('should create a pipeline record in pending_confirm status', async () => { + const pipelineId = await createPendingPipeline({ + chatId: 'chat1', + userId: 'user1', + messageId: 'msg1', + rootId: 'root1', + prompt: 'build a feature', + workingDir: '/tmp/work', + }); + + expect(pipelineId).toMatch(/^pipe_/); + + const record = pipelineStore.get(pipelineId); + expect(record).toBeDefined(); + expect(record!.status).toBe('pending_confirm'); + expect(record!.prompt).toBe('build a feature'); + expect(record!.workingDir).toBe('/tmp/work'); + }); + + it('should send a confirmation card', async () => { + await createPendingPipeline({ + chatId: 'chat1', + userId: 'user1', + messageId: 'msg1', + rootId: 'root1', + prompt: 'task', + workingDir: '/tmp', + }); + + expect(feishuClient.replyCardInThread).toHaveBeenCalled(); + }); + }); + + describe('cancelPipeline', () => { + it('should cancel a pending_confirm pipeline', async () => { + const pipelineId = await createPendingPipeline({ + chatId: 'chat1', + userId: 'user1', + messageId: 'msg1', + prompt: 'task', + workingDir: '/tmp', + }); + + const result = cancelPipeline(pipelineId); + expect(result).toBe(true); + + const record = pipelineStore.get(pipelineId); + expect(record!.status).toBe('cancelled'); + }); + + it('should not cancel a non-pending pipeline', async () => { + const pipelineId = await createPendingPipeline({ + chatId: 'chat1', + userId: 'user1', + messageId: 'msg1', + prompt: 'task', + workingDir: '/tmp', + }); + + pipelineStore.tryStart(pipelineId); + + const result = cancelPipeline(pipelineId); + expect(result).toBe(false); + }); + }); +}); diff --git a/src/pipeline/__tests__/store.test.ts b/src/pipeline/__tests__/store.test.ts new file mode 100644 index 00000000..b1852af1 --- /dev/null +++ b/src/pipeline/__tests__/store.test.ts @@ -0,0 +1,291 @@ +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; + +vi.mock('../../utils/logger.js', () => ({ + logger: { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + }, +})); + +vi.mock('../../config.js', () => ({ + config: { + db: { sessionDbPath: '/tmp/test.db', pipelineDbPath: '/tmp/test-pipeline.db' }, + }, +})); + +import { PipelineStore, generatePipelineId } from '../store.js'; + +describe('PipelineStore', () => { + let store: PipelineStore; + let tempDir: string; + + beforeEach(() => { + tempDir = mkdtempSync(join(tmpdir(), 'pipeline-store-test-')); + store = new PipelineStore(join(tempDir, 'test.db')); + }); + + afterEach(() => { + store.close(); + rmSync(tempDir, { recursive: true, force: true }); + }); + + // ============================================================ + // CRUD + // ============================================================ + + describe('create and get', () => { + it('should create and retrieve a pipeline record', () => { + const record = store.create({ + id: 'pipe_123_abcd', + chatId: 'chat1', + userId: 'user1', + messageId: 'msg1', + threadRootMsgId: 'root1', + progressMsgId: 'progress1', + workingDir: '/tmp/work', + prompt: 'build a feature', + }); + + expect(record.id).toBe('pipe_123_abcd'); + expect(record.status).toBe('pending_confirm'); + expect(record.phase).toBe(''); + expect(record.stateJson).toBe('{}'); + + const retrieved = store.get('pipe_123_abcd'); + expect(retrieved).toBeDefined(); + expect(retrieved!.chatId).toBe('chat1'); + expect(retrieved!.userId).toBe('user1'); + expect(retrieved!.prompt).toBe('build a feature'); + expect(retrieved!.workingDir).toBe('/tmp/work'); + expect(retrieved!.threadRootMsgId).toBe('root1'); + expect(retrieved!.progressMsgId).toBe('progress1'); + }); + + it('should return undefined for non-existent pipeline', () => { + expect(store.get('nonexistent')).toBeUndefined(); + }); + + it('should handle null optional fields', () => { + store.create({ + id: 'pipe_456', + chatId: 'chat1', + userId: 'user1', + messageId: 'msg1', + workingDir: '/tmp', + prompt: 'task', + }); + + const record = store.get('pipe_456'); + expect(record!.threadRootMsgId).toBeUndefined(); + expect(record!.progressMsgId).toBeUndefined(); + }); + }); + + // ============================================================ + // tryStart — CAS atomicity + // ============================================================ + + describe('tryStart (CAS)', () => { + it('should succeed on first call', () => { + store.create({ + id: 'pipe_cas', + chatId: 'chat1', + userId: 'user1', + messageId: 'msg1', + workingDir: '/tmp', + prompt: 'task', + }); + + const result = store.tryStart('pipe_cas'); + expect(result).toBe(true); + + const record = store.get('pipe_cas'); + expect(record!.status).toBe('running'); + }); + + it('should fail on second call (double-click prevention)', () => { + store.create({ + id: 'pipe_cas2', + chatId: 'chat1', + userId: 'user1', + messageId: 'msg1', + workingDir: '/tmp', + prompt: 'task', + }); + + expect(store.tryStart('pipe_cas2')).toBe(true); + expect(store.tryStart('pipe_cas2')).toBe(false); + }); + + it('should fail if pipeline is already in non-pending state', () => { + store.create({ + id: 'pipe_cas3', + chatId: 'chat1', + userId: 'user1', + messageId: 'msg1', + workingDir: '/tmp', + prompt: 'task', + }); + + store.updateState('pipe_cas3', 'cancelled', '', '{}'); + expect(store.tryStart('pipe_cas3')).toBe(false); + }); + }); + + // ============================================================ + // updateState + // ============================================================ + + describe('updateState', () => { + it('should update status, phase, and stateJson', () => { + store.create({ + id: 'pipe_update', + chatId: 'chat1', + userId: 'user1', + messageId: 'msg1', + workingDir: '/tmp', + prompt: 'task', + }); + + store.updateState('pipe_update', 'running', 'plan', '{"phase":"plan"}'); + + const record = store.get('pipe_update'); + expect(record!.status).toBe('running'); + expect(record!.phase).toBe('plan'); + expect(record!.stateJson).toBe('{"phase":"plan"}'); + }); + }); + + // ============================================================ + // updateProgressMsgId + // ============================================================ + + describe('updateProgressMsgId', () => { + it('should update the progress message ID', () => { + store.create({ + id: 'pipe_msg', + chatId: 'chat1', + userId: 'user1', + messageId: 'msg1', + workingDir: '/tmp', + prompt: 'task', + }); + + store.updateProgressMsgId('pipe_msg', 'new_progress_msg'); + + const record = store.get('pipe_msg'); + expect(record!.progressMsgId).toBe('new_progress_msg'); + }); + }); + + // ============================================================ + // findByStatus + // ============================================================ + + describe('findByStatus', () => { + it('should find pipelines by status', () => { + store.create({ id: 'p1', chatId: 'c', userId: 'u', messageId: 'm', workingDir: '/tmp', prompt: 'a' }); + store.create({ id: 'p2', chatId: 'c', userId: 'u', messageId: 'm', workingDir: '/tmp', prompt: 'b' }); + + store.tryStart('p1'); + + const pending = store.findByStatus('pending_confirm'); + expect(pending).toHaveLength(1); + expect(pending[0].id).toBe('p2'); + + const running = store.findByStatus('running'); + expect(running).toHaveLength(1); + expect(running[0].id).toBe('p1'); + }); + }); + + // ============================================================ + // markRunningAsInterrupted + // ============================================================ + + describe('markRunningAsInterrupted', () => { + it('should mark running pipelines as interrupted', () => { + store.create({ id: 'p1', chatId: 'c', userId: 'u', messageId: 'm', workingDir: '/tmp', prompt: 'a' }); + store.create({ id: 'p2', chatId: 'c', userId: 'u', messageId: 'm', workingDir: '/tmp', prompt: 'b' }); + + store.tryStart('p1'); + store.tryStart('p2'); + + const count = store.markRunningAsInterrupted(); + expect(count).toBe(2); + + const interrupted = store.findByStatus('interrupted'); + expect(interrupted).toHaveLength(2); + + const running = store.findByStatus('running'); + expect(running).toHaveLength(0); + }); + + it('should not affect non-running pipelines', () => { + store.create({ id: 'p1', chatId: 'c', userId: 'u', messageId: 'm', workingDir: '/tmp', prompt: 'a' }); + store.create({ id: 'p2', chatId: 'c', userId: 'u', messageId: 'm', workingDir: '/tmp', prompt: 'b' }); + + store.tryStart('p1'); + // p2 stays in pending_confirm + + const count = store.markRunningAsInterrupted(); + expect(count).toBe(1); + + const pending = store.findByStatus('pending_confirm'); + expect(pending).toHaveLength(1); + expect(pending[0].id).toBe('p2'); + }); + + it('should return 0 when no running pipelines', () => { + expect(store.markRunningAsInterrupted()).toBe(0); + }); + }); + + // ============================================================ + // cleanExpired + // ============================================================ + + describe('cleanExpired', () => { + it('should not clean recent pipelines', () => { + store.create({ id: 'p1', chatId: 'c', userId: 'u', messageId: 'm', workingDir: '/tmp', prompt: 'a' }); + + const cleaned = store.cleanExpired(30); + expect(cleaned).toBe(0); + }); + + it('should clean old pipelines', () => { + store.create({ id: 'p1', chatId: 'c', userId: 'u', messageId: 'm', workingDir: '/tmp', prompt: 'a' }); + + // Back-date via raw SQL + const oldDate = new Date(Date.now() - 31 * 24 * 60 * 60 * 1000).toISOString(); + (store as any).db.prepare( + "UPDATE pipelines SET created_at = ? WHERE id = 'p1'" + ).run(oldDate); + + const cleaned = store.cleanExpired(30); + expect(cleaned).toBe(1); + expect(store.get('p1')).toBeUndefined(); + }); + }); + + // ============================================================ + // generatePipelineId + // ============================================================ + + describe('generatePipelineId', () => { + it('should generate IDs with pipe_ prefix', () => { + const id = generatePipelineId(); + expect(id).toMatch(/^pipe_\d+_[0-9a-f]{8}$/); + }); + + it('should generate unique IDs', () => { + const ids = new Set(Array.from({ length: 100 }, () => generatePipelineId())); + expect(ids.size).toBe(100); + }); + }); +}); diff --git a/src/pipeline/orchestrator.ts b/src/pipeline/orchestrator.ts index a598ceeb..96818eed 100644 --- a/src/pipeline/orchestrator.ts +++ b/src/pipeline/orchestrator.ts @@ -25,6 +25,30 @@ import { parallelReview } from './reviewer.js'; const MAX_RETRIES = 2; export class PipelineOrchestrator { + private aborted = false; + private currentSessionKey?: string; + + /** + * 中止管道 — 设置标志，阻止下一阶段启动 + */ + abort(): void { + this.aborted = true; + } + + /** + * 是否已被中止 + */ + isAborted(): boolean { + return this.aborted; + } + + /** + * 获取当前正在执行的 session key（用于外部 kill） + */ + getCurrentSessionKey(): string | undefined { + return this.currentSessionKey; + } + /** * 执行完整管道 */ @@ -52,6 +76,12 @@ export class PipelineOrchestrator { let iterations = 0; while (state.phase !== 'done' && state.phase !== 'failed') { + // 中止检查：用户手动中止 + if (this.aborted) { + state = { ...state, phase: 'failed', failedAtPhase: state.phase, failureReason: '用户手动中止' }; + break; + } + if (++iterations > MAX_ITERATIONS) { state = { ...state, phase: 'failed', failedAtPhase: state.phase, failureReason: '管道超过最大迭代次数，可能存在循环' }; break; @@ -361,6 +391,7 @@ export class PipelineOrchestrator { historySummaries?: string, onStreamUpdate?: (text: string) => Promise, ): Promise { + this.currentSessionKey = sessionKey; return claudeExecutor.execute({ sessionKey, prompt, diff --git a/src/pipeline/runner.ts b/src/pipeline/runner.ts new file mode 100644 index 00000000..e5c22560 --- /dev/null +++ b/src/pipeline/runner.ts @@ -0,0 +1,325 @@ +import { logger } from '../utils/logger.js'; +import { feishuClient } from '../feishu/client.js'; +import { sessionManager } from '../session/manager.js'; +import { claudeExecutor } from '../claude/executor.js'; +import { PipelineOrchestrator } from './orchestrator.js'; +import { pipelineStore, generatePipelineId } from './store.js'; +import { PHASE_META, TOTAL_PHASES } from './types.js'; +import { + buildPipelineConfirmCard, + buildPipelineCard, + buildCancelledCard, + buildInterruptedCard, +} from '../feishu/message-builder.js'; + +// ============================================================ +// Pipeline Runner — 管道生命周期管理 +// +// 桥接 event-handler ↔ orchestrator ↔ store +// ============================================================ + +/** 正在运行的管道注册表 */ +const runningPipelines = new Map(); + +export interface CreatePipelineParams { + chatId: string; + userId: string; + messageId: string; + rootId?: string; + prompt: string; + workingDir: string; +} + +import { ensureThread } from '../feishu/thread-utils.js'; + +/** + * 创建待确认的管道（发送确认卡片，写入 store） + */ +export async function createPendingPipeline(params: CreatePipelineParams): Promise { + const { chatId, userId, messageId, rootId, prompt, workingDir } = params; + const pipelineId = generatePipelineId(); + + // 确保话题存在 + const threadRootMsgId = await ensureThread(chatId, userId, messageId, rootId); + + // 发送确认卡片 + const confirmCard = buildPipelineConfirmCard(prompt, pipelineId, workingDir); + let progressMsgId: string | undefined; + if (threadRootMsgId) { + progressMsgId = await feishuClient.replyCardInThread(threadRootMsgId, confirmCard); + } + if (!progressMsgId) { + progressMsgId = await feishuClient.sendCard(chatId, confirmCard); + } + + // 保存到 store + pipelineStore.create({ + id: pipelineId, + chatId, + userId, + messageId, + threadRootMsgId, + progressMsgId, + workingDir, + prompt, + }); + + logger.info({ pipelineId, chatId, userId }, 'Pending pipeline created'); + return pipelineId; +} + +/** + * 启动管道（确认后调用） + */ +export async function startPipeline(pipelineId: string): Promise { + const record = pipelineStore.get(pipelineId); + if (!record) { + logger.warn({ pipelineId }, 'Pipeline not found'); + return; + } + + // CAS 已在 handlePipelineConfirm 中完成（同步执行以确保卡片更新正确） + // 这里做防御性检查：如果状态不是 running，说明 CAS 未执行或被并发修改 + if (record.status !== 'running') { + // 兜底尝试：兼容直接调用 startPipeline 的场景 + if (!pipelineStore.tryStart(pipelineId)) { + logger.info({ pipelineId }, 'Pipeline already started or not in pending state'); + return; + } + } + + const { chatId, userId, prompt, workingDir, progressMsgId, threadRootMsgId } = record; + + // 获取会话锁 + if (!sessionManager.tryAcquire(chatId, userId)) { + pipelineStore.updateState(pipelineId, 'failed', '', JSON.stringify({ failureReason: '会话正忙' })); + if (progressMsgId) { + await feishuClient.updateCard(progressMsgId, buildPipelineCard( + prompt, 'failed', 1, TOTAL_PHASES, 0, undefined, '⚠️ 会话正忙，请等待当前任务完成', pipelineId, + )); + } + return; + } + + const pipelineStartTime = Date.now(); + + // 获取历史摘要 + const summaries = sessionManager.getRecentSummaries(chatId, userId, 5); + let historySummaries: string | undefined; + if (summaries.length > 0) { + let combined = summaries.join('\n'); + if (combined.length > 3000) { + combined = combined.slice(-3000); + } + historySummaries = combined; + } + + const orchestrator = new PipelineOrchestrator(); + runningPipelines.set(pipelineId, { orchestrator, chatId, userId }); + + let currentPipelinePhase = 'plan'; + let currentPhaseIndex = 1; + + try { + // 更新卡片为初始执行状态 + if (progressMsgId) { + await feishuClient.updateCard(progressMsgId, buildPipelineCard( + prompt, 'plan', 1, TOTAL_PHASES, 0, undefined, undefined, pipelineId, + )); + } + + const pipelineResult = await orchestrator.run( + prompt, + workingDir, + { + onPhaseChange: async (state) => { + currentPipelinePhase = state.phase; + currentPhaseIndex = PHASE_META[state.phase]?.index ?? currentPhaseIndex; + + // 同步到 store + pipelineStore.updateState(pipelineId, 'running', state.phase, JSON.stringify(state)); + + if (!progressMsgId) return; + const elapsed = Math.floor((Date.now() - pipelineStartTime) / 1000); + await feishuClient.updateCard( + progressMsgId, + buildPipelineCard( + prompt, state.phase, currentPhaseIndex, TOTAL_PHASES, + elapsed, state.totalCostUsd || undefined, undefined, pipelineId, + ), + ); + }, + onStreamUpdate: async (text: string) => { + if (!progressMsgId) return; + const elapsed = Math.floor((Date.now() - pipelineStartTime) / 1000); + const tail = text.length > 2000 ? '...\n' + text.slice(-2000) : text; + await feishuClient.updateCard( + progressMsgId, + buildPipelineCard( + prompt, currentPipelinePhase, currentPhaseIndex, TOTAL_PHASES, + elapsed, undefined, tail, pipelineId, + ), + ); + }, + }, + historySummaries, + ); + + // 更新最终状态（中止的管道保留 aborted 状态，不覆盖为 failed） + const finalStatus = orchestrator.isAborted() + ? 'aborted' as const + : pipelineResult.success ? 'done' as const : 'failed' as const; + pipelineStore.updateState(pipelineId, finalStatus, pipelineResult.state.phase, JSON.stringify(pipelineResult.state)); + + // 最终卡片 + const totalElapsed = Math.floor((Date.now() - pipelineStartTime) / 1000); + const failedIndex = pipelineResult.state.failedAtPhase + ? PHASE_META[pipelineResult.state.failedAtPhase]?.index ?? TOTAL_PHASES + : TOTAL_PHASES; + + const finalCard = buildPipelineCard( + prompt, + pipelineResult.success ? 'done' : 'failed', + pipelineResult.success ? TOTAL_PHASES + 1 : failedIndex, + TOTAL_PHASES, + totalElapsed, + pipelineResult.totalCostUsd || undefined, + pipelineResult.summary.slice(0, 2500), + pipelineId, + ); + + if (progressMsgId) { + await feishuClient.updateCard(progressMsgId, finalCard); + } else if (threadRootMsgId) { + await feishuClient.replyCardInThread(threadRootMsgId, finalCard); + } else { + await feishuClient.sendCard(chatId, finalCard); + } + + // 如果摘要太长，额外发送完整文本 + if (pipelineResult.summary.length > 2500) { + if (threadRootMsgId) { + await feishuClient.replyTextInThread(threadRootMsgId, pipelineResult.summary); + } else { + await feishuClient.sendText(chatId, pipelineResult.summary); + } + } + + // 保存摘要 + if (pipelineResult.summary.length > 100) { + try { + const date = new Date().toISOString().slice(0, 10); + const tail = pipelineResult.summary.slice(-500).trim(); + const summary = `[${date}] [pipeline] dir: ${workingDir} | ${tail}`; + sessionManager.saveSummary(chatId, userId, workingDir, summary); + } catch (err) { + logger.warn({ err }, 'Failed to save pipeline summary'); + } + } + } catch (err) { + logger.error({ err, pipelineId }, 'Error executing pipeline'); + pipelineStore.updateState(pipelineId, 'failed', '', JSON.stringify({ failureReason: String(err) })); + + if (progressMsgId) { + try { + const elapsed = Math.floor((Date.now() - pipelineStartTime) / 1000); + await feishuClient.updateCard(progressMsgId, buildPipelineCard( + prompt, 'failed', 1, TOTAL_PHASES, elapsed, undefined, + `❌ 管道执行出错: ${(err as Error).message}`, pipelineId, + )); + } catch (cardErr) { + logger.warn({ cardErr, pipelineId }, 'Failed to update error card'); + } + } + } finally { + runningPipelines.delete(pipelineId); + try { + sessionManager.setStatus(chatId, userId, 'idle'); + } catch (err) { + logger.error({ err, chatId, userId }, 'Failed to reset session status'); + } + } +} + +/** + * 中止管道 + */ +export function abortPipeline(pipelineId: string): boolean { + const entry = runningPipelines.get(pipelineId); + if (!entry) return false; + + entry.orchestrator.abort(); + + // 尝试 kill 当前 Claude session + const sessionKey = entry.orchestrator.getCurrentSessionKey(); + if (sessionKey) { + claudeExecutor.killSession(sessionKey); + } + + // 不在这里设置 aborted 状态 — startPipeline 的完成处理器会检查 + // orchestrator.isAborted() 来决定最终状态是 aborted 还是 failed + logger.info({ pipelineId }, 'Pipeline abort requested'); + return true; +} + +/** + * 取消管道（pending_confirm 状态） + */ +export function cancelPipeline(pipelineId: string): boolean { + const record = pipelineStore.get(pipelineId); + if (!record || record.status !== 'pending_confirm') return false; + + pipelineStore.updateState(pipelineId, 'cancelled', '', '{}'); + logger.info({ pipelineId }, 'Pipeline cancelled'); + return true; +} + +/** + * 重试管道 — 创建新的待确认管道 + */ +export async function retryPipeline(pipelineId: string): Promise { + const record = pipelineStore.get(pipelineId); + if (!record) return undefined; + + return createPendingPipeline({ + chatId: record.chatId, + userId: record.userId, + messageId: record.messageId, + rootId: record.threadRootMsgId, + prompt: record.prompt, + workingDir: record.workingDir, + }); +} + +/** + * 恢复被中断的管道（服务启动时调用） + */ +export async function recoverInterruptedPipelines(): Promise { + const count = pipelineStore.markRunningAsInterrupted(); + if (count === 0) return; + + const interrupted = pipelineStore.findByStatus('interrupted'); + logger.info({ count: interrupted.length }, 'Recovering interrupted pipelines'); + + for (const record of interrupted) { + if (!record.progressMsgId) continue; + + try { + const card = buildInterruptedCard(record.prompt, record.id); + await feishuClient.updateCard(record.progressMsgId, card); + } catch (err) { + logger.warn({ err, pipelineId: record.id }, 'Failed to update interrupted pipeline card'); + } + } +} + +/** + * 检查管道是否正在运行 + */ +export function isPipelineRunning(pipelineId: string): boolean { + return runningPipelines.has(pipelineId); +} diff --git a/src/pipeline/store.ts b/src/pipeline/store.ts new file mode 100644 index 00000000..7c11d670 --- /dev/null +++ b/src/pipeline/store.ts @@ -0,0 +1,233 @@ +import Database from 'better-sqlite3'; +import { randomBytes } from 'node:crypto'; +import { mkdirSync } from 'node:fs'; +import { dirname, resolve } from 'node:path'; +import { logger } from '../utils/logger.js'; +import { config } from '../config.js'; +import type { PipelineStatus } from './types.js'; + +// ============================================================ +// Pipeline 持久化存储 +// ============================================================ + +export interface PipelineRecord { + id: string; + chatId: string; + userId: string; + messageId: string; + threadRootMsgId?: string; + progressMsgId?: string; + workingDir: string; + prompt: string; + status: PipelineStatus; + phase: string; + stateJson: string; + createdAt: string; + updatedAt: string; +} + +interface PipelineRow { + id: string; + chat_id: string; + user_id: string; + message_id: string; + thread_root_msg_id: string | null; + progress_msg_id: string | null; + working_dir: string; + prompt: string; + status: string; + phase: string; + state_json: string; + created_at: string; + updated_at: string; +} + +const VALID_STATUSES = new Set([ + 'pending_confirm', 'running', 'done', 'failed', + 'aborted', 'interrupted', 'cancelled', +]); + +function validStatus(s: string): PipelineStatus { + return VALID_STATUSES.has(s) ? (s as PipelineStatus) : 'failed'; +} + +export class PipelineStore { + private db: Database.Database; + private stmtCreate: Database.Statement; + private stmtGet: Database.Statement; + private stmtTryStart: Database.Statement; + private stmtUpdateState: Database.Statement; + private stmtUpdateProgressMsgId: Database.Statement; + private stmtFindByStatus: Database.Statement; + private stmtMarkRunningAsInterrupted: Database.Statement; + private stmtCleanExpired: Database.Statement; + + constructor(dbPath: string) { + dbPath = resolve(dbPath); + mkdirSync(dirname(dbPath), { recursive: true, mode: 0o700 }); + + this.db = new Database(dbPath); + this.db.pragma('journal_mode = WAL'); + + this.db.exec(` + CREATE TABLE IF NOT EXISTS pipelines ( + id TEXT PRIMARY KEY, + chat_id TEXT NOT NULL, + user_id TEXT NOT NULL, + message_id TEXT NOT NULL, + thread_root_msg_id TEXT, + progress_msg_id TEXT, + working_dir TEXT NOT NULL, + prompt TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending_confirm', + phase TEXT NOT NULL DEFAULT '', + state_json TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + `); + + this.stmtCreate = this.db.prepare(` + INSERT INTO pipelines (id, chat_id, user_id, message_id, thread_root_msg_id, progress_msg_id, working_dir, prompt, status, phase, state_json, created_at, updated_at) + VALUES (@id, @chat_id, @user_id, @message_id, @thread_root_msg_id, @progress_msg_id, @working_dir, @prompt, @status, @phase, @state_json, @created_at, @updated_at) + `); + + this.stmtGet = this.db.prepare('SELECT * FROM pipelines WHERE id = ?'); + + // CAS: pending_confirm → running (atomic double-click prevention) + this.stmtTryStart = this.db.prepare(` + UPDATE pipelines SET status = 'running', updated_at = ? + WHERE id = ? AND status = 'pending_confirm' + `); + + this.stmtUpdateState = this.db.prepare(` + UPDATE pipelines SET status = ?, phase = ?, state_json = ?, updated_at = ? + WHERE id = ? + `); + + this.stmtUpdateProgressMsgId = this.db.prepare(` + UPDATE pipelines SET progress_msg_id = ?, updated_at = ? + WHERE id = ? + `); + + this.stmtFindByStatus = this.db.prepare( + 'SELECT * FROM pipelines WHERE status = ?', + ); + + this.stmtMarkRunningAsInterrupted = this.db.prepare(` + UPDATE pipelines SET status = 'interrupted', updated_at = ? + WHERE status = 'running' + `); + + this.stmtCleanExpired = this.db.prepare(` + DELETE FROM pipelines WHERE created_at < ? + `); + + logger.info({ dbPath }, 'Pipeline store initialized'); + } + + create(record: Omit): PipelineRecord { + const now = new Date().toISOString(); + const full: PipelineRecord = { + ...record, + status: 'pending_confirm', + phase: '', + stateJson: '{}', + createdAt: now, + updatedAt: now, + }; + + this.stmtCreate.run({ + id: full.id, + chat_id: full.chatId, + user_id: full.userId, + message_id: full.messageId, + thread_root_msg_id: full.threadRootMsgId ?? null, + progress_msg_id: full.progressMsgId ?? null, + working_dir: full.workingDir, + prompt: full.prompt, + status: full.status, + phase: full.phase, + state_json: full.stateJson, + created_at: full.createdAt, + updated_at: full.updatedAt, + }); + + return full; + } + + get(id: string): PipelineRecord | undefined { + const row = this.stmtGet.get(id) as PipelineRow | undefined; + if (!row) return undefined; + return this.rowToRecord(row); + } + + /** + * Atomic CAS: pending_confirm → running + * Returns true if transition succeeded, false if already started + */ + tryStart(id: string): boolean { + const result = this.stmtTryStart.run(new Date().toISOString(), id); + return result.changes === 1; + } + + updateState(id: string, status: PipelineStatus, phase: string, stateJson: string): void { + this.stmtUpdateState.run(status, phase, stateJson, new Date().toISOString(), id); + } + + updateProgressMsgId(id: string, msgId: string): void { + this.stmtUpdateProgressMsgId.run(msgId, new Date().toISOString(), id); + } + + findByStatus(status: PipelineStatus): PipelineRecord[] { + const rows = this.stmtFindByStatus.all(status) as PipelineRow[]; + return rows.map((r) => this.rowToRecord(r)); + } + + /** + * Mark all running pipelines as interrupted (for crash recovery) + * Returns the number of pipelines marked + */ + markRunningAsInterrupted(): number { + const result = this.stmtMarkRunningAsInterrupted.run(new Date().toISOString()); + if (result.changes > 0) { + logger.info({ count: result.changes }, 'Marked running pipelines as interrupted'); + } + return result.changes; + } + + cleanExpired(maxAgeDays: number): number { + const cutoff = new Date(Date.now() - maxAgeDays * 24 * 60 * 60 * 1000).toISOString(); + const result = this.stmtCleanExpired.run(cutoff); + return result.changes; + } + + close(): void { + this.db.close(); + logger.info('Pipeline store closed'); + } + + private rowToRecord(row: PipelineRow): PipelineRecord { + return { + id: row.id, + chatId: row.chat_id, + userId: row.user_id, + messageId: row.message_id, + threadRootMsgId: row.thread_root_msg_id ?? undefined, + progressMsgId: row.progress_msg_id ?? undefined, + workingDir: row.working_dir, + prompt: row.prompt, + status: validStatus(row.status), + phase: row.phase, + stateJson: row.state_json, + createdAt: row.created_at, + updatedAt: row.updated_at, + }; + } +} + +export function generatePipelineId(): string { + return `pipe_${Date.now()}_${randomBytes(4).toString('hex')}`; +} + +export const pipelineStore = new PipelineStore(config.db.pipelineDbPath); diff --git a/src/pipeline/types.ts b/src/pipeline/types.ts index dadc4d7a..d1500786 100644 --- a/src/pipeline/types.ts +++ b/src/pipeline/types.ts @@ -12,6 +12,16 @@ export type PipelinePhase = | 'done' | 'failed'; +/** 管道生命周期状态（比 PipelinePhase 更宽泛） */ +export type PipelineStatus = + | 'pending_confirm' + | 'running' + | 'done' + | 'failed' + | 'aborted' + | 'interrupted' + | 'cancelled'; + /** 阶段元信息（用于卡片展示） */ export const PHASE_META: Record = { plan: { label: '方案设计', index: 1 }, diff --git a/src/server.ts b/src/server.ts index 358c7652..ebb1c69b 100644 --- a/src/server.ts +++ b/src/server.ts @@ -84,13 +84,25 @@ function startWebSocketMode(eventDispatcher: lark.EventDispatcher, port: number) logger.error({ err }, 'Failed to connect Feishu WebSocket'); }); - // 仍然启动 Express 用于健康检查 + // 仍然启动 Express 用于健康检查 + 卡片交互回调 + // 注意：飞书卡片交互始终通过 HTTP POST 回调（即使事件使用 WebSocket）， + // 因此需要在两种模式下都注册卡片回调端点 const app = express(); + app.use(express.json()); + app.get('/health', (_req, res) => { res.json({ status: 'ok', mode: 'websocket', timestamp: new Date().toISOString() }); }); + // 飞书卡片交互回调 + const cardHandler = createCardActionHandler(); + app.post( + '/feishu/card', + lark.adaptExpress(cardHandler, { autoChallenge: true }), + ); + app.listen(port, () => { - logger.info({ port, mode: 'websocket' }, 'Health check server started'); + logger.info({ port, mode: 'websocket' }, 'Health check + card action server started'); + logger.info(` Card action URL: http://localhost:${port}/feishu/card`); }); }