From 0a47c5a92da5385786c334083a0cd1377d04fcd5 Mon Sep 17 00:00:00 2001 From: jkooe <62494645+jkooe@users.noreply.github.com> Date: Tue, 29 Sep 2026 14:19:09 +0800 Subject: [PATCH] =?UTF-8?q?feat(E1+E2):=20=E4=BA=8B=E4=BB=B6=E8=AE=A2?= =?UTF-8?q?=E9=98=85=E4=B8=8E=E7=9B=91=E6=8E=A7=20=E2=80=94=E2=80=94=20age?= =?UTF-8?q?nt=20=E4=B8=BB=E5=8A=A8=E6=8E=A8=E9=80=81=EF=BC=88=E8=83=BD?= =?UTF-8?q?=E5=8A=9B=2031=20->=2035=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 协议 - 新增 Methods.Event = 'event'：协议中唯一的**无 id 通知**（JSON-RPC notification）， 被控端 → 控制端主动推送，控制端不回复 - 4 项能力：event.watch / event.unwatch / event.list / event.poll - 能力处理器签名扩展为 (args, ctx?)：ctx 携带 owner（连接标识）与 emit（推送出口） ## 被控端（apps/agent/src/events.ts） 三类 watcher： - file —— fs.watch（支持 recursive + glob 过滤），报 created_or_removed / modified - process —— 轮询 tasklist(win)/ps(posix) 做差分，报 process_started / process_exited - net —— 轮询 netstat 提取 LISTEN 端口做差分，报 port_opened / port_closed 两条出口并存： 1) **推送**：通过 event 通知实时发给创建订阅的连接 2) **环形缓冲 + 游标**：500 条上限、seq 游标增量拉取（适配无法收推送的 MCP） 其它：首次采样只建基线（避免开订阅就刷出几百条）；连接断开自动清理该连接的订阅（防泄漏） ## 控制端 - NodeAgentClient 新增 onEvent 回调；handleMessage 把无 id 的 event 通知分发过去 - CLI: nodeagent events [--kind file|process|net] [--path/--pattern/--recursive/--seconds] 无 --kind 则列出当前订阅 - MCP: na_event_watch / na_event_poll / na_event_list / na_event_unwatch ## 真机验证（三类全部实测） - 文件：创建/修改/删除 t1.txt -> 3 个事件实时推送（14:16:33/34/35） - 进程：启动并关闭 Notepad.exe -> process_started / process_exited - 端口：开-关 8899 -> port_opened / port_closed（基线 45 个监听端口） 全部**零轮询**到达控制端（由被控端主动推）。 单测 61 -> 63（补 event.watch 枚举/边界与 event.poll 游标校验）、e2e 21/21 全绿 --- apps/agent/src/capabilities/index.ts | 19 +- apps/agent/src/events.ts | 274 ++++++++++++++++++++++++++ apps/agent/src/server.ts | 18 +- apps/cli/src/index.ts | 111 +++++++++++ apps/mcp/src/index.ts | 52 +++++ packages/client/src/client.ts | 18 +- packages/protocol/src/capabilities.ts | 120 +++++++++++ packages/protocol/src/methods.ts | 5 + tests/unit/capabilities.test.mjs | 23 ++- 9 files changed, 635 insertions(+), 5 deletions(-) create mode 100644 apps/agent/src/events.ts diff --git a/apps/agent/src/capabilities/index.ts b/apps/agent/src/capabilities/index.ts index b3e2a4d..c33edbf 100644 --- a/apps/agent/src/capabilities/index.ts +++ b/apps/agent/src/capabilities/index.ts @@ -7,10 +7,21 @@ import { windowList, windowFocus, screenFind } from './window.js'; import { screenRecord } from './record.js'; import { taskList, taskGet, taskKill } from './task.js'; import { clipGet, clipSet } from './clipboard.js'; +import { eventWatch, eventUnwatch, eventList, eventPoll } from '../events.js'; import { mouseMove, mouseClick, mouseScroll, mouseDrag, keyType, keyPress, setInputPolicy } from './input.js'; import { fsList, fsStat, fsRead, fsWrite, setFsRoots } from './fs.js'; -export type CapabilityHandler = (args: Record) => Promise; +export interface HandlerContext { + /** 连接标识（事件订阅归属 / 断开清理） */ + owner: string; + /** 向该连接推送事件 */ + emit: (evt: unknown) => void; +} + +export type CapabilityHandler = ( + args: Record, + ctx?: HandlerContext, +) => Promise; /** * 装配能力注册表。 @@ -60,5 +71,11 @@ export function createCapabilityRegistry(cfg: AgentConfig): Record + eventWatch(args, ctx ?? { owner: 'unknown', emit: () => undefined }), + [CapabilityNames.EventUnwatch]: eventUnwatch, + [CapabilityNames.EventList]: eventList, + [CapabilityNames.EventPoll]: eventPoll, }; } diff --git a/apps/agent/src/events.ts b/apps/agent/src/events.ts new file mode 100644 index 0000000..b65e94b --- /dev/null +++ b/apps/agent/src/events.ts @@ -0,0 +1,274 @@ +import { watch, type FSWatcher } from 'node:fs'; +import { exec } from 'node:child_process'; +import { CapabilityError, ErrorCodes } from '@nodeagent/protocol'; + +type Args = Record; + +/** + * 被控端事件订阅（v12 / E1+E2）。 + * + * 三类 watcher： + * file —— fs.watch 文件变动（增/删/改），支持递归 + * process —— 轮询进程列表做差分，报「启动/退出」 + * net —— 轮询监听端口做差分，报「端口开/关」 + * + * 两条出口（同时提供）： + * 1) **推送**：通过 event 通知发给创建订阅的那条连接（低延迟、无需轮询） + * 2) **环形缓冲 + 游标**：供无法接收推送的调用方（MCP）拉取 + */ +export interface WatchEvent { + seq: number; + watch_id: string; + kind: 'file' | 'process' | 'net'; + ts: number; + action: string; + target: string; + detail?: string; +} + +interface WatchRecord { + id: string; + kind: 'file' | 'process' | 'net'; + owner: string; // 连接标识，断开时清理 + description: string; + createdAt: number; + count: number; + /** 推送出口（由 server 注入） */ + emit: (evt: WatchEvent) => void; + dispose: () => void; +} + +const watches = new Map(); +const buffer: WatchEvent[] = []; +const MAX_BUFFER = 500; +let seq = 0; +let dropped = 0; + +function nextId(): string { + return `w_${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 6)}`; +} + +function push(rec: WatchRecord, action: string, target: string, detail?: string): void { + seq += 1; + const evt: WatchEvent = { + seq, + watch_id: rec.id, + kind: rec.kind, + ts: Date.now(), + action, + target, + ...(detail ? { detail } : {}), + }; + rec.count += 1; + buffer.push(evt); + if (buffer.length > MAX_BUFFER) { + const overflow = buffer.length - MAX_BUFFER; + buffer.splice(0, overflow); + dropped += overflow; + } + try { + rec.emit(evt); + } catch { + /* 推送失败不影响缓冲 */ + } +} + +/** 简易 glob（仅支持 * 与 ?），用于文件名/进程名过滤。 */ +function globMatch(pattern: string, value: string): boolean { + const re = new RegExp( + `^${pattern + .replace(/[.+^${}()|[\]\\]/g, '\\$&') + .replace(/\*/g, '.*') + .replace(/\?/g, '.')}$`, + 'i', + ); + return re.test(value); +} + +// ---------------- 各类型 watcher ---------------- + +function startFileWatch( + rec: WatchRecord, + path: string, + recursive: boolean, + pattern?: string, +): () => void { + let watcher: FSWatcher; + try { + watcher = watch(path, { recursive, persistent: true }); + } catch (err) { + throw new CapabilityError(ErrorCodes.EXECUTION_FAILED, `无法监控路径: ${path}`, { + detail: String(err), + }); + } + watcher.on('change', (eventType, filename) => { + const name = filename ? String(filename) : ''; + if (pattern && name && !globMatch(pattern, name)) return; + push(rec, eventType === 'rename' ? 'created_or_removed' : 'modified', name || path); + }); + watcher.on('error', (err) => { + push(rec, 'watch_error', path, String(err)); + }); + return () => watcher.close(); +} + +function startProcessWatch(rec: WatchRecord, intervalMs: number, pattern?: string): () => void { + let previous = new Set(); + let first = true; + const tick = (): void => { + // Windows: tasklist / POSIX: ps（避免依赖额外包） + const cmd = + process.platform === 'win32' + ? 'tasklist /fo csv /nh' + : 'ps -eo comm= 2>/dev/null || ps -axo comm='; + exec(cmd, { windowsHide: true, maxBuffer: 4 * 1024 * 1024 }, (err, stdout) => { + if (err) return; + const names = new Set(); + for (const line of stdout.split('\n')) { + const raw = process.platform === 'win32' ? (line.split(',')[0] ?? '').replace(/"/g, '') : line.trim(); + const name = raw.trim(); + if (!name || name.toLowerCase() === 'image name') continue; + if (pattern && !globMatch(pattern, name)) continue; + names.add(name); + } + if (first) { + previous = names; + first = false; + push(rec, 'baseline', `${names.size} 个进程`, '首次采样，仅建立基线不报差异'); + return; + } + for (const n of names) if (!previous.has(n)) push(rec, 'process_started', n); + for (const n of previous) if (!names.has(n)) push(rec, 'process_exited', n); + previous = names; + }); + }; + tick(); + const timer = setInterval(tick, intervalMs); + timer.unref?.(); + return () => clearInterval(timer); +} + +function startNetWatch(rec: WatchRecord, intervalMs: number): () => void { + let previous = new Set(); + let first = true; + const cmd = + process.platform === 'win32' + ? 'netstat -an' + : 'netstat -an 2>/dev/null || ss -ltn'; + const tick = (): void => { + exec(cmd, { windowsHide: true, maxBuffer: 4 * 1024 * 1024 }, (err, stdout) => { + if (err) return; + const listening = new Set(); + for (const line of stdout.split('\n')) { + const m = /[:.](\d{2,5})\s+(LISTEN|LISTENING|ESTABLISHED)?/i.exec(line); + const isListen = /LISTEN/i.test(line); + if (m && isListen) listening.add(m[1]!); + } + if (first) { + previous = listening; + first = false; + push(rec, 'baseline', `${listening.size} 个监听端口`, '首次采样，仅建立基线'); + return; + } + for (const p of listening) if (!previous.has(p)) push(rec, 'port_opened', `:${p}`); + for (const p of previous) if (!listening.has(p)) push(rec, 'port_closed', `:${p}`); + previous = listening; + }); + }; + tick(); + const timer = setInterval(tick, intervalMs); + timer.unref?.(); + return () => clearInterval(timer); +} + +// ---------------- 对外能力 ---------------- + +export interface WatchContext { + /** 连接标识（用于断开时清理） */ + owner: string; + /** 该连接的推送出口 */ + emit: (evt: WatchEvent) => void; +} + +export async function eventWatch(args: Args, ctx: WatchContext): Promise { + const kind = args['kind'] as 'file' | 'process' | 'net'; + if (!['file', 'process', 'net'].includes(kind)) { + throw new CapabilityError(ErrorCodes.PARAM_INVALID, `不支持的 kind: ${String(kind)}`); + } + const intervalMs = Math.max(1000, Math.min(60_000, (args['interval_ms'] as number | undefined) ?? 5000)); + const pattern = args['pattern'] as string | undefined; + const id = nextId(); + const rec: WatchRecord = { + id, + kind, + owner: ctx.owner, + description: '', + createdAt: Date.now(), + count: 0, + emit: ctx.emit, + dispose: () => undefined, + }; + + if (kind === 'file') { + const path = args['path'] as string | undefined; + if (!path) throw new CapabilityError(ErrorCodes.PARAM_INVALID, 'file 类型需要 path'); + const recursive = args['recursive'] === true; + rec.dispose = startFileWatch(rec, path, recursive, pattern); + rec.description = `文件变动 ${path}${recursive ? '（递归）' : ''}${pattern ? ` pattern=${pattern}` : ''}`; + } else if (kind === 'process') { + rec.dispose = startProcessWatch(rec, intervalMs, pattern); + rec.description = `进程启停${pattern ? ` pattern=${pattern}` : ''} 每 ${intervalMs}ms`; + } else { + rec.dispose = startNetWatch(rec, intervalMs); + rec.description = `监听端口开闭 每 ${intervalMs}ms`; + } + + watches.set(id, rec); + return { watch_id: id, kind, description: rec.description }; +} + +export async function eventUnwatch(args: Args): Promise { + const id = args['watch_id'] as string; + const rec = watches.get(id); + if (!rec) return { removed: false }; + rec.dispose(); + watches.delete(id); + return { removed: true }; +} + +export async function eventList(_args: Args): Promise { + return { + watches: [...watches.values()].map((w) => ({ + watch_id: w.id, + kind: w.kind, + description: w.description, + events: w.count, + created_at: w.createdAt, + })), + buffered: buffer.length, + }; +} + +export async function eventPoll(args: Args): Promise { + const limit = Math.max(1, Math.min(500, (args['limit'] as number | undefined) ?? 50)); + const since = (args['since'] as number | undefined) ?? 0; + const watchId = args['watch_id'] as string | undefined; + let list = buffer.filter((e) => e.seq > since); + if (watchId) list = list.filter((e) => e.watch_id === watchId); + const sliced = list.slice(0, limit); + const nextCursor = sliced.length > 0 ? sliced[sliced.length - 1]!.seq : since; + return { events: sliced, next_cursor: nextCursor, dropped }; +} + +/** 连接断开时清理该连接创建的所有订阅（避免 watcher 泄漏）。 */ +export function disposeWatchesByOwner(owner: string): number { + let n = 0; + for (const [id, rec] of watches) { + if (rec.owner === owner) { + rec.dispose(); + watches.delete(id); + n += 1; + } + } + return n; +} diff --git a/apps/agent/src/server.ts b/apps/agent/src/server.ts index e557ff4..1e6987a 100644 --- a/apps/agent/src/server.ts +++ b/apps/agent/src/server.ts @@ -28,6 +28,7 @@ import { } from '@nodeagent/protocol'; import { createCapabilityRegistry } from './capabilities/index.js'; import { audit, describeArgs } from './audit.js'; +import { disposeWatchesByOwner } from './events.js'; import type { AgentConfig } from './config.js'; import type { TlsMaterial } from './certs.js'; @@ -42,6 +43,8 @@ interface ConnState { authorized: string[] | null; /** 来源地址（审计用） */ remote: string; + /** v12：连接标识，用于清理该连接创建的事件订阅 */ + connId: string; } export interface AgentServer { @@ -291,7 +294,11 @@ export function createAgentCore(cfg: AgentConfig): AgentCore { const started = Date.now(); const argsInfo = describeArgs(finalArgs); try { - const data = await handler(finalArgs); + // v12：把连接上下文交给能力处理器（事件订阅需要「谁订阅的」与「往哪推」） + const data = await handler(finalArgs, { + owner: state.connId, + emit: (evt: unknown) => send(ws, { jsonrpc: '2.0', method: 'event', params: evt }), + }); const duration = Date.now() - started; log('info', `invoke ${name} → ok (${duration}ms)`); audit({ @@ -373,6 +380,7 @@ export function createAgentCore(cfg: AgentConfig): AgentCore { function attachConnection(ws: WebSocket, remote: string): void { log('info', `新连接: ${remote}`); + const connId = `c_${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 6)}`; states.set(ws, { authenticated: false, nonce: null, @@ -380,11 +388,17 @@ export function createAgentCore(cfg: AgentConfig): AgentCore { clientId: null, authorized: null, remote, + connId, }); ws.on('message', (raw: RawData) => { void handleMessage(ws, raw.toString()); }); - ws.on('close', () => log('info', `连接关闭: ${remote}`)); + ws.on('close', () => { + log('info', `连接关闭: ${remote}`); + // v12：连接断开即释放其事件订阅，避免 watcher 泄漏 + const n = disposeWatchesByOwner(connId); + if (n > 0) log('debug', `已清理 ${n} 个事件订阅（连接 ${connId}）`); + }); ws.on('error', (err: Error) => log('warn', `连接错误: ${err.message}`)); } diff --git a/apps/cli/src/index.ts b/apps/cli/src/index.ts index e060532..899eb52 100644 --- a/apps/cli/src/index.ts +++ b/apps/cli/src/index.ts @@ -66,6 +66,8 @@ const HELP = `nodeagent —— 跨机 AI 接管框架（控制端 CLI） nodeagent task [--offset N] [--kill] 读取/终止后台任务（支持增量续读） nodeagent clip [--set "文本"] [--out <路径>] [--image-file <路径>] 读/写剪贴板文本或图片（--out 保存图片）(v11) + nodeagent events [--kind file|process|net] [--path <路径>] [--pattern G] [--seconds 15] + 订阅事件并实时接收推送（无 --kind 则列出订阅）(v12) nodeagent record [--duration 5000] [--fps 2] [--scale 0.5] [--region x,y,w,h] 录屏为帧序列（有 ffmpeg 则封装 mp4）(v11) nodeagent deploy [--path <远端路径>] @@ -146,6 +148,10 @@ interface Options { nodes?: string; /** v11 部署目标路径 */ path?: string; + /** v12 事件订阅 */ + kind?: string; + seconds?: string; + intervalMs?: string; } function getClientConfig(): ClientConfig { @@ -770,6 +776,102 @@ async function cmdDeploy(localFile: string, opts: Options): Promise { }); } +/** + * v12 / E1+E2：订阅被控端事件并实时接收推送。 + * 与普通命令不同，它需要**保持连接**一段时间（长连接 + 推送），因此不走 withClient 的「用完即关」。 + */ +async function cmdEvents(opts: Options): Promise { + const cfg = getClientConfig(); + let target: ResolvedTarget; + try { + target = resolveTarget(cfg, currentNodeOverride); + } catch (err) { + fail(err instanceof Error ? err.message : String(err)); + } + + // 无 --kind 时列出当前订阅 + if (!opts.kind) { + await withClient((c) => + callAndPrint(c, CapabilityNames.EventList, {}, opts.json, (data) => { + const d = data as { + watches: Array<{ watch_id: string; kind: string; description: string; events: number }>; + buffered: number; + }; + if (d.watches.length === 0) { + console.log(`当前没有事件订阅（缓冲 ${d.buffered} 条）`); + console.log('用法: nodeagent events --kind file --path <路径> [--seconds 20]'); + return; + } + console.log(`${'订阅ID'.padEnd(24)} ${'类型'.padEnd(8)} 事件数 说明`); + for (const w of d.watches) { + console.log(`${w.watch_id.padEnd(24)} ${w.kind.padEnd(8)} ${String(w.events).padStart(6)} ${w.description}`); + } + }), + ); + return; + } + + const seconds = Math.max(1, Math.min(600, Number(opts.seconds ?? 15))); + const { profile, clientId } = target; + const keys = profile.auth_mode === 'ed25519' ? loadKeys() : null; + const received: Array> = []; + + const client = new NodeAgentClient({ + url: toWsUrl(profile), + key: profile.key ?? '', + clientId, + insecure: profile.insecure, + authMode: profile.auth_mode, + privateKey: keys?.privateKey, + hub: profile.hub ? { token: profile.hub.token, nodeId: profile.hub.node_id } : undefined, + onEvent: (evt) => { + received.push(evt); + const ts = new Date(Number(evt['ts'] ?? Date.now())).toLocaleTimeString('zh-CN'); + const line = `${ts} ${String(evt['action'] ?? '').padEnd(20)} ${String(evt['target'] ?? '')}${evt['detail'] ? ` (${evt['detail']})` : ''}`; + if (opts.json) console.log(JSON.stringify(evt)); + else console.log(` ${line}`); + }, + }); + + try { + await client.connect(); + const args: Record = { kind: opts.kind }; + if (opts.path) args['path'] = opts.path; + if (opts.pattern) args['pattern'] = opts.pattern; + if (opts.recursive) args['recursive'] = true; + if (opts.intervalMs) args['interval_ms'] = Number(opts.intervalMs); + + const r = await client.invoke<{ watch_id: string; description: string }>( + CapabilityNames.EventWatch, + args, + ); + if (r.status === 'failed') fail(`${r.error?.name}: ${r.error?.message}`); + const watchId = (r.data as { watch_id: string }).watch_id; + if (!opts.json) { + console.log(`已订阅：${(r.data as { description: string }).description}`); + console.log(`订阅 ID: ${watchId} 监听 ${seconds}s（Ctrl+C 可提前结束）\n`); + } + + await new Promise((resolve) => { + const t = setTimeout(resolve, seconds * 1000); + process.once('SIGINT', () => { + clearTimeout(t); + resolve(); + }); + }); + + await client.invoke(CapabilityNames.EventUnwatch, { watch_id: watchId }); + if (!opts.json) { + console.log(`\n共收到 ${received.length} 个事件（订阅已取消）`); + } + } catch (err) { + if (err instanceof ClientError) fail(`${err.name}: ${err.message}`); + throw err; + } finally { + client.close(); + } +} + async function cmdClip(opts: Options): Promise { // 图片上传：从本地文件读入（避免超长命令行参数） if (opts.imageFile) { @@ -1273,6 +1375,9 @@ function parseOptions(rest: string[]): { opts: Options; positionals: string[] } fps: { type: 'string' }, nodes: { type: 'string' }, path: { type: 'string' }, + kind: { type: 'string' }, + seconds: { type: 'string' }, + 'interval-ms': { type: 'string' }, }, allowPositionals: true, strict: false, @@ -1309,6 +1414,9 @@ function parseOptions(rest: string[]): { opts: Options; positionals: string[] } fps: values['fps'] as string | undefined, nodes: values['nodes'] as string | undefined, path: values['path'] as string | undefined, + kind: values['kind'] as string | undefined, + seconds: values['seconds'] as string | undefined, + intervalMs: values['interval-ms'] as string | undefined, name: values['name'] as string | undefined, note: values['note'] as string | undefined, recursive: Boolean(values['recursive']), @@ -1447,6 +1555,9 @@ async function main(): Promise { case 'record': await cmdRecord(opts); return; + case 'events': + await cmdEvents(opts); + return; case 'fanout': { const cap = positionals[0]; if (!cap) fail('用法: nodeagent fanout <能力名> [--nodes a,b] [--args JSON]'); diff --git a/apps/mcp/src/index.ts b/apps/mcp/src/index.ts index 2f32ef2..e39d99b 100644 --- a/apps/mcp/src/index.ts +++ b/apps/mcp/src/index.ts @@ -370,6 +370,54 @@ const TOOLS = [ additionalProperties: false, }, }, + { + name: 'na_event_watch', + description: + '订阅被控端事件：file=文件变动、process=进程启停、net=监听端口开闭。' + + 'MCP 无法接收推送，事件会进入服务端缓冲，用 na_event_poll 拉取。', + inputSchema: { + type: 'object', + properties: { + kind: { type: 'string', enum: ['file', 'process', 'net'] }, + path: { type: 'string', description: 'file 类型：监控的目录/文件路径' }, + pattern: { type: 'string', description: '可选：文件名或进程名 glob（如 *.log、chrome*）' }, + recursive: { type: 'boolean', description: 'file 类型：是否递归子目录' }, + interval_ms: { type: 'integer', description: 'process/net 采样间隔（1000~60000，默认 5000）' }, + }, + required: ['kind'], + additionalProperties: false, + }, + }, + { + name: 'na_event_poll', + description: + '拉取已缓冲的被控端事件（增量：传上次返回的 next_cursor）。' + + '典型用法：先 na_event_watch 订阅，稍后 na_event_poll 查看期间发生了什么。', + inputSchema: { + type: 'object', + properties: { + limit: { type: 'integer', description: '最多返回条数（默认 50）' }, + since: { type: 'integer', description: '游标：上次返回的 next_cursor' }, + watch_id: { type: 'string', description: '可选：只看某个订阅' }, + }, + additionalProperties: false, + }, + }, + { + name: 'na_event_list', + description: '列出当前生效的事件订阅（含各自已产生的事件数）与缓冲区大小。', + inputSchema: { type: 'object', properties: {}, additionalProperties: false }, + }, + { + name: 'na_event_unwatch', + description: '取消一个事件订阅（长时间不用的订阅应主动取消，避免浪费被控端资源）。', + inputSchema: { + type: 'object', + properties: { watch_id: { type: 'string' } }, + required: ['watch_id'], + additionalProperties: false, + }, + }, { name: 'na_record', description: @@ -456,6 +504,10 @@ const TOOL_TO_CAPABILITY: Record = { na_fs_write: CapabilityNames.FsWrite, na_fs_stat: CapabilityNames.FsStat, na_restart: CapabilityNames.AgentRestart, + na_event_watch: CapabilityNames.EventWatch, + na_event_poll: CapabilityNames.EventPoll, + na_event_list: CapabilityNames.EventList, + na_event_unwatch: CapabilityNames.EventUnwatch, na_window_list: CapabilityNames.WindowList, na_window_focus: CapabilityNames.WindowFocus, na_screen_find: CapabilityNames.ScreenFind, diff --git a/packages/client/src/client.ts b/packages/client/src/client.ts index f211254..8f78f7d 100644 --- a/packages/client/src/client.ts +++ b/packages/client/src/client.ts @@ -56,6 +56,8 @@ export interface ClientOptions { hub?: { token: string; nodeId: string }; /** 日志回调 */ onLog?: (msg: string) => void; + /** v12：接收被控端主动推送的事件（无 id 的 JSON-RPC 通知） */ + onEvent?: (event: Record) => void; } interface Pending { @@ -215,7 +217,21 @@ export class NodeAgentClient { } const id = (msg as { id?: string | null }).id; - if (!id) return; + if (!id) { + // v12：无 id = 通知（事件推送），交给 onEvent；控制端不应回复 + const method = (msg as { method?: string }).method; + if (method === Methods.Event) { + const params = (msg as { params?: Record }).params; + if (params) { + try { + this.opts.onEvent?.(params); + } catch (err) { + this.opts.onLog?.(`onEvent 回调异常: ${String(err)}`); + } + } + } + return; + } const pending = this.pending.get(id); if (!pending) return; diff --git a/packages/protocol/src/capabilities.ts b/packages/protocol/src/capabilities.ts index e38233d..6b4a9d7 100644 --- a/packages/protocol/src/capabilities.ts +++ b/packages/protocol/src/capabilities.ts @@ -40,6 +40,11 @@ export const CapabilityNames = { ClipSet: 'clip.set', // v11 安全加固 AuditVerify: 'system.audit.verify', + // v12 事件订阅与监控 + EventWatch: 'event.watch', + EventUnwatch: 'event.unwatch', + EventList: 'event.list', + EventPoll: 'event.poll', } as const; export type CapabilityName = (typeof CapabilityNames)[keyof typeof CapabilityNames]; @@ -985,6 +990,121 @@ export const CAPABILITY_MANIFEST: CapabilityDescriptor[] = [ }, }, }, + { + name: CapabilityNames.EventWatch, + version: '1.0', + description: + '订阅被控端事件：file=文件变动（增删改）、process=进程启停、net=监听端口开闭。' + + '被控端通过 event 通知主动推送（无需轮询）；事件同时进入环形缓冲，可用 event.poll 拉取。', + risk: 'low', + params_schema: { + type: 'object', + properties: { + kind: { type: 'string', enum: ['file', 'process', 'net'] }, + path: { type: 'string', description: 'file 类型：监控的目录/文件路径' }, + pattern: { type: 'string', description: '可选：文件名 glob（如 *.log）或进程名子串' }, + recursive: { type: 'boolean', default: false, description: 'file 类型：是否递归子目录' }, + interval_ms: { + type: 'integer', + minimum: 1000, + maximum: 60000, + default: 5000, + description: 'process/net 类型的采样间隔', + }, + }, + required: ['kind'], + additionalProperties: false, + }, + returns_schema: { + type: 'object', + properties: { + watch_id: { type: 'string' }, + kind: { type: 'string' }, + description: { type: 'string' }, + }, + }, + }, + { + name: CapabilityNames.EventUnwatch, + version: '1.0', + description: '取消一个事件订阅（连接断开时也会自动清理）。', + risk: 'low', + params_schema: { + type: 'object', + properties: { watch_id: { type: 'string' } }, + required: ['watch_id'], + additionalProperties: false, + }, + returns_schema: { + type: 'object', + properties: { removed: { type: 'boolean' } }, + }, + }, + { + name: CapabilityNames.EventList, + version: '1.0', + description: '列出当前生效的事件订阅及其已产生事件数。', + risk: 'low', + params_schema: { type: 'object', properties: {}, additionalProperties: false }, + returns_schema: { + type: 'object', + properties: { + watches: { + type: 'array', + items: { + type: 'object', + properties: { + watch_id: { type: 'string' }, + kind: { type: 'string' }, + description: { type: 'string' }, + events: { type: 'integer' }, + created_at: { type: 'integer' }, + }, + }, + }, + buffered: { type: 'integer', description: '环形缓冲中的事件总数' }, + }, + }, + }, + { + name: CapabilityNames.EventPoll, + version: '1.0', + description: + '拉取已缓冲的事件（适合无法接收推送的调用方，如 MCP：先 watch 再 poll）。' + + 'since 传入上次返回的 next_cursor 可增量拉取。', + risk: 'low', + params_schema: { + type: 'object', + properties: { + limit: { type: 'integer', minimum: 1, maximum: 500, default: 50 }, + since: { type: 'integer', description: '游标（上次返回的 next_cursor）' }, + watch_id: { type: 'string', description: '可选：只看某订阅' }, + }, + additionalProperties: false, + }, + returns_schema: { + type: 'object', + properties: { + events: { + type: 'array', + items: { + type: 'object', + properties: { + seq: { type: 'integer' }, + watch_id: { type: 'string' }, + kind: { type: 'string' }, + ts: { type: 'integer' }, + action: { type: 'string' }, + target: { type: 'string' }, + detail: { type: 'string' }, + }, + }, + }, + next_cursor: { type: 'integer' }, + dropped: { type: 'integer', description: '因缓冲上限被丢弃的事件数' }, + }, + }, + }, ]; /** 按名称查找能力描述。 */ diff --git a/packages/protocol/src/methods.ts b/packages/protocol/src/methods.ts index 96fbd4f..a1c4f39 100644 --- a/packages/protocol/src/methods.ts +++ b/packages/protocol/src/methods.ts @@ -16,6 +16,11 @@ export const Methods = { Ping: 'ping', /** 双向：心跳响应 */ Pong: 'pong', + /** + * 被控端 → 控制端：主动推送事件（v12 事件订阅）。 + * 这是协议中唯一的**无 id 通知**（JSON-RPC notification）——控制端不应回复。 + */ + Event: 'event', } as const; export type MethodName = (typeof Methods)[keyof typeof Methods]; diff --git a/tests/unit/capabilities.test.mjs b/tests/unit/capabilities.test.mjs index 9c5899e..c165578 100644 --- a/tests/unit/capabilities.test.mjs +++ b/tests/unit/capabilities.test.mjs @@ -47,11 +47,15 @@ test('清单：v7~v11 新能力均已登记', () => { 'system.audit.verify', 'input.mouse.drag', 'screen.record', + 'event.watch', + 'event.unwatch', + 'event.list', + 'event.poll', ]; for (const name of expected) { assert.ok(findCapability(name), `缺少 ${name}`); } - assert.equal(CAPABILITY_MANIFEST.length, 31, '当前应有 31 项能力'); + assert.equal(CAPABILITY_MANIFEST.length, 35, '当前应有 35 项能力'); }); // ---------- 参数校验（新能力） ---------- @@ -130,6 +134,23 @@ test('system.shell.exec：async 与 wait_forever 为合法布尔', () => { assert.ok(check('system.shell.exec', { command: 'x', async: 'yes' }).length > 0, '类型错误应被拒'); }); +test('event.watch：kind 枚举受控，file 需 path，interval 有上下限', () => { + assert.equal(check('event.watch', { kind: 'file', path: 'C:\\tmp' }).length, 0); + assert.equal(check('event.watch', { kind: 'process', pattern: 'chrome*' }).length, 0); + assert.equal(check('event.watch', { kind: 'net', interval_ms: 2000 }).length, 0); + assert.ok(check('event.watch', { kind: 'disk' }).length > 0, '非法 kind 应被拒'); + assert.ok(check('event.watch', { kind: 'net', interval_ms: 100 }).length > 0, 'interval 下限 1s'); + assert.ok(check('event.watch', { kind: 'net', interval_ms: 99999 }).length > 0, 'interval 上限 60s'); +}); + +test('event.poll：游标与条数可省略，且不接受未知参数', () => { + assert.equal(check('event.poll', {}).length, 0); + assert.equal(check('event.poll', { since: 12, limit: 100 }).length, 0); + assert.ok(check('event.poll', { limit: 0 }).length > 0, 'limit 下限 1'); + assert.ok(check('event.poll', { limit: 999 }).length > 0, 'limit 上限 500'); + assert.ok(check('event.poll', { bogus: 1 }).length > 0, '未知参数应被拒'); +}); + // ---------- 默认值 ---------- test('applyDefaults：新能力默认值正确填充', () => {