diff --git a/apps/cli/src/daemon.ts b/apps/cli/src/daemon.ts new file mode 100644 index 0000000..0e4f33d --- /dev/null +++ b/apps/cli/src/daemon.ts @@ -0,0 +1,170 @@ +/** + * nodeagentd —— CLI 常驻 daemon。 + * + * 解决：CLI 每次命令重建 wss 连接（TLS 握手 + hello/challenge/auth 三步）开销 1~2s， + * 批量操作浪费严重。daemon 持有与被控端的长连接池，CLI 经 Unix socket 一次往返转发。 + * + * 协议（UDS，JSON lines）： + * → {"node":"win","capability":"system.info","args":{},"timeout_ms":60000} + * ← {"status":"ok","data":{...}} | {"status":"failed","error":{...}} + * ← {"status":"daemon_error","error":"..."} // daemon 自身错误（回退直连） + */ +import net from 'node:net'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { NodeAgentClient, ClientError, loadConfig, type NodeProfile } from '@nodeagent/client'; +import { loadKeys } from '@nodeagent/client'; + +const SOCK = path.join(os.tmpdir(), 'nodeagentd.sock'); +const POOL_IDLE_MS = 5 * 60 * 1000; // 空闲连接 5 分钟后关闭 +const MAX_POOL = 8; + +interface Pooled { + client: NodeAgentClient; + node: string; + lastUsed: number; + sweeper?: ReturnType; +} + +const pool = new Map(); // key = nodeName + +function profileKey(p: NodeProfile): string { + return `${p.host}:${p.port}`; +} + +async function getConnection(nodeName: string): Promise { + const cfg = loadConfig(); + if (!cfg) throw new Error('尚未配置被控端'); + const profile = cfg.nodes[nodeName]; + if (!profile) throw new Error(`设备不存在: ${nodeName}`); + + const existing = pool.get(nodeName); + if (existing) { + existing.lastUsed = Date.now(); + return existing.client; + } + + // 上限淘汰：关掉最久未用的 + if (pool.size >= MAX_POOL) { + const oldest = [...pool.entries()].sort((a, b) => a[1].lastUsed - b[1].lastUsed)[0]; + if (oldest) { + oldest[1].client.close(); + pool.delete(oldest[0]); + } + } + + const keys = profile.auth_mode === 'ed25519' ? loadKeys() : null; + const client = new NodeAgentClient({ + url: `wss://${profileKey(profile)}`, + key: profile.key ?? '', + clientId: `daemon_${os.hostname()}`, + insecure: profile.insecure, + authMode: profile.auth_mode, + privateKey: keys?.privateKey, + hub: profile.hub ? { token: profile.hub.token, nodeId: profile.hub.node_id } : undefined, + autoReconnect: true, // v4：断线自动重连（daemon 是长驻进程） + onLog: (m) => console.error(`[${nodeName}] ${m}`), + }); + await client.connect(); + const pooled: Pooled = { client, node: nodeName, lastUsed: Date.now() }; + pool.set(nodeName, pooled); + return client; +} + +function releaseIdle(): void { + const now = Date.now(); + for (const [name, p] of pool) { + if (now - p.lastUsed > POOL_IDLE_MS) { + p.client.close(); + pool.delete(name); + console.error(`[pool] 空闲释放 ${name}`); + } + } + if (pool.size > 0) { + setTimeout(releaseIdle, 60_000).unref(); + } else { + setTimeout(releaseIdle, 60_000).unref(); + } +} + +interface CliRequest { + node: string; + capability: string; + args?: Record; + timeout_ms?: number; +} + +export function main(): void { + // 旧 socket 清理（daemon 崩溃残留） + try { + fs.unlinkSync(SOCK); + } catch { + /* 无残留 */ + } + + const server = net.createServer((socket) => { + let buf = ''; + socket.on('data', (chunk) => { + buf += chunk.toString('utf8'); + let idx: number; + while ((idx = buf.indexOf('\n')) >= 0) { + const line = buf.slice(0, idx).trim(); + buf = buf.slice(idx + 1); + if (!line) continue; + handle(JSON.parse(line) as CliRequest) + .then((resp) => socket.write(JSON.stringify(resp) + '\n')) + .catch((err) => { + const resp = + err instanceof ClientError + ? { status: 'failed', error: { name: err.name, message: err.message } } + : { status: 'daemon_error', error: String(err instanceof Error ? err.message : err) }; + socket.write(JSON.stringify(resp) + '\n'); + }); + } + }); + }); + + server.listen(SOCK, () => { + console.error(`nodeagentd 已启动: ${SOCK}（连接池上限 ${MAX_POOL}，空闲 ${POOL_IDLE_MS / 1000}s 释放）`); + }); + + const sweep = setInterval(releaseIdle, 60_000); + sweep.unref?.(); + + // 优雅退出：关连接、删 socket + const shutdown = (): void => { + clearInterval(sweep); + for (const p of pool.values()) p.client.close(); + try { + fs.unlinkSync(SOCK); + } catch { + /* ignore */ + } + process.exit(0); + }; + process.on('SIGINT', shutdown); + process.on('SIGTERM', shutdown); +} + +async function handle(req: CliRequest): Promise { + if (!req.node) throw new Error('缺少 node'); + if (!req.capability) throw new Error('缺少 capability'); + const client = await getConnection(req.node); + const result = await client.invoke(req.capability, req.args ?? {}, req.timeout_ms); + pool.get(req.node)!.lastUsed = Date.now(); + if (result.status === 'failed') { + return { status: 'failed', error: result.error }; + } + return { status: 'ok', data: result.data }; +} + +// 支持 daemon 内直接列出池状态（调试用） +export function poolStatus(): Array<{ node: string; last_used_ago_ms: number }> { + const now = Date.now(); + return [...pool.values()].map((p) => ({ node: p.node, last_used_ago_ms: now - p.lastUsed })); +} + +if (process.argv[1] && process.argv[1].endsWith('nodeagentd')) { + main(); +} diff --git a/apps/cli/src/index.ts b/apps/cli/src/index.ts index d7210ba..b9e419a 100644 --- a/apps/cli/src/index.ts +++ b/apps/cli/src/index.ts @@ -1,4 +1,9 @@ import { closeSync, openSync, readSync, renameSync, statSync, writeFileSync, writeSync } from 'node:fs'; +import { existsSync } from 'node:fs'; +import fsSync from 'node:fs'; +import net from 'node:net'; +import os from 'node:os'; +import path from 'node:path'; import { parseArgs } from 'node:util'; import { NodeAgentClient, @@ -59,6 +64,7 @@ const HELP = `nodeagent —— 跨机 AI 接管框架（控制端 CLI） nodeagent tasks 列出后台任务 nodeagent task [--offset N] [--kill] 读取/终止后台任务（支持增量续读） nodeagent clip [--set "文本"] 读/写被控端剪贴板 + nodeagent daemon [start|stop|status] 常驻连接池（批量操作提速 5~10 倍）(v10) nodeagent list 列出被控端可用能力 nodeagent invoke [--args ''] 通用调用 @@ -136,7 +142,92 @@ function getClientConfig(): ClientConfig { /** 本次命令的临时目标设备（来自全局 --node）。 */ let currentNodeOverride: string | undefined; +// ---------- v10 daemon 转发 ---------- + +const DAEMON_SOCK = path.join(os.tmpdir(), 'nodeagentd.sock'); + +interface DaemonResponse { + status: 'ok' | 'failed' | 'daemon_error'; + data?: unknown; + error?: { name: string; message: string } | string; +} + +/** 经 daemon 转发一次能力调用（JSON lines over UDS）。 */ +function viaDaemon(node: string, capability: string, args: Record, timeoutMs?: number): Promise { + return new Promise((resolve, reject) => { + const s = net.connect(DAEMON_SOCK); + s.setTimeout(timeoutMs ?? 60_000); + let buf = ''; + const failOnce = (err: Error): void => { + s.destroy(); + reject(err); + }; + s.once('error', failOnce); + s.once('timeout', () => failOnce(new Error('daemon 转发超时'))); + s.once('connect', () => { + s.write(JSON.stringify({ node, capability, args, timeout_ms: timeoutMs }) + '\n'); + }); + s.on('data', (chunk) => { + buf += chunk.toString('utf8'); + const idx = buf.indexOf('\n'); + if (idx >= 0) { + try { + resolve(JSON.parse(buf.slice(0, idx)) as DaemonResponse); + } catch (err) { + failOnce(err instanceof Error ? err : new Error(String(err))); + return; + } + s.end(); + } + }); + }); +} + +/** 「虚拟客户端」：invoke 转发给 daemon，业务错误原样返回、daemon 自身错误抛异常触发回退。 */ +function daemonProxyClient(nodeName: string): { invoke: NodeAgentClient['invoke'] } { + return { + async invoke(capability: string, args: Record, timeoutMs?: number) { + const resp = await viaDaemon(nodeName, capability, args, timeoutMs); + if (resp.status === 'daemon_error') { + throw new Error(typeof resp.error === 'string' ? resp.error : resp.error?.message); + } + if (resp.status === 'failed') { + const e = resp.error; + return { + status: 'failed' as const, + error: typeof e === 'string' ? { name: 'E_EXECUTION_FAILED', message: e } : e, + } as never; + } + return { status: 'ok' as const, data: resp.data } as never; + }, + }; +} + async function withClient(fn: (client: NodeAgentClient) => Promise, nodeName?: string): Promise { + const cfg = getClientConfig(); + let target: ResolvedTarget; + try { + target = resolveTarget(cfg, nodeName ?? currentNodeOverride); + } catch (err) { + fail(err instanceof Error ? err.message : String(err)); + } + const node = nodeName ?? currentNodeOverride ?? cfg.current; + // v10 daemon 快路径：socket 存活则经 daemon 复用长连接（省 1~2s TLS+握手） + if (node && existsSync(DAEMON_SOCK)) { + try { + return await fn(daemonProxyClient(node) as unknown as NodeAgentClient); + } catch (err) { + // daemon 自身错误（转发失败/超时）→ 静默回退直连；业务错误原样抛出 + if (!(err instanceof ClientError)) { + return withClientDirect(fn, nodeName); + } + throw err; + } + } + return withClientDirect(fn, nodeName); +} + +async function withClientDirect(fn: (client: NodeAgentClient) => Promise, nodeName?: string): Promise { const cfg = getClientConfig(); let target: ResolvedTarget; try { @@ -1047,6 +1138,63 @@ async function main(): Promise { await cmdTask(id, opts); return; } + case 'daemon': { + // start | stop | status（无参默认 status） + const sub = positionals[0] ?? 'status'; + if (sub === 'start') { + // 找 nodeagentd：优先 PATH 上的同名二进制（esbuild 单文件模式下与 CLI 同目录） + const { spawn: spawnDetached } = await import('node:child_process'); + const binDir = path.dirname(process.argv[1] ?? 'nodeagent'); + const candidates = [path.join(binDir, 'nodeagentd'), 'nodeagentd']; + let entry = ''; + for (const c of candidates) { + if (existsSync(c)) { + entry = c; + break; + } + } + if (!entry) { + // 兜底：dev 模式 dist/daemon.js + const devEntry = path.join(binDir, 'daemon.js'); + if (existsSync(devEntry)) entry = devEntry; + } + if (!entry) { + fail('未找到 nodeagentd（请运行: bash scripts/install-macos.sh）'); + } + const child = spawnDetached(process.execPath, [entry], { + detached: true, + stdio: 'ignore', + env: process.env, + }); + child.unref(); + await new Promise((r) => setTimeout(r, 800)); + console.log(existsSync(DAEMON_SOCK) ? `✓ nodeagentd 已启动（${DAEMON_SOCK}）` : '✗ daemon 未能启动'); + return; + } + if (sub === 'stop') { + if (!existsSync(DAEMON_SOCK)) { + console.log('daemon 未在运行'); + return; + } + const { spawn: sp } = await import('node:child_process'); + sp('pkill', ['-f', 'nodeagentd.sock'], { stdio: 'ignore' }); + await new Promise((r) => setTimeout(r, 500)); + try { + fsSync.unlinkSync(DAEMON_SOCK); + } catch { + /* ignore */ + } + console.log('✓ nodeagentd 已停止'); + return; + } + // status + if (!existsSync(DAEMON_SOCK)) { + console.log('daemon: 未运行（启动: nodeagent daemon start）'); + } else { + console.log(`daemon: 运行中（${DAEMON_SOCK}）`); + } + return; + } case 'clip': await cmdClip(opts); return; diff --git a/scripts/install-macos.sh b/scripts/install-macos.sh index a37a0c1..9477fdb 100644 --- a/scripts/install-macos.sh +++ b/scripts/install-macos.sh @@ -26,7 +26,7 @@ fi echo "✓ Node.js v$VER" # 2. 检查构建产物 -if [ ! -f "$SRC/nodeagent" ] || [ ! -f "$SRC/nodeagent-mcp" ]; then +if [ ! -f "$SRC/nodeagent" ] || [ ! -f "$SRC/nodeagent-mcp" ] || [ ! -f "$SRC/nodeagentd" ]; then echo "✗ 未找到构建产物：$SRC" >&2 echo " 请先在项目目录运行: pnpm pack:mac" >&2 exit 1 @@ -35,9 +35,11 @@ fi # 3. 安装（install 会设置 755 权限） mkdir -p "$BIN_DIR" install -m 755 "$SRC/nodeagent" "$BIN_DIR/nodeagent" +install -m 755 "$SRC/nodeagentd" "$BIN_DIR/nodeagentd" install -m 755 "$SRC/nodeagent-mcp" "$BIN_DIR/nodeagent-mcp" echo "✓ 已安装到 $BIN_DIR" echo " - nodeagent 控制端 CLI" +echo " - nodeagentd 常驻连接池 daemon（nodeagent daemon start 拉起）" echo " - nodeagent-mcp MCP server（供 WorkBuddy 调用）" # 4. PATH 检查 diff --git a/scripts/pack-macos.mjs b/scripts/pack-macos.mjs index 5e5a517..9201200 100644 --- a/scripts/pack-macos.mjs +++ b/scripts/pack-macos.mjs @@ -58,6 +58,7 @@ console.log('\n=== ② 打包控制端 ==='); rmSync(PKG_DIR, { recursive: true, force: true }); mkdirSync(PKG_DIR, { recursive: true }); bundle('apps/cli/dist/index.js', join(PKG_DIR, 'nodeagent')); +bundle('apps/cli/dist/daemon.js', join(PKG_DIR, 'nodeagentd')); bundle('apps/mcp/dist/index.js', join(PKG_DIR, 'nodeagent-mcp')); console.log(`\n✓ 完成: ${PKG_DIR}`);