宜博 / pixelphone

/**
 * Hermes Chat Server
 *
 * Express + SSE server that wraps Claude CLI (via ClaudeCliProtocol from aipcrm)
 * and exposes Agent Phone tools to Hermes.
 *
 * POST /api/hermes/chat   — SSE stream, same event schema as aipcrm master-chat
 * GET  /api/hermes/status — session list + device reachability
 * DEL  /api/hermes/sessions/:id
 */

const express = require('express');
const cors = require('cors');
const path = require('path');
const fs = require('fs');
const { randomUUID } = require('crypto');
const { EventEmitter } = require('events');

const app = express();
app.use(cors());
app.use(express.json({ limit: '4mb' }));
app.use(express.static(path.join(__dirname, '../public')));

// ─── Paths ───────────────────────────────────────────────────────────────────

// Reuse aipcrm's battle-tested ClaudeCliProtocol directly
const AIPCRM = path.resolve('/Users/yibo/Documents/aipcrm');
const { ClaudeCliProtocol } = require(path.join(AIPCRM, 'packages/acp-core/protocol/ClaudeCliProtocol'));

const { createPhoneTools } = require('./tools/phone-tools');

const PORT = process.env.PORT || 3020;
const WORK_DIR = path.resolve(__dirname, '..');

// ─── Tool Registry (self-contained copy) ─────────────────────────────────────

class ToolRegistry {
    constructor() { this.tools = new Map(); }
    register(tool) { this.tools.set(tool.name, tool); }
    registerAll(tools) { tools.forEach(t => this.register(t)); }

    async execute(name, params) {
        const tool = this.tools.get(name);
        if (!tool) return { error: `Unknown tool: ${name}`, status: 'error' };
        try {
            const result = await tool.execute(params);
            return { result, status: 'success' };
        } catch (err) {
            return { error: err.message, status: 'error' };
        }
    }

    generateToolPrompt() {
        const schemas = Array.from(this.tools.values()).map(t => ({
            name: t.name, description: t.description, parameters: t.parameters || {},
        }));
        if (!schemas.length) return '';

        const descs = schemas.map(s => {
            const props = Object.entries(s.parameters.properties || {}).map(([k, v]) => {
                const req = (s.parameters.required || []).includes(k);
                return `    - ${k} (${v.type}${req ? ', 必需' : ', 可选'}): ${v.description || ''}`;
            }).join('\n');
            return `### ${s.name}\n${s.description}\n\n**参数：**\n${props || '    (无参数)'}`;
        }).join('\n\n---\n\n');

        return `## 可用工具\n\n通过以下格式调用工具：\n\n\`\`\`\n[TOOL_CALL: tool_name]\n{\n  "param1": "value1"\n}\n[/TOOL_CALL]\n\`\`\`\n\n${descs}`;
    }

    static parseToolCalls(output) {
        const regex = /\[TOOL_CALL:\s*([^\]]+)\]\s*([\s\S]*?)\[\/TOOL_CALL\]/g;
        const calls = [];
        let m;
        while ((m = regex.exec(output)) !== null) {
            const toolName = m[1].trim();
            let params = {};
            try { params = JSON.parse(m[2].trim()); } catch { params = { task: m[2].trim() }; }
            calls.push({ tool: toolName, params });
        }
        return calls;
    }
}

// ─── Hermes Session ───────────────────────────────────────────────────────────

const SKILL_MD = fs.existsSync(path.join(__dirname, 'skill.md'))
    ? fs.readFileSync(path.join(__dirname, 'skill.md'), 'utf-8') : '';

class HermesSession extends EventEmitter {
    constructor() {
        super();
        this.protocol = null;
        this.toolRegistry = new ToolRegistry();
        this._initialized = false;
        this._busy = false;
        this._queue = [];
        this._messageCount = 0;
    }

    async init() {
        if (this._initialized) return;
        this.toolRegistry.registerAll(createPhoneTools());

        this.protocol = new ClaudeCliProtocol(
            { name: 'hermes' },
            { systemPrompt: SKILL_MD, effort: process.env.CLAUDE_EFFORT || 'medium' }
        );
        await this.protocol.connect(null, WORK_DIR);
        await this.protocol.newSession();

        this._initialized = true;
        console.log('[HermesSession] 初始化完成');
    }

    async handleMessage(message, options = {}) {
        if (this._busy) {
            return new Promise(resolve => this._queue.push({ message, options, resolve }));
        }
        return this._execute(message, options);
    }

    async _execute(message, options = {}) {
        if (!this._initialized) await this.init();
        this._busy = true;

        const { onData, onToolCall, maxToolRounds = 8 } = options;

        try {
            let prompt;
            if (this._messageCount === 0) {
                const toolPrompt = this.toolRegistry.generateToolPrompt();
                prompt = `${toolPrompt}\n\n---\n\n## 用户消息\n\n${message}\n\n---\n\n**行为约束：** 需要调用工具时直接输出 [TOOL_CALL:] 块，不要先解释。`;
            } else {
                prompt = `## 用户后续消息\n\n${message}`;
            }
            this._messageCount++;

            let allOutput = '';
            let toolCallHistory = [];

            for (let round = 0; round < maxToolRounds; round++) {
                // Collect streaming output for this round
                let roundOutput = '';
                const textChunks = [];

                await new Promise((resolve, reject) => {
                    const onUpdate = ({ update }) => {
                        if (update?.sessionUpdate === 'agent_message_chunk' && update.content?.text) {
                            const chunk = update.content.text;
                            textChunks.push(chunk);
                            roundOutput += chunk;
                            if (onData) onData(chunk);
                        }
                    };
                    const onEvent = ({ method, params }) => {
                        if (method === 'session/result') {
                            this.protocol.off('session_update', onUpdate);
                            this.protocol.off('acp_event', onEvent);
                            resolve(params);
                        }
                    };
                    const onErr = (err) => {
                        this.protocol.off('session_update', onUpdate);
                        this.protocol.off('acp_event', onEvent);
                        reject(err);
                    };
                    this.protocol.once('error', onErr);
                    this.protocol.on('session_update', onUpdate);
                    this.protocol.on('acp_event', onEvent);
                    this.protocol.sendPrompt(prompt).catch(onErr);
                });

                allOutput += roundOutput + '\n';
                const toolCalls = ToolRegistry.parseToolCalls(roundOutput);

                if (!toolCalls.length) {
                    return { type: 'reply', output: roundOutput, toolCalls: toolCallHistory, rounds: round + 1 };
                }

                // Execute tools and build result prompt
                const toolResults = [];
                for (const tc of toolCalls) {
                    if (onToolCall) onToolCall(tc);
                    const result = await this.toolRegistry.execute(tc.tool, tc.params);
                    toolResults.push({ ...tc, result });
                    toolCallHistory.push({ ...tc, result });
                }

                const resultBlocks = toolResults.map(tr => {
                    const r = typeof tr.result === 'object' ? JSON.stringify(tr.result, null, 2) : String(tr.result);
                    return `### 工具 ${tr.tool} 结果\n\`\`\`json\n${r}\n\`\`\``;
                }).join('\n\n');

                prompt = `## 工具执行结果\n\n${resultBlocks}\n\n---\n\n请根据结果决定下一步：还有后续步骤则继续调用工具；全部完成则给用户最终回复。`;
            }

            return { type: 'max_rounds', output: allOutput, toolCalls: toolCallHistory, rounds: maxToolRounds };

        } finally {
            this._busy = false;
            this._drainQueue();
        }
    }

    _drainQueue() {
        if (!this._queue.length || this._busy) return;
        const next = this._queue.shift();
        this._execute(next.message, next.options).then(next.resolve).catch(e => {
            next.resolve({ type: 'error', output: e.message, toolCalls: [], rounds: 0 });
        });
    }

    async shutdown() {
        this._queue = [];
        if (this.protocol) this.protocol.disconnect();
    }
}

// ─── Session pool ─────────────────────────────────────────────────────────────

const sessions = new Map();
const SESSION_IDLE_MS = 15 * 60 * 1000;

function getOrCreateSession(sessionId) {
    let entry = sessions.get(sessionId);
    if (entry) {
        entry.lastActive = Date.now();
        clearTimeout(entry.timer);
        entry.timer = null;
        return entry.session;
    }
    const session = new HermesSession();
    entry = { session, lastActive: Date.now(), timer: null };
    sessions.set(sessionId, entry);
    console.log(`[HermesChat] 新建 session: ${sessionId}`);
    return session;
}

function resetTimer(sessionId) {
    const entry = sessions.get(sessionId);
    if (!entry) return;
    clearTimeout(entry.timer);
    entry.timer = setTimeout(async () => {
        console.log(`[HermesChat] Session 超时: ${sessionId}`);
        await entry.session.shutdown().catch(() => {});
        sessions.delete(sessionId);
    }, SESSION_IDLE_MS);
}

// ─── SSE helper ───────────────────────────────────────────────────────────────

function setupSSE(res) {
    res.writeHead(200, {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
        'Connection': 'keep-alive',
    });
    return (event, data) => res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
}

// ─── Routes ───────────────────────────────────────────────────────────────────

app.post('/api/hermes/chat', async (req, res) => {
    const { message, sessionId: reqId } = req.body;
    if (!message) return res.status(400).json({ error: 'message is required' });

    const sessionId = reqId || `hermes-${Date.now()}-${randomUUID().slice(0, 6)}`;
    const sendSSE = setupSSE(res);
    const session = getOrCreateSession(sessionId);

    sendSSE('session', { sessionId });

    try {
        sendSSE('progress', { step: 'initializing', message: '🤖 Hermes 启动中…' });

        const result = await session.handleMessage(message, {
            onData: (chunk) => {
                if (chunk?.trim()) sendSSE('agent_output', { agent: 'hermes', message: chunk, streamType: 'agent_output' });
            },
            onToolCall: (tc) => {
                sendSSE('tool_call', { tool: tc.tool, params: tc.params, step: tc.tool, message: `调用: ${tc.tool}` });
            },
        });

        sendSSE('result', {
            sessionId,
            type: result.type,
            output: result.output,
            toolCalls: (result.toolCalls || []).map(tc => ({ tool: tc.tool, params: tc.params, status: tc.result?.status })),
            rounds: result.rounds,
        });
    } catch (err) {
        console.error('[HermesChat] 错误:', err);
        sendSSE('error', { message: err.message });
    } finally {
        resetTimer(sessionId);
        res.end();
    }
});

app.get('/api/hermes/sessions', (req, res) => {
    const list = [];
    for (const [id, e] of sessions) {
        list.push({ sessionId: id, lastActive: new Date(e.lastActive).toISOString(), age: Date.now() - e.lastActive });
    }
    res.json(list);
});

app.delete('/api/hermes/sessions/:id', async (req, res) => {
    const entry = sessions.get(req.params.id);
    if (!entry) return res.status(404).json({ error: 'Session not found' });
    clearTimeout(entry.timer);
    await entry.session.shutdown().catch(() => {});
    sessions.delete(req.params.id);
    res.json({ success: true });
});

app.get('/api/hermes/status', (req, res) => {
    res.json({
        sessions: sessions.size,
        phoneUrl: process.env.PHONE_MCP_URL || 'http://localhost:8766',
        eventUrl: process.env.PHONE_EVENT_URL || 'http://localhost:8765',
    });
});

// ─── Start ────────────────────────────────────────────────────────────────────

app.listen(PORT, () => {
    console.log(`\n🤖 Hermes Chat 已启动`);
    console.log(`   UI:      http://localhost:${PORT}`);
    console.log(`   API:     http://localhost:${PORT}/api/hermes/chat`);
    console.log(`   Phone:   ${process.env.PHONE_MCP_URL || 'http://localhost:8766 (adb forward)'}`);
    console.log(`   Events:  ${process.env.PHONE_EVENT_URL || 'http://localhost:8765 (adb forward)'}\n`);
});