ClaudeCliProtocol.js 27.2 KB

Edit Raw Blame History

/**
 * ClaudeCliProtocol — Claude Code CLI stream-json 交互协议
 *
 * 使用 `claude --input-format stream-json --output-format stream-json --verbose
 *   --include-partial-messages --dangerously-skip-permissions` 与 Claude CLI 通信。
 *
 * 持久化进程：启动一次，跨多次 sendPrompt 调用复用。
 *
 * 与 AcpProtocol 的区别：
 *   - 不使用 JSON-RPC（Claude CLI 自有协议）
 *   - 输入：{"type":"user","message":{"role":"user","content":[{"type":"text","text":"..."}]}}
 *   - 输出：NDJSON 流
 *     - {type:"stream_event", event:{type:"content_block_delta", delta:{type:"text_delta",...}}}
 *     - {type:"assistant", message:{...}}  (带 --include-partial-messages)
 *     - {type:"result", result:"...", stop_reason:"end_turn"}
 *   - initialize() / newSession() 为 no-op
 *
 * 事件兼容 AcpProtocol 接口：
 *   - emit('session_update', { update: { sessionUpdate, content: { text } } })
 *   - emit('acp_event', { type, method, params, timestamp })
 *   - emit('error', Error)
 */

const { EventEmitter } = require('events');
const { spawn } = require('child_process');
const fs = require('fs');
const path = require('path');

// LEMONBOSS_CLAUDE_MODE=mock 时,跳过真实 claude CLI 子进程,
// 改读 fixtures/devloop-mock/{agentName}.json 模拟流式响应。
// 用于 DevLoop phase0 acceptance (scripts/phase0-test-call.js)。
const MOCK_MODE = process.env.LEMONBOSS_CLAUDE_MODE === 'mock';

class ClaudeCliProtocol extends EventEmitter {
    constructor(agentInfo = {}, options = {}) {
        super();
        this.agentInfo = agentInfo;
        this.options = options;
        this.isConnected = false;
        this.sessionId = null;
        this._child = null;
        this._workingDir = null;
        this._buffer = '';
        this._promptResolve = null;
        this._promptReject = null;
        // 工具执行状态追踪（用于动态调整空闲超时）
        this._inToolExecution = false;
        // mock 模式（构造时锁定，环境变量优先于 options.mock）
        this._mockMode = MOCK_MODE || options.mock === true;
        this._mockFixtureDir = options.mockFixtureDir
            || path.resolve(__dirname, '../../../fixtures/devloop-mock');
        // P3 场景测试: 如 scenario 目录无 {agent}.json,回落到 baseline 目录
        this._mockFallbackDir = options.mockFallbackDir
            || path.resolve(__dirname, '../../../fixtures/devloop-mock');
    }

    /**
     * 连接：启动 Claude CLI 持久进程
     * transport 参数被忽略：本协议自行管理子进程
     */
    async connect(transport, workingDir) {
        this._workingDir = workingDir;

        // Mock 模式：跳过子进程，立即"已连接"
        if (this._mockMode) {
            console.log('[ClaudeCliProtocol] 🪄 mock 模式启用 (LEMONBOSS_CLAUDE_MODE=mock 或 options.mock)');
            this.isConnected = true;
            return;
        }

        const args = [
            '--input-format', 'stream-json',
            '--output-format', 'stream-json',
            '--verbose',
            '--include-partial-messages',
            '--dangerously-skip-permissions',
        ];

        // DevLoop 真模式: 注入 agent 专属 system prompt (来自 system/agents/{agent}/prompt.md)
        // 不用 --bare — 我们仍想要 Read/Write/Edit/Bash 自动可用 + 工具权限提示等正常默认。
        // 用 --append-system-prompt 而不是 --system-prompt: 保留 Claude 默认提示
        // (含工具使用约定),再追加 agent 的 ROLE / SANDBOX / WORKFLOW 段。
        if (this.options.systemPrompt && typeof this.options.systemPrompt === 'string') {
            args.push('--append-system-prompt', this.options.systemPrompt);
        }

        // 单次会话预算上限 (USD); 防 Agent 跑飞烧钱
        if (typeof this.options.maxBudgetUsd === 'number' && this.options.maxBudgetUsd > 0) {
            args.push('--max-budget-usd', String(this.options.maxBudgetUsd));
        }

        // 支持禁用特定原生工具 (用于编排型 Agent，强制纯文本输出)
        if (this.options.disallowedTools && this.options.disallowedTools.length > 0) {
            args.push('--disallowedTools', ...this.options.disallowedTools);
        }
        // 支持仅开放特定工具 (空字符串 = 禁用全部)
        if (this.options.allowedTools !== undefined) {
            if (this.options.allowedTools === '') {
                args.push('--tools', '');
            } else if (Array.isArray(this.options.allowedTools)) {
                args.push('--allowedTools', ...this.options.allowedTools);
            }
        }

        // 支持 effort 级别控制 (low/medium/high/max)
        // low = 禁用 extended thinking，大幅节省 token
        const effort = this.options.effort || process.env.CLAUDE_EFFORT;
        if (effort) {
            args.push('--effort', effort);
            console.log(`[ClaudeCliProtocol] 💡 effort=${effort}`);
        }

        const processEnv = { ...process.env };
        delete processEnv.NODE_OPTIONS;
        delete processEnv.ELECTRON_RUN_AS_NODE;
        delete processEnv.CLAUDECODE;

        console.log(`[ClaudeCliProtocol] 启动: claude ${args.join(' ')}`);

        this._child = spawn('claude', args, {
            stdio: ['pipe', 'pipe', 'pipe'],
            env: processEnv,
            cwd: workingDir,
        });

        // stdout → NDJSON 解析
        this._child.stdout.on('data', (data) => {
            const raw = data.toString();
            // console.log(`[ClaudeCliProtocol] 📥 stdout (${raw.length} bytes):`, raw.substring(0, 200));
            this._buffer += raw;
            const lines = this._buffer.split('\n');
            this._buffer = lines.pop() || '';

            // 有 stdout 活动 → 重置超时计时器
            if (this._resetIdleTimer) this._resetIdleTimer();

            for (const line of lines) {
                if (!line.trim()) continue;
                try {
                    const event = JSON.parse(line);
                    this._handleStreamEvent(event);
                } catch (e) {
                    // 非 JSON 行：拦截纯文本的崩溃日志或 API 拒绝日志
                    const msg = line.trim();
                    if (msg) {
                        console.log(`[ClaudeCliProtocol] 📝 非JSON文本:`, msg.substring(0, 300));
                        // 触发类似 stderr 的检测逻辑
                        if (this._promptReject && /error|rate.?limit|429|401|403|authentication|api.?key|quota|billing|失败/i.test(msg)) {
                            const err = new Error(`API 请求失败 (Raw): ${msg.substring(0, 500)}`);
                            console.error(`[ClaudeCliProtocol] 拦截到底层纯文本错误:`, err.message);
                            if (this._idleTimer) { clearTimeout(this._idleTimer); this._idleTimer = null; }
                            this._promptReject(err);
                            this._promptResolve = null;
                            this._promptReject = null;
                        }
                    }
                }
            }
        });

        // stderr 转发 + 致命错误检测
        this._child.stderr.on('data', (data) => {
            const msg = data.toString().trim();
            if (msg) {
                console.log(`[ClaudeCliProtocol] 🚨 stderr (${msg.length} bytes):`, msg.substring(0, 500));
                this.emit('stderr', msg);

                // 检测致命错误：如果 sendPrompt 正在等待，主动 reject 防止 Promise 永远挂起
                if (this._promptReject && /rate.?limit|usage.?limit|429|401|403|authentication|api.?key|quota|billing/i.test(msg)) {
                    const err = new Error(`Claude CLI 致命错误: ${msg.substring(0, 500)}`);
                    this._promptReject(err);
                    this._promptResolve = null;
                    this._promptReject = null;
                    console.error('[ClaudeCliProtocol] 致命错误已触发 Promise reject');
                }
            }
        });

        // 进程退出
        this._child.on('exit', (code, signal) => {
            console.log(`[ClaudeCliProtocol] 进程退出: code=${code} signal=${signal}`);
            this.isConnected = false;
            this._child = null;

            if (this._promptReject) {
                this._promptReject(new Error(`Claude CLI 进程退出: code=${code} signal=${signal}`));
                this._promptResolve = null;
                this._promptReject = null;
            }

            this.emit('disconnected', { code, signal });
        });

        this._child.on('error', (err) => {
            console.error('[ClaudeCliProtocol] 进程错误:', err.message);
            this.emit('error', err);
        });

        // 等待进程启动
        await new Promise((resolve, reject) => {
            this._child.on('error', reject);
            setTimeout(resolve, 500);
        });

        this.isConnected = true;
        console.log('[ClaudeCliProtocol] 进程已启动');
    }

    /**
     * Claude CLI 不需要 JSON-RPC initialize — no-op
     */
    async initialize() {
        console.log('[ClaudeCliProtocol] skip initialize (Claude CLI 自有协议)');
        return { protocolVersion: 1 };
    }

    /**
     * Claude CLI 不需要 session/new — no-op
     */
    async newSession(options = {}) {
        this._workingDir = options.cwd || this._workingDir || '.';
        this.sessionId = `claude-cli-${Date.now()}`;
        console.log('[ClaudeCliProtocol] newSession (virtual):', this.sessionId);
        return { sessionId: this.sessionId };
    }

    /**
     * 发送 prompt 到 Claude CLI
     *
     * 写入 stream-json 格式消息到 stdin：
     *   {"type":"user","message":{"role":"user","content":[{"type":"text","text":"..."}]}}
     *
     * 等待 type=result 消息后 resolve。
     *
     * @param {string} text - prompt 文本
     * @returns {Promise<void>}
     */
    sendPrompt(text) {
        // Mock 模式：从 fixture 读 scripted 响应
        if (this._mockMode) {
            return this._sendMockPrompt(text);
        }

        // 按照用户需求取消闲置超时限制 (设置为 24 小时)
        const IDLE_TIMEOUT_CHAT_MS = 24 * 60 * 60_000;     // 已取消正常对话无活动上限
        const IDLE_TIMEOUT_TOOL_MS = 24 * 60 * 60_000;     // 已取消工具执行无活动上限

        return new Promise((resolve, reject) => {
            if (!this._child || !this.isConnected) {
                return reject(new Error('[ClaudeCliProtocol] 进程未启动'));
            }

            this._promptResolve = resolve;
            this._promptReject = reject;
            this._inToolExecution = false;

            // 统一的超时触发逻辑
            const fireTimeout = () => {
                if (this._promptReject) {
                    const mode = this._inToolExecution ? 'tool' : 'chat';
                    const ms = this._inToolExecution ? IDLE_TIMEOUT_TOOL_MS : IDLE_TIMEOUT_CHAT_MS;
                    const err = new Error(`Claude CLI 响应超时 (${ms / 1000}s 无活动, mode=${mode})`);
                    console.error('[ClaudeCliProtocol]', err.message);
                    this._promptReject(err);
                    this._promptResolve = null;
                    this._promptReject = null;
                    this._idleTimer = null;
                }
            };

            // 启动初始超时 (对话态)
            const currentTimeoutMs = () => this._inToolExecution ? IDLE_TIMEOUT_TOOL_MS : IDLE_TIMEOUT_CHAT_MS;
            this._idleTimer = setTimeout(fireTimeout, IDLE_TIMEOUT_CHAT_MS);

            // 每次收到 stdout 数据时重置超时（说明 CLI 还活着）
            this._resetIdleTimer = () => {
                if (this._idleTimer) {
                    clearTimeout(this._idleTimer);
                    this._idleTimer = setTimeout(fireTimeout, currentTimeoutMs());
                }
            };

            // 工具开始 → 切到长超时
            this._onToolStart = () => {
                this._inToolExecution = true;
                if (this._idleTimer) {
                    clearTimeout(this._idleTimer);
                    this._idleTimer = setTimeout(fireTimeout, IDLE_TIMEOUT_TOOL_MS);
                }
                console.log(`[ClaudeCliProtocol] 🔧 工具执行中 → 超时切为 ${IDLE_TIMEOUT_TOOL_MS / 1000}s`);
            };

            // 工具结束 → 切回短超时
            this._onToolEnd = () => {
                this._inToolExecution = false;
                if (this._idleTimer) {
                    clearTimeout(this._idleTimer);
                    this._idleTimer = setTimeout(fireTimeout, IDLE_TIMEOUT_CHAT_MS);
                }
            };

            // Claude CLI stream-json 输入格式
            const message = {
                type: 'user',
                message: {
                    role: 'user',
                    content: [{ type: 'text', text }],
                },
            };

            const jsonLine = JSON.stringify(message) + '\n';

            try {
                this._child.stdin.write(jsonLine);
            } catch (e) {
                clearTimeout(this._idleTimer);
                this._promptResolve = null;
                this._promptReject = null;
                this._idleTimer = null;
                reject(new Error(`[ClaudeCliProtocol] stdin 写入失败: ${e.message}`));
            }
        });
    }

    /**
     * 处理 Claude CLI stream-json 输出事件
     *
     * 事件类型：
     *   - {type:"system"} — 系统初始化
     *   - {type:"stream_event", event:{type:"content_block_delta", delta:{...}}} — 流式增量
     *   - {type:"assistant", message:{content:[{type:"text",text:"..."}]}} — 完整/部分消息
     *   - {type:"result", result:"...", stop_reason:"end_turn"} — 最终结果
     *
     * @private
     */
    _handleStreamEvent(event) {
        // console.log(`[ClaudeCliProtocol] 🔄 event.type=${event.type}`);
        switch (event.type) {
            case 'system':
                this.sessionId = event.session_id || this.sessionId;
                this.emit('acp_event', {
                    type: 'notification',
                    method: 'system/init',
                    params: event,
                    timestamp: Date.now(),
                });
                break;

            case 'stream_event':
                this._handleInnerStreamEvent(event.event, event.session_id);
                break;

            case 'assistant':
                // 带 --include-partial-messages 时，每次 API 响应更新都发送完整 assistant 消息
                // 仅提取增量文本（由 stream_event 处理），这里用于工具检测
                if (event.message?.content) {
                    for (const block of event.message.content) {
                        if (block.type === 'text' && block.text) {
                            // assistant 消息的完整文本（包含已流式发送的内容）
                            // 不再 emit session_update 避免重复，stream_event 已处理增量
                        }
                    }
                }
                break;

            case 'result':
                this.sessionId = event.session_id || this.sessionId;

                // 【核心发现】Claude CLI 会在自己内部重试（长达3分钟），最后如果依然失败，
                // 会输出 result 事件，但带有 "is_error": true，并且 result 字段放着错误信息。
                if (event.is_error) {
                    if (this._idleTimer) { clearTimeout(this._idleTimer); this._idleTimer = null; }
                    const err = new Error(`API 请求失败 (内部重试后放弃): ${event.result || 'Unknown HTTP/RateLimit Error'}`);
                    console.error(`[ClaudeCliProtocol] 收到错误 result:`, err.message);
                    if (this._promptReject) {
                        this._promptReject(err);
                        this._promptResolve = null;
                        this._promptReject = null;
                    }
                    // 不要再发送正常 result notification
                    break;
                }

                this.emit('acp_event', {
                    type: 'notification',
                    method: 'session/result',
                    params: {
                        result: event.result,
                        duration_ms: event.duration_ms,
                        cost_usd: event.total_cost_usd,
                        session_id: event.session_id,
                        stop_reason: event.stop_reason,
                        usage: this._lastUsage || { input: 0, output: 0 },
                        model: this._lastModel || 'unknown',
                    },
                    timestamp: Date.now(),
                });

                this._lastUsage = null;
                this._lastModel = null;

                if (this._promptResolve) {
                    if (this._idleTimer) { clearTimeout(this._idleTimer); this._idleTimer = null; }
                    this._promptResolve();
                    this._promptResolve = null;
                    this._promptReject = null;
                }
                break;
            case 'error':
                this.emit('acp_event', {
                    type: 'notification',
                    method: 'claude/error',
                    params: event,
                    timestamp: Date.now(),
                });

                // API 报错（如 MiniMax 429）会通过 stdout 下发 error 对象，此时不会下发 result
                if (this._promptReject) {
                    const errMsg = event.error?.message || event.message || JSON.stringify(event);
                    const err = new Error(`API 请求失败: ${errMsg}`);
                    console.error(`[ClaudeCliProtocol] 拦截到底层 API 错误:`, err.message);

                    if (this._idleTimer) { clearTimeout(this._idleTimer); this._idleTimer = null; }
                    this._promptReject(err);
                    this._promptResolve = null;
                    this._promptReject = null;
                }
                break;

            default:
                this.emit('acp_event', {
                    type: 'notification',
                    method: `claude/${event.type}`,
                    params: event,
                    timestamp: Date.now(),
                });
        }
    }

    /**
     * 处理 stream_event 内层事件（Anthropic Streaming API 格式）
     *
     * event.type:
     *   - message_start — 消息开始
     *   - content_block_start — 内容块开始（text / thinking / tool_use）
     *   - content_block_delta — 增量内容
     *     - delta.type: "text_delta" → emit session_update
     *     - delta.type: "thinking_delta" → emit session_update (thought)
     *   - content_block_stop — 内容块结束
     *   - message_delta — 消息级更新（stop_reason 等）
     *   - message_stop — 消息结束
     *
     * @private
     */
    _handleInnerStreamEvent(innerEvent, sessionId) {
        if (!innerEvent) return;

        if (innerEvent.usage) {
            this._lastUsage = this._lastUsage || { input: 0, output: 0 };
            this._lastUsage.input += (innerEvent.usage.input_tokens || 0);
            this._lastUsage.output += (innerEvent.usage.output_tokens || 0);
        }
        if (innerEvent.message?.usage) {
            this._lastUsage = this._lastUsage || { input: 0, output: 0 };
            this._lastUsage.input += (innerEvent.message.usage.input_tokens || 0);
            this._lastUsage.output += (innerEvent.message.usage.output_tokens || 0);
        }
        if (innerEvent.message?.model) {
            this._lastModel = innerEvent.message.model;
        }

        switch (innerEvent.type) {
            case 'content_block_delta':
                if (innerEvent.delta?.type === 'text_delta' && innerEvent.delta.text) {
                    this.emit('session_update', {
                        update: {
                            sessionUpdate: 'agent_message_chunk',
                            content: { text: innerEvent.delta.text },
                        },
                    });
                } else if (innerEvent.delta?.type === 'thinking_delta' && innerEvent.delta.thinking) {
                    this.emit('session_update', {
                        update: {
                            sessionUpdate: 'agent_thought_chunk',
                            content: { text: innerEvent.delta.thinking },
                        },
                    });
                }
                break;

            case 'content_block_start':
                // 可用于检测 tool_use 块开始
                if (innerEvent.content_block?.type === 'tool_use') {
                    // 通知超时管理器：工具开始执行，切换到长超时
                    if (this._onToolStart) this._onToolStart();
                    this.emit('acp_event', {
                        type: 'notification',
                        method: 'claude/tool_use_start',
                        params: {
                            tool_use_id: innerEvent.content_block.id,
                            name: innerEvent.content_block.name,
                        },
                        timestamp: Date.now(),
                    });
                }
                break;

            case 'message_start':
            case 'content_block_stop':
            case 'message_delta':
            case 'message_stop':
                // 工具执行结束 → 恢复正常超时
                if (innerEvent.type === 'content_block_stop' && this._inToolExecution) {
                    if (this._onToolEnd) this._onToolEnd();
                }
                // 透传为 acp_event
                this.emit('acp_event', {
                    type: 'notification',
                    method: `claude/stream/${innerEvent.type}`,
                    params: innerEvent,
                    timestamp: Date.now(),
                });
                break;
        }
    }

    /**
     * 断开连接 — 终止 Claude CLI 进程
     */
    disconnect() {
        if (this._child) {
            try { this._child.stdin.end(); } catch { }
            try { this._child.kill('SIGTERM'); } catch { }
            this._child = null;
        }
        this.isConnected = false;
        this.sessionId = null;
    }

    /**
     * Mock 模式 sendPrompt 实现：
     *   1. 找 fixture: {mockFixtureDir}/{agentName}.json (若无,用 default.json,再无 → 走 echo fallback)
     *   2. fixture 两种形态:
     *      (a) 单次:   { chunks, result, files }
     *      (b) 多次:   { sequence: [ {chunks,result,files}, ... ] }
     *                每次调用消耗一个 variant;耗尽后 stick on 最后一个。
     *   3. 同步 emit session_update (agent_message_chunk) + mock/files + session/result
     *   4. resolve
     *
     * 不发起任何子进程,不依赖外部网络;用于 phase0/P1/P2 测试。
     */
    async _sendMockPrompt(text) {
        const rawName = this.agentInfo?.name || this.options.agentName || 'default';
        // 图执行侧的 agent 名带公司前缀（graph-engine.resolveAgentName 返回 `{cid}:{agentId}`），
        // 而 fixture 是按 **agent 类型** 组织的、不该按公司复制一份。取冒号后那段来查。
        // DevLoop 侧的名字不含冒号，split 后原样返回，行为不变。
        const agentName = rawName.split(':').pop();
        const fixturePath = path.join(this._mockFixtureDir, `${agentName}.json`);
        const fallbackPath = this._mockFallbackDir !== this._mockFixtureDir
            ? path.join(this._mockFallbackDir, `${agentName}.json`)
            : null;
        const defaultPath = path.join(this._mockFallbackDir, 'default.json');

        let raw;
        const tryPaths = [fixturePath];
        if (fallbackPath) tryPaths.push(fallbackPath);
        tryPaths.push(defaultPath);

        for (const p of tryPaths) {
            if (!fs.existsSync(p)) continue;
            try {
                raw = JSON.parse(fs.readFileSync(p, 'utf8'));
                break;
            } catch (e) {
                console.warn(`[ClaudeCliProtocol mock] fixture parse failed (${p}): ${e.message}`);
            }
        }
        if (!raw) {
            // echo fallback: 把 prompt 头 80 字符回显
            raw = {
                chunks: [`[mock] received prompt for ${agentName}: `, text.slice(0, 80)],
                result: `mock-ok (no fixture for ${agentName})`,
            };
        }

        // 选 variant:有 sequence → 按调用计数取;否则就是 raw 本身
        let fixture = raw;
        if (Array.isArray(raw.sequence) && raw.sequence.length > 0) {
            const key = `${this._mockFixtureDir}:${agentName}`;
            const idx = ClaudeCliProtocol._mockCallCount.get(key) || 0;
            const pick = Math.min(idx, raw.sequence.length - 1);
            fixture = raw.sequence[pick] || raw.sequence[0];
            ClaudeCliProtocol._mockCallCount.set(key, idx + 1);
        }

        this.sessionId = `claude-cli-mock-${Date.now()}`;

        // 发 chunks（同步触发 session_update,模拟 stream_event delta）
        const chunks = Array.isArray(fixture.chunks) ? fixture.chunks : [];
        for (const c of chunks) {
            this.emit('session_update', {
                update: {
                    sessionUpdate: 'agent_message_chunk',
                    content: { text: typeof c === 'string' ? c : JSON.stringify(c) },
                },
            });
        }

        // 发 mock/files (若 fixture 含 files[]):
        // LoopController 听到这事件后 把 files 落盘到项目根。
        // 这是 mock 模式下让 Agent 真正"产出文件"的桥梁。
        if (Array.isArray(fixture.files) && fixture.files.length > 0) {
            this.emit('acp_event', {
                type: 'notification',
                method: 'mock/files',
                params: {
                    agent: agentName,
                    files: fixture.files, // [{ path, content }] (path 相对项目根)
                },
                timestamp: Date.now(),
            });
        }

        // 发 session/result（模拟 type=result）
        const resultText = fixture.result || chunks.join('');
        this.emit('acp_event', {
            type: 'notification',
            method: 'session/result',
            params: {
                result: resultText,
                duration_ms: 1,
                cost_usd: 0,
                session_id: this.sessionId,
                stop_reason: 'end_turn',
                usage: { input: 0, output: 0 },
                model: 'mock',
                mock: true,
                files_produced: Array.isArray(fixture.files) ? fixture.files.length : 0,
            },
            timestamp: Date.now(),
        });
    }
}

// Mock 模式下的"调用次数"全局计数 (跨实例共享,因为同个 fixture 会被多个
// protocol 实例消费,例如 LoopController 每次 callAgent 都 new 一个 protocol)。
// key = `${mockFixtureDir}:${agentName}`,value = 已用过的 variant 数量。
ClaudeCliProtocol._mockCallCount = new Map();

// 测试 / runUnit 跑前可调用清零,确保 sequence 重新从 0 开始
ClaudeCliProtocol.resetMockState = function () {
    ClaudeCliProtocol._mockCallCount.clear();
};

module.exports = { ClaudeCliProtocol };