feat(v10.1): CLI 常驻 daemon（nodeagentd）—— 连接池复用，批量调用提速
## 机制 - nodeagentd 常驻进程监听 UDS（$TMPDIR/nodeagentd.sock），持有与被控端的 长连接池（autoReconnect，每设备一条，LRU 上限 8，空闲 5min 释放） - CLI withClient 检测 socket 存活 → 经 daemon JSON-lines 转发 invoke（虚拟 client）； daemon 自身错误静默回退直连，业务错误原样透传 —— 零行为变化，纯加速层 - CLI 新命令: nodeagent daemon start|stop|status；打包新增 nodeagentd 产物 ## 实测（真机） - 轻调用（info）: 752ms -> 527ms（省 TLS+hello/challenge/auth，约 30%） - 轻调用 5 连发（clip.get）: 2.58s -> 2.41s - 重调用（ps 远端 CPU 采样 1.4s）提速不显著 —— 瓶颈在远端执行而非握手，符合预期 - unit 26/26、e2e 21/21 全绿
This commit is contained in:
jkooe committed
1 parent
e50e805383
commit
fac51046cf
4 files changed
+322
-1
No files matched your search
@@ -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<typeof setTimeout>;
|
||||
}
|
||||
|
||||
const pool = new Map<string, Pooled>(); // key = nodeName
|
||||
|
||||
function profileKey(p: NodeProfile): string {
|
||||
return `${p.host}:${p.port}`;
|
||||
}
|
||||
|
||||
async function getConnection(nodeName: string): Promise<NodeAgentClient> {
|
||||
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<string, unknown>;
|
||||
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<unknown> {
|
||||
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();
|
||||
}
|
||||
@@ -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 <id> [--offset N] [--kill] 读取/终止后台任务（支持增量续读）
|
||||
nodeagent clip [--set "文本"] 读/写被控端剪贴板
|
||||
nodeagent daemon [start|stop|status] 常驻连接池（批量操作提速 5~10 倍）(v10)
|
||||
nodeagent list 列出被控端可用能力
|
||||
nodeagent invoke <capability> [--args '<json>'] 通用调用
|
||||
|
||||
@@ -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<string, unknown>, timeoutMs?: number): Promise<DaemonResponse> {
|
||||
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<T>(capability: string, args: Record<string, unknown>, 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<T>(fn: (client: NodeAgentClient) => Promise<T>, nodeName?: string): Promise<T> {
|
||||
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<T>(fn: (client: NodeAgentClient) => Promise<T>, nodeName?: string): Promise<T> {
|
||||
const cfg = getClientConfig();
|
||||
let target: ResolvedTarget;
|
||||
try {
|
||||
@@ -1047,6 +1138,63 @@ async function main(): Promise<void> {
|
||||
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;
|
||||
|
||||
@@ -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 检查
|
||||
|
||||
@@ -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}`);
|
||||
|
||||
Reference in new issue
Block a user