From c411bbad52f7ed2ff490f258b9629f29390e8f90 Mon Sep 17 00:00:00 2001 From: wing Date: Tue, 10 Feb 2026 11:52:38 +0800 Subject: [PATCH] feat(contracts): add multi-agent communication protocol (agent-message.js) - Define 4 message types: task-request, task-result, status-update, knowledge-share - Validators with { ok, value } | { ok: false, error } pattern - Factory functions for convenient message creation - 41 unit tests + 5 MessageBus integration tests - Update contracts/index.js and core/index.js exports - Add Claude Code Agent Teams comparison analysis to architecture doc (8.5/8.6) via [HAPI](https://hapi.run) --- docs/sdk-architecture-design.md | 104 +++++ js/agents/core/contracts/CLAUDE.md | 23 + js/agents/core/contracts/agent-message.js | 407 ++++++++++++++++ js/agents/core/contracts/index.js | 14 + js/agents/core/index.js | 12 + .../core/message-bus-agent-protocol.test.js | 194 ++++++++ .../core/contracts/agent-message.test.js | 437 ++++++++++++++++++ 7 files changed, 1191 insertions(+) create mode 100644 js/agents/core/contracts/agent-message.js create mode 100644 tests/integration/agents/core/message-bus-agent-protocol.test.js create mode 100644 tests/unit/agents/core/contracts/agent-message.test.js diff --git a/docs/sdk-architecture-design.md b/docs/sdk-architecture-design.md index fe9ee8f9..776f6a5a 100644 --- a/docs/sdk-architecture-design.md +++ b/docs/sdk-architecture-design.md @@ -278,3 +278,107 @@ import { DesignAgentLoop } from "js/agents/stages/design/agent-loop.js"; | 容错 | checkpoint | 无 | 无 | **WAL + DegradationMatrix + CircuitBreaker** | | 运行时替换 | 无 | 无 | 无 | **Plugin 热插拔 + ServiceBus 中间件** | | 浏览器支持 | 无 | 无 | 无 | **完整支持** | + +### 8.5 Claude Code Agent Teams 对比分析 + +> 2026-02-10 分析，基于 Claude Code CLI v2.1.32+ Agent Teams (Research Preview) + +#### 8.5.1 Claude Code Agent Teams 架构摘要 + +Claude Code Agent Teams 采用 **Lead + Teammates 星型拓扑**： + +- **Lead Agent**: 永久角色，负责任务编排和协调，不可转移 +- **Teammates**: 独立 Claude Session，各有自己的 context window，不继承 Lead 对话历史 +- **Shared Task List**: 文件持久化（`~/.claude/tasks/{team-name}/`），DAG 依赖感知 +- **Mailbox**: Lead 可 `message`（点对点）或 `broadcast`（一对多）给 Teammates +- **空闲通知**: Teammate 完成后自动 notify Lead +- **成本模型**: 每个 Teammate 是完整 session，成本线性增长 + +#### 8.5.2 架构维度对比 + +| 维度 | Claude Code Agent Teams | Paper-Burner | +|------|------------------------|-------------| +| **拓扑** | Lead + Teammates（星型，Lead 不可转移） | Orchestrator + AgentLoop（可嵌套，Sequential/Parallel） | +| **通信** | Mailbox（文件系统），message/broadcast | MessageBus（EventBus + RPC），内存级，channel 路由 | +| **任务管理** | SharedTaskList（文件持久化，DAG 依赖） | TaskGraph（内存 DAG）+ TaskTool（三种 ContextMode） | +| **上下文传递** | 每个 Teammate 独立 context window | ContextMode: isolated/shared/handoff + SubagentBudget | +| **安全** | Tool permissions per teammate | InjectionScanner + quarantineOutput + ToolPermissions | +| **状态共享** | 文件系统 + Git worktree | SharedMemory（SAB 零拷贝）+ StateBus 响应式 | +| **数据传输** | 序列化（文件） | SAB 零拷贝 / MessagePort 256KB 分块回退 | +| **回溯** | 无 | DMailTool 软回溯（Steins;Gate D-Mail） | +| **持久化** | 文件系统原生 | Archive + Checkpoint Schema | +| **可观测性** | 基础 CLI 输出 | Telemetry + EventBus hooks | + +#### 8.5.3 我们已有的优势 + +1. **通信层更强** — MessageBus 是内存级 RPC + pub/sub，支持 channel 路由、request/response 语义、幂等去重。Claude 用文件系统 Mailbox，延迟高且无类型安全。`agent-message.js` 协议（4 种消息类型 + 验证器 + 工厂函数）进一步强化了这一点。 + +2. **上下文管理更精细** — `ContextMode` 三级（isolated/shared/handoff）+ `SubagentBudgetManager` 按模式分配 token 比例（15%/25%/35%）。Claude Teammates 只有"独立 context window"一种模式。 + +3. **安全层更深** — `InjectionScanner` + `quarantineOutput` 检疫机制：子代理输出经过 schema 验证 + Prompt Injection 扫描后才能回传父代理。Claude 仅有 tool permissions。 + +4. **零拷贝数据传输** — `SharedMemory` 支持 SharedArrayBuffer 零拷贝，处理大科研数据集时无需序列化。Claude 完全没有等价能力。 + +5. **DMailTool 软回溯** — 标记历史 turns 为 superseded 而不删除，支持 minor/major/critical 严重级别。Claude 没有等价物。 + +6. **CRDT 无冲突合并** — LWW/GCounter/PNCounter/ORSet/Document 全套 CRDT 原语，支持多 Agent 并发写入自动合并。Claude 依赖文件系统锁，建议"assign teammates to separate directories"来避免冲突。 + +7. **Plugin 热插拔** — ServiceBus 中间件 + Plugin 系统支持运行时扩展。Claude Agent Teams 是封闭系统。 + +#### 8.5.4 Claude Code 值得学习的点 + +1. **共享任务列表（Shared Task List）的持久化与跨 Agent 可见性** + - Claude 的 DAG 依赖感知任务列表是文件持久化的，所有 Agent 可以认领（claim）和更新任务状态 + - 我们的 `TaskGraph` 有 DAG 能力，但缺少持久化和跨 Agent 可见性 + - **建议**: 在 StateBus 上构建 `SharedTaskBoard`，支持任务持久化、claim 语义、依赖阻塞 + +2. **Broadcast 语义的显式化** + - Claude 的 broadcast 虽然简单，但"一对多"广播在协作场景中非常实用 + - 我们的 MessageBus 有 `emit`（pub/sub），但缺少显式的 `broadcast` API 和 `agent:*` 通配订阅 + - **建议**: 在 `agent-message.js` 上加一层 `AgentBroadcaster` + +3. **Teammate 自动空闲通知** + - Teammate 完成后自动 notify Lead，Lead 不需要轮询 + - 我们的 Orchestrator 通过回调知道 Stage 结束，但 MessageBus 层没有标准化的生命周期事件 + - **建议**: 用 `createStatusUpdate(agentId, 'stopped')` 作为标准退出信号，Orchestrator 自动订阅 + +4. **成本感知与汇总** + - Claude 明确指出"每个 Teammate 是完整 session，成本线性增长" + - 我们有 `SubagentBudgetManager` 管理 token 预算，但缺少跨 Agent 成本汇总报告 + - **建议**: 添加 `CostAggregator` 插件 + +5. **文件级任务隔离** + - Claude 建议"assign teammates to separate directories"避免文件冲突 + - 这是一个简单但有效的策略，适合无 CRDT 的场景 + - 我们的 CRDT 已经解决了并发冲突，但可以在任务分配时加入目录亲和性提示 + +#### 8.5.5 实施状态更新 + +阶段 1 进展（2026-02-10）： + +| 子任务 | 状态 | 说明 | +|--------|------|------| +| 定义 Agent 通信协议 | ✅ 完成 | `contracts/agent-message.js` — 4 种消息 + 验证器 + 工厂 | +| 单元测试 | ✅ 完成 | 41 个测试全通过 | +| MessageBus 集成测试 | ✅ 完成 | 5 个集成测试（RPC roundtrip, broadcast, channel routing, delegation chain, timeout） | +| contracts/index.js 导出 | ✅ 完成 | 14 个新导出 | +| core/index.js 导出 | ✅ 完成 | 上层可直接 import | +| Stage 使用 MessageBus RPC | ⏳ 待做 | | +| Orchestrator.getStatus() | ⏳ 待做 | | +| SharedTaskBoard | ⏳ 待做 | 受 Claude Agent Teams 启发新增 | +| AgentLifecycleEvents | ⏳ 待做 | 受 Claude Agent Teams 启发新增 | + +### 8.6 阶段 1 修订后的实施计划 + +基于 Claude Code Agent Teams 分析，阶段 1 新增 3 个子任务： + +``` +阶段 1：单进程多 Agent（修订版） + ├── [✅] 定义 Agent 通信协议（contracts/agent-message.js） + ├── [⏳] 让现有 Stage 使用 MessageBus RPC + ├── [⏳] Orchestrator 暴露 getStatus() API + ├── [⏳] SharedTaskBoard — 持久化共享任务列表 + claim 语义（新增，受 Claude 启发） + ├── [⏳] AgentLifecycleEvents — 标准化 idle/busy/stopped 广播（新增，受 Claude 启发） + ├── [⏳] CostAggregator — 跨 Agent token 使用汇总（新增） + └── 验证场景：DeepSearch + CodeSearch 协作研究 +``` diff --git a/js/agents/core/contracts/CLAUDE.md b/js/agents/core/contracts/CLAUDE.md index 623b2b26..23e8c1e4 100644 --- a/js/agents/core/contracts/CLAUDE.md +++ b/js/agents/core/contracts/CLAUDE.md @@ -14,6 +14,7 @@ Agent 边界的轻量级结构验证，防止类型欺骗。 |------|------| | `index.js` | Contracts 聚合导出（推荐入口） | | `rpc-message.js` | 跨 Agent RPC request/response 消息契约 | +| `agent-message.js` | 多 Agent 通信协议（task-request/result, status-update, knowledge-share） | | `llm-response.js` | LLM 响应与 Tool Call 契约 | | `tool-result.js` | 工具执行结果契约（validate + normalize） | | `disposable.js` | 资源生命周期 Disposable 契约与释放辅助 | @@ -45,6 +46,28 @@ if (!res.ok) { console.warn('Invalid RPC response:', res.error); } +// === 多 Agent 通信协议 === +import { + validateAgentMessage, + createTaskRequest, + createTaskResult, + createStatusUpdate, + createKnowledgeShare, +} from 'js/agents/core/contracts'; + +// 创建并验证任务请求 +const taskReq = createTaskRequest('agent-a', 'search:execute', { query: 'hello' }); +const validated = validateAgentMessage(taskReq); + +// 创建任务结果 +const result = createTaskResult('agent-b', taskReq.correlationId, 'completed', { data: {...} }); + +// 广播状态更新 +const status = createStatusUpdate('agent-a', 'busy', { progress: 50 }); + +// 共享知识 +const knowledge = createKnowledgeShare('agent-a', 'findings', { ... }, { contentType: 'json' }); + // 验证 LLM 响应 const llm = validateLlmResponse(response); if (!llm.ok) { diff --git a/js/agents/core/contracts/agent-message.js b/js/agents/core/contracts/agent-message.js new file mode 100644 index 00000000..69637595 --- /dev/null +++ b/js/agents/core/contracts/agent-message.js @@ -0,0 +1,407 @@ +/** + * Agent Message Contracts - 多 Agent 通信协议 + * + * 定义跨 Agent 标准消息格式，用于 MessageBus 通信： + * - task-request: 请求另一个 Agent 执行任务 + * - task-result: 任务执行结果 + * - status-update: Agent 状态广播 + * - knowledge-share: 知识共享（发现的信息传递给其他 Agent） + * + * 设计原则： + * - 与 rpc-message.js 互补（RPC 是底层传输，agent-message 是应用协议） + * - 每个消息携带 agentId 发送方标识 + * - task-request 使用 domain:action 格式的 taskType + * - 所有 validate 返回 { ok, value } | { ok: false, error } + */ + +// ─── 常量 ─────────────────────────────────────────────── + +/** @type {readonly ['task-request', 'task-result', 'status-update', 'knowledge-share']} */ +export const AgentMessageKind = /** @type {const} */ ([ + 'task-request', + 'task-result', + 'status-update', + 'knowledge-share', +]); + +/** @type {readonly ['pending', 'running', 'completed', 'failed', 'cancelled']} */ +export const TaskStatus = /** @type {const} */ ([ + 'pending', + 'running', + 'completed', + 'failed', + 'cancelled', +]); + +/** @type {readonly ['idle', 'busy', 'degraded', 'stopping', 'stopped']} */ +export const AgentRunStatus = /** @type {const} */ ([ + 'idle', + 'busy', + 'degraded', + 'stopping', + 'stopped', +]); + +const TASK_TYPE_PATTERN = /^[a-z][a-zA-Z0-9]*:[a-z][a-zA-Z0-9]*$/; + +// ─── 类型定义 ─────────────────────────────────────────── + +/** + * @typedef {{ ok: true, value: T }} ValidResult + * @template T + */ + +/** + * @typedef {{ ok: false, error: string }} InvalidResult + */ + +/** + * @typedef {ValidResult | InvalidResult} ValidationResult + * @template T + */ + +/** + * @typedef {Object} TaskRequest + * @property {'task-request'} kind + * @property {string} agentId - 发送方 Agent ID + * @property {string} [targetAgentId] - 目标 Agent ID（可选，广播时省略） + * @property {string} taskType - domain:action 格式 + * @property {unknown} payload - 任务参数 + * @property {string} [correlationId] - 关联 ID（用于追踪） + * @property {number} [priority] - 优先级 (0=最高, 默认 5) + * @property {number} [timeoutMs] - 超时毫秒 + * @property {number} ts - 时间戳 + */ + +/** + * @typedef {Object} TaskResult + * @property {'task-result'} kind + * @property {string} agentId - 发送方 Agent ID + * @property {string} correlationId - 对应 TaskRequest 的 correlationId + * @property {typeof TaskStatus[number]} status - 任务状态 + * @property {unknown} [data] - 返回数据 + * @property {string} [error] - 错误信息 + * @property {number} [durationMs] - 执行耗时 + * @property {number} ts - 时间戳 + */ + +/** + * @typedef {Object} StatusUpdate + * @property {'status-update'} kind + * @property {string} agentId - 发送方 Agent ID + * @property {typeof AgentRunStatus[number]} status - Agent 运行状态 + * @property {string} [currentTask] - 当前任务描述 + * @property {number} [progress] - 进度百分比 (0-100) + * @property {Record} [meta] - 附加元数据 + * @property {number} ts - 时间戳 + */ + +/** + * @typedef {Object} KnowledgeShare + * @property {'knowledge-share'} kind + * @property {string} agentId - 发送方 Agent ID + * @property {string} [targetAgentId] - 目标 Agent ID（可选） + * @property {string} topic - 知识主题 + * @property {unknown} content - 知识内容 + * @property {string} [contentType] - 内容类型 (text/json/reference) + * @property {string} [correlationId] - 关联 ID + * @property {number} ts - 时间戳 + */ + +/** + * @typedef {TaskRequest | TaskResult | StatusUpdate | KnowledgeShare} AgentMessage + */ + +// ─── 内部工具 ─────────────────────────────────────────── + +/** + * @param {unknown} v + * @returns {string | null} + */ +function str(v) { + return typeof v === 'string' && v.trim().length > 0 ? v.trim() : null; +} + +/** + * @param {unknown} v + * @returns {Record | null} + */ +function obj(v) { + return v !== null && typeof v === 'object' && !Array.isArray(v) + ? /** @type {Record} */ (v) + : null; +} + +/** + * @param {unknown} v + * @param {number} def + * @returns {number} + */ +function num(v, def) { + return typeof v === 'number' && Number.isFinite(v) ? v : def; +} + +// ─── 验证函数 ─────────────────────────────────────────── + +/** + * 验证 TaskRequest + * @param {unknown} msg + * @returns {ValidationResult} + */ +export function validateTaskRequest(msg) { + const o = obj(msg); + if (!o) return { ok: false, error: 'TaskRequest: expected object' }; + + const agentId = str(o.agentId); + if (!agentId) return { ok: false, error: 'TaskRequest.agentId: required non-empty string' }; + + const taskType = str(o.taskType); + if (!taskType) return { ok: false, error: 'TaskRequest.taskType: required non-empty string' }; + if (!TASK_TYPE_PATTERN.test(taskType)) { + return { ok: false, error: 'TaskRequest.taskType: must be domain:action format (e.g. "search:execute")' }; + } + + const priority = num(o.priority, 5); + if (priority < 0 || priority > 10) { + return { ok: false, error: 'TaskRequest.priority: must be 0-10' }; + } + + const timeoutMs = o.timeoutMs !== undefined ? num(o.timeoutMs, -1) : undefined; + if (timeoutMs !== undefined && timeoutMs <= 0) { + return { ok: false, error: 'TaskRequest.timeoutMs: must be positive' }; + } + + return { + ok: true, + value: { + kind: 'task-request', + agentId, + ...(str(o.targetAgentId) ? { targetAgentId: str(o.targetAgentId) } : {}), + taskType, + payload: o.payload, + ...(str(o.correlationId) ? { correlationId: str(o.correlationId) } : {}), + priority, + ...(timeoutMs !== undefined ? { timeoutMs } : {}), + ts: num(o.ts, Date.now()), + }, + }; +} + +/** + * 验证 TaskResult + * @param {unknown} msg + * @returns {ValidationResult} + */ +export function validateTaskResult(msg) { + const o = obj(msg); + if (!o) return { ok: false, error: 'TaskResult: expected object' }; + + const agentId = str(o.agentId); + if (!agentId) return { ok: false, error: 'TaskResult.agentId: required non-empty string' }; + + const correlationId = str(o.correlationId); + if (!correlationId) return { ok: false, error: 'TaskResult.correlationId: required non-empty string' }; + + const status = str(o.status); + if (!status || !TaskStatus.includes(/** @type {any} */ (status))) { + return { ok: false, error: `TaskResult.status: must be one of ${TaskStatus.join(', ')}` }; + } + + return { + ok: true, + value: { + kind: 'task-result', + agentId, + correlationId, + status: /** @type {typeof TaskStatus[number]} */ (status), + ...(o.data !== undefined ? { data: o.data } : {}), + ...(typeof o.error === 'string' ? { error: o.error } : {}), + ...(typeof o.durationMs === 'number' ? { durationMs: o.durationMs } : {}), + ts: num(o.ts, Date.now()), + }, + }; +} + +/** + * 验证 StatusUpdate + * @param {unknown} msg + * @returns {ValidationResult} + */ +export function validateStatusUpdate(msg) { + const o = obj(msg); + if (!o) return { ok: false, error: 'StatusUpdate: expected object' }; + + const agentId = str(o.agentId); + if (!agentId) return { ok: false, error: 'StatusUpdate.agentId: required non-empty string' }; + + const status = str(o.status); + if (!status || !AgentRunStatus.includes(/** @type {any} */ (status))) { + return { ok: false, error: `StatusUpdate.status: must be one of ${AgentRunStatus.join(', ')}` }; + } + + const progress = o.progress !== undefined ? num(o.progress, -1) : undefined; + if (progress !== undefined && (progress < 0 || progress > 100)) { + return { ok: false, error: 'StatusUpdate.progress: must be 0-100' }; + } + + return { + ok: true, + value: { + kind: 'status-update', + agentId, + status: /** @type {typeof AgentRunStatus[number]} */ (status), + ...(str(o.currentTask) ? { currentTask: str(o.currentTask) } : {}), + ...(progress !== undefined ? { progress } : {}), + ...(obj(o.meta) ? { meta: obj(o.meta) } : {}), + ts: num(o.ts, Date.now()), + }, + }; +} + +/** + * 验证 KnowledgeShare + * @param {unknown} msg + * @returns {ValidationResult} + */ +export function validateKnowledgeShare(msg) { + const o = obj(msg); + if (!o) return { ok: false, error: 'KnowledgeShare: expected object' }; + + const agentId = str(o.agentId); + if (!agentId) return { ok: false, error: 'KnowledgeShare.agentId: required non-empty string' }; + + const topic = str(o.topic); + if (!topic) return { ok: false, error: 'KnowledgeShare.topic: required non-empty string' }; + + if (o.content === undefined || o.content === null) { + return { ok: false, error: 'KnowledgeShare.content: required' }; + } + + const contentType = str(o.contentType); + if (contentType && !['text', 'json', 'reference'].includes(contentType)) { + return { ok: false, error: 'KnowledgeShare.contentType: must be text/json/reference' }; + } + + return { + ok: true, + value: { + kind: 'knowledge-share', + agentId, + ...(str(o.targetAgentId) ? { targetAgentId: str(o.targetAgentId) } : {}), + topic, + content: o.content, + ...(contentType ? { contentType } : {}), + ...(str(o.correlationId) ? { correlationId: str(o.correlationId) } : {}), + ts: num(o.ts, Date.now()), + }, + }; +} + +/** + * 通用 Agent 消息验证 — 根据 kind 字段分派 + * @param {unknown} msg + * @returns {ValidationResult} + */ +export function validateAgentMessage(msg) { + const o = obj(msg); + if (!o) return { ok: false, error: 'AgentMessage: expected object' }; + + const kind = str(o.kind); + if (!kind || !AgentMessageKind.includes(/** @type {any} */ (kind))) { + return { ok: false, error: `AgentMessage.kind: must be one of ${AgentMessageKind.join(', ')}` }; + } + + switch (kind) { + case 'task-request': return validateTaskRequest(o); + case 'task-result': return validateTaskResult(o); + case 'status-update': return validateStatusUpdate(o); + case 'knowledge-share': return validateKnowledgeShare(o); + default: + return { ok: false, error: `AgentMessage.kind: unknown kind "${kind}"` }; + } +} + +// ─── 工厂函数（便捷创建） ───────────────────────────────── + +/** + * 创建 TaskRequest + * @param {string} agentId + * @param {string} taskType - domain:action 格式 + * @param {unknown} payload + * @param {Partial>} [opts] + * @returns {TaskRequest} + */ +export function createTaskRequest(agentId, taskType, payload, opts = {}) { + return { + kind: 'task-request', + agentId, + ...(opts.targetAgentId ? { targetAgentId: opts.targetAgentId } : {}), + taskType, + payload, + ...(opts.correlationId ? { correlationId: opts.correlationId } : {}), + priority: opts.priority ?? 5, + ...(opts.timeoutMs ? { timeoutMs: opts.timeoutMs } : {}), + ts: Date.now(), + }; +} + +/** + * 创建 TaskResult + * @param {string} agentId + * @param {string} correlationId + * @param {typeof TaskStatus[number]} status + * @param {Partial>} [opts] + * @returns {TaskResult} + */ +export function createTaskResult(agentId, correlationId, status, opts = {}) { + return { + kind: 'task-result', + agentId, + correlationId, + status, + ...(opts.data !== undefined ? { data: opts.data } : {}), + ...(opts.error ? { error: opts.error } : {}), + ...(opts.durationMs !== undefined ? { durationMs: opts.durationMs } : {}), + ts: Date.now(), + }; +} + +/** + * 创建 StatusUpdate + * @param {string} agentId + * @param {typeof AgentRunStatus[number]} status + * @param {Partial>} [opts] + * @returns {StatusUpdate} + */ +export function createStatusUpdate(agentId, status, opts = {}) { + return { + kind: 'status-update', + agentId, + status, + ...(opts.currentTask ? { currentTask: opts.currentTask } : {}), + ...(opts.progress !== undefined ? { progress: opts.progress } : {}), + ...(opts.meta ? { meta: opts.meta } : {}), + ts: Date.now(), + }; +} + +/** + * 创建 KnowledgeShare + * @param {string} agentId + * @param {string} topic + * @param {unknown} content + * @param {Partial>} [opts] + * @returns {KnowledgeShare} + */ +export function createKnowledgeShare(agentId, topic, content, opts = {}) { + return { + kind: 'knowledge-share', + agentId, + ...(opts.targetAgentId ? { targetAgentId: opts.targetAgentId } : {}), + topic, + content, + ...(opts.contentType ? { contentType: opts.contentType } : {}), + ...(opts.correlationId ? { correlationId: opts.correlationId } : {}), + ts: Date.now(), + }; +} diff --git a/js/agents/core/contracts/index.js b/js/agents/core/contracts/index.js index 542c2f2b..9d2b386e 100644 --- a/js/agents/core/contracts/index.js +++ b/js/agents/core/contracts/index.js @@ -11,3 +11,17 @@ export { validateRpcRequest, validateRpcResponse } from "./rpc-message.js"; export { validateLlmResponse, validateToolCall } from "./llm-response.js"; export { validateToolResult, normalizeToolResult } from "./tool-result.js"; +export { + AgentMessageKind, + TaskStatus, + AgentRunStatus, + validateTaskRequest, + validateTaskResult, + validateStatusUpdate, + validateKnowledgeShare, + validateAgentMessage, + createTaskRequest, + createTaskResult, + createStatusUpdate, + createKnowledgeShare, +} from "./agent-message.js"; diff --git a/js/agents/core/index.js b/js/agents/core/index.js index 51c0917d..7558ee01 100644 --- a/js/agents/core/index.js +++ b/js/agents/core/index.js @@ -87,6 +87,18 @@ export { validateToolCall, validateToolResult, normalizeToolResult, + AgentMessageKind, + TaskStatus, + AgentRunStatus, + validateTaskRequest, + validateTaskResult, + validateStatusUpdate, + validateKnowledgeShare, + validateAgentMessage, + createTaskRequest, + createTaskResult, + createStatusUpdate, + createKnowledgeShare, } from './contracts/index.js'; // === Dependency Injection === diff --git a/tests/integration/agents/core/message-bus-agent-protocol.test.js b/tests/integration/agents/core/message-bus-agent-protocol.test.js new file mode 100644 index 00000000..b0efd20d --- /dev/null +++ b/tests/integration/agents/core/message-bus-agent-protocol.test.js @@ -0,0 +1,194 @@ +import { describe, it, expect, vi } from 'vitest'; + +// Mock lamport-clock to avoid TDZ issue (class used before declaration at module level) +vi.mock('../../../../js/agents/core/lamport-clock.js', () => { + let seq = 0; + return { + nextTick: () => ({ seq: ++seq, ts: Date.now(), id: `mock_${seq}` }), + sync: () => {}, + currentSeq: () => seq, + resetClock: () => { seq = 0; }, + compare: (a, b) => ((a?.seq ?? 0) - (b?.seq ?? 0)), + stampEvent: (e) => ({ ...e, _clock: { seq: ++seq, ts: Date.now(), id: `mock_${seq}` } }), + sortByLogicalOrder: (events) => [...(events || [])], + LamportClock: class LamportClock { + constructor() { this._seq = 0; } + tick() { return { seq: ++this._seq, ts: Date.now(), id: `lc_${this._seq}` }; } + update() { return this.tick(); } + get() { return { seq: this._seq, ts: Date.now(), id: `lc_${this._seq}` }; } + }, + LamportClockService: class LamportClockService { + constructor() { this._seq = 0; } + nextTick() { return { seq: ++this._seq, ts: Date.now(), id: `svc_${this._seq}` }; } + sync() {} + currentSeq() { return this._seq; } + resetClock() { this._seq = 0; } + }, + getDefaultClockService: () => null, + setDefaultClockService: () => {}, + default: { + nextTick: () => ({ seq: ++seq, ts: Date.now(), id: `mock_${seq}` }), + sync: () => {}, + currentSeq: () => seq, + resetClock: () => { seq = 0; }, + compare: () => 0, + stampEvent: (e) => e, + sortByLogicalOrder: (events) => [...(events || [])], + }, + }; +}); + +import { MessageBus } from '../../../../js/agents/core/message-bus.js'; +import { EventBus } from '../../../../js/agents/core/event-bus.js'; +import { + validateAgentMessage, + createTaskRequest, + createTaskResult, + createStatusUpdate, + createKnowledgeShare, +} from '../../../../js/agents/core/contracts/agent-message.js'; + +/** + * Integration: agent-message protocol over MessageBus RPC + * + * Uses snake_case event names per EventBus validation: /^[a-z0-9_]+([.:][a-z0-9_]+)*$/ + */ + +describe('MessageBus + agent-message protocol', () => { + it('task-request → task-result roundtrip via RPC', async () => { + const bus = new EventBus(); + const requester = new MessageBus(bus); + const worker = new MessageBus(bus); + + worker.on('agent:task_request', (payload) => { + const validated = validateAgentMessage(payload); + expect(validated.ok).toBe(true); + expect(validated.value.kind).toBe('task-request'); + + return createTaskResult( + 'worker-1', + validated.value.correlationId || 'no-corr', + 'completed', + { data: { found: 42 } }, + ); + }); + + const request = createTaskRequest('requester-1', 'search:execute', { q: 'test' }, { + correlationId: 'corr-001', + }); + + const result = await requester.request('agent:task_request', request, { timeoutMs: 2000 }); + const validated = validateAgentMessage({ ...result, kind: 'task-result' }); + expect(validated.ok).toBe(true); + expect(validated.value.status).toBe('completed'); + expect(validated.value.data).toEqual({ found: 42 }); + expect(validated.value.correlationId).toBe('corr-001'); + + requester.dispose(); + worker.dispose(); + }); + + it('status-update broadcast via emit/on', async () => { + const bus = new EventBus(); + const sender = new MessageBus(bus); + const listener = new MessageBus(bus); + + const received = []; + listener.on('agent:status_update', (payload) => { + const v = validateAgentMessage(payload); + if (v.ok) received.push(v.value); + }); + + sender.emit('agent:status_update', createStatusUpdate('agent-a', 'busy', { progress: 25 })); + sender.emit('agent:status_update', createStatusUpdate('agent-a', 'busy', { progress: 75 })); + sender.emit('agent:status_update', createStatusUpdate('agent-a', 'idle')); + + expect(received).toHaveLength(3); + expect(received[0].status).toBe('busy'); + expect(received[0].progress).toBe(25); + expect(received[1].progress).toBe(75); + expect(received[2].status).toBe('idle'); + + sender.dispose(); + listener.dispose(); + }); + + it('knowledge-share via channel routing', async () => { + const bus = new EventBus(); + const sharer = new MessageBus(bus); + const receiver = new MessageBus(bus); + + const received = []; + receiver.onChannel('research', 'agent:knowledge_share', (payload) => { + const v = validateAgentMessage(payload); + if (v.ok) received.push(v.value); + }); + + sharer.emit( + 'agent:knowledge_share', + createKnowledgeShare('agent-a', 'paper-summary', { abstract: 'AI is great' }, { contentType: 'json' }), + { channel: 'research' }, + ); + + expect(received).toHaveLength(1); + expect(received[0].topic).toBe('paper-summary'); + expect(received[0].contentType).toBe('json'); + + sharer.dispose(); + receiver.dispose(); + }); + + it('multi-agent task delegation chain', async () => { + const bus = new EventBus(); + const coordinator = new MessageBus(bus); + const searchAgent = new MessageBus(bus); + const analyzeAgent = new MessageBus(bus); + + searchAgent.on('agent:search', (payload) => { + const v = validateAgentMessage(payload); + if (!v.ok) throw new Error(v.error); + return createTaskResult('search-agent', v.value.correlationId || '', 'completed', { + data: { documents: ['doc1', 'doc2'] }, + }); + }); + + analyzeAgent.on('agent:analyze', (payload) => { + const v = validateAgentMessage(payload); + if (!v.ok) throw new Error(v.error); + return createTaskResult('analyze-agent', v.value.correlationId || '', 'completed', { + data: { summary: 'Two documents found' }, + }); + }); + + const searchResult = await coordinator.request( + 'agent:search', + createTaskRequest('coordinator', 'search:execute', { q: 'AI' }, { correlationId: 'step-1' }), + { timeoutMs: 2000 }, + ); + + const analyzeResult = await coordinator.request( + 'agent:analyze', + createTaskRequest('coordinator', 'analyze:summarize', searchResult.data, { correlationId: 'step-2' }), + { timeoutMs: 2000 }, + ); + + expect(analyzeResult.data).toEqual({ summary: 'Two documents found' }); + + coordinator.dispose(); + searchAgent.dispose(); + analyzeAgent.dispose(); + }); + + it('RPC timeout produces error for unhandled task types', async () => { + const bus = new EventBus(); + const sender = new MessageBus(bus); + + const request = createTaskRequest('a1', 'unknown:task', {}); + + await expect( + sender.request('agent:no_handler', request, { timeoutMs: 100 }), + ).rejects.toThrow(/timeout/i); + + sender.dispose(); + }); +}); diff --git a/tests/unit/agents/core/contracts/agent-message.test.js b/tests/unit/agents/core/contracts/agent-message.test.js new file mode 100644 index 00000000..70c76ba9 --- /dev/null +++ b/tests/unit/agents/core/contracts/agent-message.test.js @@ -0,0 +1,437 @@ +import { describe, it, expect } from 'vitest'; +import { + AgentMessageKind, + TaskStatus, + AgentRunStatus, + validateTaskRequest, + validateTaskResult, + validateStatusUpdate, + validateKnowledgeShare, + validateAgentMessage, + createTaskRequest, + createTaskResult, + createStatusUpdate, + createKnowledgeShare, +} from '../../../../../js/agents/core/contracts/agent-message.js'; + +// ─── Constants ────────────────────────────────────────── + +describe('AgentMessageKind', () => { + it('exposes four message kinds', () => { + expect(AgentMessageKind).toEqual([ + 'task-request', + 'task-result', + 'status-update', + 'knowledge-share', + ]); + }); +}); + +describe('TaskStatus', () => { + it('exposes five statuses', () => { + expect(TaskStatus).toEqual(['pending', 'running', 'completed', 'failed', 'cancelled']); + }); +}); + +describe('AgentRunStatus', () => { + it('exposes five run statuses', () => { + expect(AgentRunStatus).toEqual(['idle', 'busy', 'degraded', 'stopping', 'stopped']); + }); +}); + +// ─── validateTaskRequest ──────────────────────────────── + +describe('validateTaskRequest', () => { + it('rejects non-object inputs', () => { + for (const v of [null, undefined, '', 42, true]) { + expect(validateTaskRequest(v)).toEqual({ ok: false, error: 'TaskRequest: expected object' }); + } + }); + + it('rejects missing agentId', () => { + expect(validateTaskRequest({ taskType: 'search:execute', payload: {} })).toEqual({ + ok: false, + error: 'TaskRequest.agentId: required non-empty string', + }); + }); + + it('rejects missing taskType', () => { + expect(validateTaskRequest({ agentId: 'a1', payload: {} })).toEqual({ + ok: false, + error: 'TaskRequest.taskType: required non-empty string', + }); + }); + + it('rejects invalid taskType format', () => { + const invalid = ['ping', 'Ping:action', 'ping:Action', '1x:y', 'x:1y']; + for (const taskType of invalid) { + const r = validateTaskRequest({ agentId: 'a1', taskType, payload: null }); + expect(r.ok).toBe(false); + expect(r.error).toContain('domain:action'); + } + }); + + it('rejects out-of-range priority', () => { + const r = validateTaskRequest({ agentId: 'a1', taskType: 'search:execute', payload: null, priority: 11 }); + expect(r).toEqual({ ok: false, error: 'TaskRequest.priority: must be 0-10' }); + }); + + it('rejects non-positive timeoutMs', () => { + const r = validateTaskRequest({ agentId: 'a1', taskType: 'search:execute', payload: null, timeoutMs: 0 }); + expect(r).toEqual({ ok: false, error: 'TaskRequest.timeoutMs: must be positive' }); + }); + + it('accepts valid minimal request', () => { + const r = validateTaskRequest({ agentId: 'a1', taskType: 'search:execute', payload: { q: 'test' } }); + expect(r.ok).toBe(true); + expect(r.value.kind).toBe('task-request'); + expect(r.value.agentId).toBe('a1'); + expect(r.value.taskType).toBe('search:execute'); + expect(r.value.payload).toEqual({ q: 'test' }); + expect(r.value.priority).toBe(5); + expect(typeof r.value.ts).toBe('number'); + }); + + it('preserves optional fields', () => { + const r = validateTaskRequest({ + agentId: 'a1', + targetAgentId: 'b2', + taskType: 'search:execute', + payload: null, + correlationId: 'c1', + priority: 0, + timeoutMs: 5000, + ts: 1000, + }); + expect(r.ok).toBe(true); + expect(r.value.targetAgentId).toBe('b2'); + expect(r.value.correlationId).toBe('c1'); + expect(r.value.priority).toBe(0); + expect(r.value.timeoutMs).toBe(5000); + expect(r.value.ts).toBe(1000); + }); + + it('defaults priority to 5 when not provided', () => { + const r = validateTaskRequest({ agentId: 'a1', taskType: 'search:execute', payload: null }); + expect(r.ok).toBe(true); + expect(r.value.priority).toBe(5); + }); +}); + +// ─── validateTaskResult ───────────────────────────────── + +describe('validateTaskResult', () => { + it('rejects non-object inputs', () => { + expect(validateTaskResult(null)).toEqual({ ok: false, error: 'TaskResult: expected object' }); + }); + + it('rejects missing agentId', () => { + expect(validateTaskResult({ correlationId: 'c1', status: 'completed' })).toEqual({ + ok: false, + error: 'TaskResult.agentId: required non-empty string', + }); + }); + + it('rejects missing correlationId', () => { + expect(validateTaskResult({ agentId: 'a1', status: 'completed' })).toEqual({ + ok: false, + error: 'TaskResult.correlationId: required non-empty string', + }); + }); + + it('rejects invalid status', () => { + const r = validateTaskResult({ agentId: 'a1', correlationId: 'c1', status: 'unknown' }); + expect(r.ok).toBe(false); + expect(r.error).toContain('must be one of'); + }); + + it('accepts all valid statuses', () => { + for (const status of TaskStatus) { + const r = validateTaskResult({ agentId: 'a1', correlationId: 'c1', status }); + expect(r.ok).toBe(true); + expect(r.value.status).toBe(status); + } + }); + + it('preserves optional data/error/durationMs', () => { + const r = validateTaskResult({ + agentId: 'a1', + correlationId: 'c1', + status: 'completed', + data: { results: [1, 2] }, + error: 'partial', + durationMs: 150, + ts: 2000, + }); + expect(r.ok).toBe(true); + expect(r.value.data).toEqual({ results: [1, 2] }); + expect(r.value.error).toBe('partial'); + expect(r.value.durationMs).toBe(150); + expect(r.value.ts).toBe(2000); + }); +}); + +// ─── validateStatusUpdate ─────────────────────────────── + +describe('validateStatusUpdate', () => { + it('rejects non-object inputs', () => { + expect(validateStatusUpdate(42)).toEqual({ ok: false, error: 'StatusUpdate: expected object' }); + }); + + it('rejects missing agentId', () => { + expect(validateStatusUpdate({ status: 'idle' })).toEqual({ + ok: false, + error: 'StatusUpdate.agentId: required non-empty string', + }); + }); + + it('rejects invalid status', () => { + const r = validateStatusUpdate({ agentId: 'a1', status: 'exploding' }); + expect(r.ok).toBe(false); + expect(r.error).toContain('must be one of'); + }); + + it('rejects out-of-range progress', () => { + expect(validateStatusUpdate({ agentId: 'a1', status: 'busy', progress: 101 })).toEqual({ + ok: false, + error: 'StatusUpdate.progress: must be 0-100', + }); + expect(validateStatusUpdate({ agentId: 'a1', status: 'busy', progress: -1 })).toEqual({ + ok: false, + error: 'StatusUpdate.progress: must be 0-100', + }); + }); + + it('accepts all valid run statuses', () => { + for (const status of AgentRunStatus) { + const r = validateStatusUpdate({ agentId: 'a1', status }); + expect(r.ok).toBe(true); + expect(r.value.status).toBe(status); + } + }); + + it('preserves optional fields', () => { + const r = validateStatusUpdate({ + agentId: 'a1', + status: 'busy', + currentTask: 'Analyzing documents', + progress: 75, + meta: { tokensUsed: 1234 }, + ts: 3000, + }); + expect(r.ok).toBe(true); + expect(r.value.currentTask).toBe('Analyzing documents'); + expect(r.value.progress).toBe(75); + expect(r.value.meta).toEqual({ tokensUsed: 1234 }); + }); +}); + +// ─── validateKnowledgeShare ───────────────────────────── + +describe('validateKnowledgeShare', () => { + it('rejects non-object inputs', () => { + expect(validateKnowledgeShare(null)).toEqual({ ok: false, error: 'KnowledgeShare: expected object' }); + }); + + it('rejects missing agentId', () => { + expect(validateKnowledgeShare({ topic: 'findings', content: 'data' })).toEqual({ + ok: false, + error: 'KnowledgeShare.agentId: required non-empty string', + }); + }); + + it('rejects missing topic', () => { + expect(validateKnowledgeShare({ agentId: 'a1', content: 'data' })).toEqual({ + ok: false, + error: 'KnowledgeShare.topic: required non-empty string', + }); + }); + + it('rejects missing content', () => { + expect(validateKnowledgeShare({ agentId: 'a1', topic: 'findings' })).toEqual({ + ok: false, + error: 'KnowledgeShare.content: required', + }); + }); + + it('rejects invalid contentType', () => { + const r = validateKnowledgeShare({ agentId: 'a1', topic: 'x', content: 'y', contentType: 'binary' }); + expect(r.ok).toBe(false); + expect(r.error).toContain('text/json/reference'); + }); + + it('accepts valid contentTypes', () => { + for (const contentType of ['text', 'json', 'reference']) { + const r = validateKnowledgeShare({ agentId: 'a1', topic: 'x', content: 'y', contentType }); + expect(r.ok).toBe(true); + expect(r.value.contentType).toBe(contentType); + } + }); + + it('preserves optional fields', () => { + const r = validateKnowledgeShare({ + agentId: 'a1', + targetAgentId: 'b2', + topic: 'findings', + content: { key: 'value' }, + contentType: 'json', + correlationId: 'c1', + ts: 4000, + }); + expect(r.ok).toBe(true); + expect(r.value.targetAgentId).toBe('b2'); + expect(r.value.correlationId).toBe('c1'); + expect(r.value.content).toEqual({ key: 'value' }); + }); +}); + +// ─── validateAgentMessage (dispatch) ──────────────────── + +describe('validateAgentMessage', () => { + it('rejects non-object inputs', () => { + expect(validateAgentMessage(null)).toEqual({ ok: false, error: 'AgentMessage: expected object' }); + }); + + it('rejects missing or invalid kind', () => { + expect(validateAgentMessage({ agentId: 'a1' })).toEqual({ + ok: false, + error: expect.stringContaining('AgentMessage.kind'), + }); + expect(validateAgentMessage({ kind: 'unknown', agentId: 'a1' })).toEqual({ + ok: false, + error: expect.stringContaining('must be one of'), + }); + }); + + it('dispatches to correct validator based on kind', () => { + const req = validateAgentMessage({ + kind: 'task-request', + agentId: 'a1', + taskType: 'search:execute', + payload: null, + }); + expect(req.ok).toBe(true); + expect(req.value.kind).toBe('task-request'); + + const res = validateAgentMessage({ + kind: 'task-result', + agentId: 'a1', + correlationId: 'c1', + status: 'completed', + }); + expect(res.ok).toBe(true); + expect(res.value.kind).toBe('task-result'); + + const upd = validateAgentMessage({ + kind: 'status-update', + agentId: 'a1', + status: 'idle', + }); + expect(upd.ok).toBe(true); + expect(upd.value.kind).toBe('status-update'); + + const ks = validateAgentMessage({ + kind: 'knowledge-share', + agentId: 'a1', + topic: 'findings', + content: 'data', + }); + expect(ks.ok).toBe(true); + expect(ks.value.kind).toBe('knowledge-share'); + }); + + it('propagates validation errors from sub-validators', () => { + const r = validateAgentMessage({ + kind: 'task-request', + agentId: 'a1', + taskType: 'bad-format', + payload: null, + }); + expect(r.ok).toBe(false); + expect(r.error).toContain('domain:action'); + }); +}); + +// ─── Factory functions ────────────────────────────────── + +describe('createTaskRequest', () => { + it('creates a valid TaskRequest', () => { + const msg = createTaskRequest('a1', 'search:execute', { q: 'test' }); + expect(msg.kind).toBe('task-request'); + expect(msg.agentId).toBe('a1'); + expect(msg.taskType).toBe('search:execute'); + expect(msg.payload).toEqual({ q: 'test' }); + expect(msg.priority).toBe(5); + expect(typeof msg.ts).toBe('number'); + + const validated = validateTaskRequest(msg); + expect(validated.ok).toBe(true); + }); + + it('passes optional fields through', () => { + const msg = createTaskRequest('a1', 'search:execute', null, { + targetAgentId: 'b2', + correlationId: 'c1', + priority: 1, + timeoutMs: 5000, + }); + expect(msg.targetAgentId).toBe('b2'); + expect(msg.correlationId).toBe('c1'); + expect(msg.priority).toBe(1); + expect(msg.timeoutMs).toBe(5000); + }); +}); + +describe('createTaskResult', () => { + it('creates a valid TaskResult', () => { + const msg = createTaskResult('a1', 'c1', 'completed', { data: { ok: true } }); + expect(msg.kind).toBe('task-result'); + expect(msg.status).toBe('completed'); + expect(msg.data).toEqual({ ok: true }); + + const validated = validateTaskResult(msg); + expect(validated.ok).toBe(true); + }); +}); + +describe('createStatusUpdate', () => { + it('creates a valid StatusUpdate', () => { + const msg = createStatusUpdate('a1', 'busy', { progress: 50 }); + expect(msg.kind).toBe('status-update'); + expect(msg.status).toBe('busy'); + expect(msg.progress).toBe(50); + + const validated = validateStatusUpdate(msg); + expect(validated.ok).toBe(true); + }); +}); + +describe('createKnowledgeShare', () => { + it('creates a valid KnowledgeShare', () => { + const msg = createKnowledgeShare('a1', 'findings', { key: 'value' }, { contentType: 'json' }); + expect(msg.kind).toBe('knowledge-share'); + expect(msg.topic).toBe('findings'); + expect(msg.contentType).toBe('json'); + + const validated = validateKnowledgeShare(msg); + expect(validated.ok).toBe(true); + }); +}); + +// ─── Roundtrip: factory → validate ────────────────────── + +describe('roundtrip: factory → validateAgentMessage', () => { + it('all factory outputs pass validateAgentMessage', () => { + const messages = [ + createTaskRequest('a1', 'search:execute', { q: 'test' }), + createTaskResult('a1', 'c1', 'completed'), + createStatusUpdate('a1', 'idle'), + createKnowledgeShare('a1', 'findings', 'some text'), + ]; + + for (const msg of messages) { + const r = validateAgentMessage(msg); + expect(r.ok).toBe(true); + } + }); +});