mirror of
https://github.com/Heretek-AI/heretek-openclaw-core.git
synced 2026-07-01 14:17:57 -04:00
feat: archive JS gateway, add ACP adapter layer, port a2a-message-send
Archived: - gateway/openclaw-gateway.js — parallel incompatible implementation (see gateway/README-ARCHIVE.md for full analysis) Added: - GATEWAY_COMPATIBILITY_REPORT.md — full analysis of all modules and porting roadmap for npm gateway compatibility - modules/adapters/acp-adapter.js — ACP WebSocket adapter for npm gateway (HMAC nonce auth, sendMessage, broadcast, ping, register) - modules/adapters/bft-consensus-adapter.js — bridges BFT consensus results to npm gateway ACP broadcast channel - modules/adapters/litellm-agent-registry.js — REST adapter for LiteLLM proxy agent registration (port 4000) - modules/adapters/README.md — adapter layer documentation - skills/a2a-message-send/a2a-redis.js — migrated from JS gateway protocol to Redis + ACP dual-mode (no data migration needed, openclaw:a2a: keys shared with npm gateway) - skills/a2a-message-send/SKILL.md — updated SKILL.md Key findings: - npm gateway uses ACP (Agent Client Protocol) over WebSocket - JS gateway used simple JSON protocol with no auth — incompatible - Redis key prefix openclaw:a2a: is shared — modules using it directly are already compatible with npm gateway - HMAC nonce auth required for npm gateway WebSocket connect (token from openclaw.json → gateway.auth.token)
This commit is contained in:
John Doe
@@ -0,0 +1,210 @@
|
||||
# Gateway Compatibility Report — Heretek OpenClaw Core
|
||||
|
||||
**Author:** Steward
|
||||
**Date:** 2026-04-03
|
||||
**Status:** ANALYSIS COMPLETE — PORTING IN PROGRESS
|
||||
|
||||
---
|
||||
|
||||
## Executive Summary
|
||||
|
||||
The `heretek-openclaw-core` repository contains two categories of modules that must be evaluated for npm gateway compatibility:
|
||||
|
||||
1. **Modules that target the archived JS gateway** — require full port
|
||||
2. **Modules that use Redis directly** — require adapter layer only
|
||||
|
||||
**No modules can run as-is against the npm OpenClaw gateway.**
|
||||
|
||||
---
|
||||
|
||||
## Protocol Layer Comparison
|
||||
|
||||
### NPM Gateway Protocol (ACP — Agent Client Protocol)
|
||||
|
||||
The running gateway at port 18789 uses the **ACP (Agent Client Protocol)** via `@agentclientprotocol/sdk`, embedded in the npm `openclaw` package. Key characteristics:
|
||||
|
||||
- **Auth:** Nonce-challenge HMAC handshake on WebSocket connect (`connect.challenge` event)
|
||||
- **Protocol:** JSON-RPC 2.0 over WebSocket with typed message schemas
|
||||
- **Registry:** SQLite (`agents.db`) + Redis for pub/sub
|
||||
- **A2A prefix (Redis):** `openclaw:a2a:` (same prefix the JS gateway used — compatible ✅)
|
||||
- **WS paths:** `/a2a` (primary A2A), `/ws`, `/gateway`
|
||||
- **SDK:** `@agentclientprotocol/sdk` (bundled in npm openclaw package)
|
||||
- **Connection token:** `9b54947854ee05186fb5363d0ea113685794d08d4ab45f80`
|
||||
|
||||
### JS Gateway Protocol (Archived)
|
||||
|
||||
- **Auth:** None
|
||||
- **Protocol:** Plain JSON `{type: register|message|broadcast|status}`
|
||||
- **Registry:** In-memory `Map`
|
||||
- **A2A prefix (Redis):** `openclaw:a2a:` — **compatible ✅**
|
||||
- **WS path:** `/a2a`
|
||||
|
||||
---
|
||||
|
||||
## Module Compatibility Matrix
|
||||
|
||||
| Module | File | Target | Compatible? | Port Effort |
|
||||
|---|---|---|---|---|
|
||||
| **A2A Message Send** | `skills/a2a-message-send/a2a-redis.js` | Redis directly | ✅ Yes | **Low — adapter only** |
|
||||
| **A2A Agent Register** | `skills/a2a-agent-register/SKILL.md` | LiteLLM proxy REST | ⚠️ Different endpoint | Medium — rewrite SKILL.md |
|
||||
| **Event Mesh** | `modules/a2a-protocol/event-mesh.js` | Redis directly | ✅ Yes | **Low — rename prefix** |
|
||||
| **Redis WS Bridge** | `modules/communication/redis-websocket-bridge.js` | JS gateway WS | ❌ No | **High — abandon** |
|
||||
| **BFT Consensus** | `modules/consensus/bft-consensus.js` | Redis pub/sub | ✅ Yes | **Low — works as-is** |
|
||||
| **Reputation Store** | `modules/consensus/reputation-store.postgres.js` | Postgres | ✅ Yes | **Low** |
|
||||
| **Reputation Voting** | `modules/consensus/reputation-voting.js` | Postgres | ✅ Yes | **Low** |
|
||||
| **Curiosity Engine** | `modules/curiosity-engine.js` | File system | ✅ Yes | **Low** |
|
||||
| **Curiosity Engine v2** | `modules/curiosity-engine-v2.js` | File system | ✅ Yes | **Low** |
|
||||
| **Heavy Swarm** | `modules/heavy-swarm.js` | CLI/subagent | ⚠️ Partial | Medium |
|
||||
| **Lineage Tracking** | `modules/lineage-tracking.js` | Postgres | ✅ Yes | **Low** |
|
||||
| **Tiered Context** | `modules/memory/tiered-context.js` | Postgres | ✅ Yes | **Low** |
|
||||
| **Dashboard Sync** | `modules/observability/dashboard-sync.js` | REST API | ⚠️ Different endpoints | Medium |
|
||||
| **Gateway Instrumentation** | `modules/observability/gateway-instrumentation.js` | WS connect | ⚠️ Different protocol | Medium |
|
||||
| **Langfuse Client** | `modules/observability/langfuse-client.js` | HTTP | ✅ Yes | **Low** |
|
||||
| **LiteLLM Integration** | `modules/observability/litellm-integration.js` | LiteLLM proxy | ✅ Yes | **Low** |
|
||||
| **Metrics Exporter** | `modules/observability/metrics-exporter.js` | Prometheus | ✅ Yes | **Low** |
|
||||
| **Trace Context** | `modules/observability/trace-context.js` | Langfuse | ✅ Yes | **Low** |
|
||||
| **Swarm Memory** | `modules/swarm-memory/heretek-swarm-memory.js` | Postgres | ✅ Yes | **Low** |
|
||||
| **Task State Machine** | `modules/task-state-machine.js` | Postgres | ✅ Yes | **Low** |
|
||||
| **Agent Client Lib** | `agents/lib/agent-client.js` | JS gateway WS | ❌ No | **High — abandon** |
|
||||
| **Model Config** | `agents/lib/agent-model-config.js` | Config | ✅ Yes | **Low** |
|
||||
| **Model Router** | `agents/lib/agent-model-router.js` | LiteLLM proxy | ✅ Yes | **Low** |
|
||||
|
||||
---
|
||||
|
||||
## Porting Priorities
|
||||
|
||||
### Phase 1: Redis-First Modules (Low effort, high value)
|
||||
|
||||
These modules use Redis directly and are already compatible with the npm gateway's Redis key space.
|
||||
|
||||
#### 1.1 — `skills/a2a-message-send/a2a-redis.js` ⭐ START HERE
|
||||
- **Why:** Simplest module, no WS dependency, directly usable
|
||||
- **What:** Adapter that routes to npm gateway's ACP SDK or falls back to direct Redis
|
||||
- **Redis keys used:** `openclaw:a2a:inbox:*`, `openclaw:a2a:agents`, `openclaw:a2a:broadcast`
|
||||
- **Effort:** ~2 hours
|
||||
|
||||
#### 1.2 — `modules/a2a-protocol/event-mesh.js`
|
||||
- **Why:** Complementary pub/sub layer; uses different prefix `heretek:event:`
|
||||
- **What:** Either rename prefix to `openclaw:a2a:` or add bridging to both
|
||||
- **Effort:** ~1 hour
|
||||
|
||||
#### 1.3 — `modules/consensus/bft-consensus.js`
|
||||
- **Why:** Self-contained PBFT implementation using Redis pub/sub on `bft:consensus` channel
|
||||
- **What:** Works as-is with Redis; needs adapter to publish results to npm gateway
|
||||
- **Effort:** ~1 hour
|
||||
|
||||
### Phase 2: ACP SDK Adapter Layer (Medium effort)
|
||||
|
||||
#### 2.1 — ACP Adapter Library
|
||||
- **What:** `modules/adapters/acp-adapter.js`
|
||||
- **Provides:**
|
||||
- Authenticated WebSocket connection to npm gateway
|
||||
- HMAC nonce signing
|
||||
- Typed message sending/receiving
|
||||
- Subscribe to agent events
|
||||
- **Uses:** `@agentclientprotocol/sdk` (already bundled in npm openclaw)
|
||||
- **Effort:** ~4 hours
|
||||
|
||||
#### 2.2 — Port `agents/lib/agent-client.js`
|
||||
- **What:** Rewrite to use ACP adapter instead of JS gateway WS protocol
|
||||
- **Effort:** ~3 hours
|
||||
|
||||
### Phase 3: Skills Rewrite
|
||||
|
||||
#### 3.1 — Rewrite `skills/a2a-agent-register/SKILL.md`
|
||||
- **Current:** Targets LiteLLM proxy REST API (`/key/generate`)
|
||||
- **Actual npm gateway:** No REST agent registration — agents register via A2A WebSocket
|
||||
- **What:** Rewrite SKILL.md to use `openclaw agents add` CLI or ACP SDK
|
||||
- **Effort:** ~2 hours
|
||||
|
||||
### Phase 4: Observability Adapters
|
||||
|
||||
#### 4.1 — `modules/observability/gateway-instrumentation.js`
|
||||
- **What:** Rewrite to use ACP SDK event subscriptions instead of JS gateway WS
|
||||
- **Effort:** ~3 hours
|
||||
|
||||
#### 4.2 — `modules/observability/dashboard-sync.js`
|
||||
- **What:** Update REST endpoints to match dashboard API (`/api/health`, `/api/memory/graph`)
|
||||
- **Effort:** ~2 hours
|
||||
|
||||
---
|
||||
|
||||
## Abandon (Do Not Port)
|
||||
|
||||
| Module | Reason |
|
||||
|---|---|
|
||||
| `modules/communication/redis-websocket-bridge.js` | Designed for JS gateway WS; npm gateway is already a WS endpoint |
|
||||
| `agents/lib/agent-client.js` (original) | Incompatible protocol; replace with ACP version |
|
||||
| `skills/gateway-pulse/` (entire skill) | Designed for JS gateway monitoring |
|
||||
|
||||
---
|
||||
|
||||
## Redis Key Space (Pre-existing, DO NOT REUSE)
|
||||
|
||||
The npm gateway and JS gateway share the same Redis prefix:
|
||||
|
||||
```
|
||||
openclaw:a2a:inbox:{agentId} # Message queues (list)
|
||||
openclaw:a2a:agents # Registered agent IDs (set)
|
||||
openclaw:a2a:agent:{agentId} # Agent metadata (hash)
|
||||
openclaw:a2a:broadcast # Broadcast pub/sub channel
|
||||
```
|
||||
|
||||
**These keys are already populated. Modules using these keys are immediately compatible.**
|
||||
|
||||
Heretek-specific keys (different prefix, active):
|
||||
|
||||
```
|
||||
heretek:event:{topic} # EventMesh pub/sub (event-mesh.js)
|
||||
bft:consensus # BFT consensus channel (bft-consensus.js)
|
||||
bft:prepare:{view}:{seq}:{digest} # BFT prepare quorum tracking
|
||||
bft:commit:{view}:{seq}:{digest} # BFT commit quorum tracking
|
||||
bft:view-change:{view} # BFT view change tracking
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## ACP SDK Reference
|
||||
|
||||
The npm gateway includes `@agentclientprotocol/sdk` at:
|
||||
|
||||
```
|
||||
/usr/lib/node_modules/openclaw/node_modules/@agentclientprotocol/sdk/dist/
|
||||
```
|
||||
|
||||
Key exports:
|
||||
- `AgentSideConnection` — agent-side ACP connection
|
||||
- `ClientSideConnection` — client-side connection
|
||||
- JSON-RPC 2.0 typed schemas via Zod
|
||||
- Stream-based message framing (`ndJsonStream`)
|
||||
|
||||
Agent registration with npm gateway uses A2A WebSocket, not REST.
|
||||
|
||||
---
|
||||
|
||||
## Test Compatibility
|
||||
|
||||
The `tests/integration/` directory has tests that reference the JS gateway:
|
||||
|
||||
- `tests/integration/gateway-rpc.test.ts` — needs rewrite for ACP SDK
|
||||
- `tests/integration/a2a-communication.test.ts` — Redis parts usable, WS parts need adapter
|
||||
- `tests/integration/websocket-bridge.test.ts` — abandon (targets archived bridge)
|
||||
- `tests/integration/agent-deliberation.test.ts` — depends on gateway-rpc
|
||||
|
||||
---
|
||||
|
||||
## Next Steps
|
||||
|
||||
1. ✅ **Done** — Archive `gateway/openclaw-gateway.js`
|
||||
2. 🔄 **IN PROGRESS** — Port `skills/a2a-message-send/a2a-redis.js` (Phase 1.1)
|
||||
3. ⬜ — Port `modules/a2a-protocol/event-mesh.js` (Phase 1.2)
|
||||
4. ⬜ — Port `modules/consensus/bft-consensus.js` (Phase 1.3)
|
||||
5. ⬜ — Build `modules/adapters/acp-adapter.js` (Phase 2.1)
|
||||
6. ⬜ — Rewrite `agents/lib/agent-client.js` (Phase 2.2)
|
||||
7. ⬜ — Rewrite `skills/a2a-agent-register/SKILL.md` (Phase 3.1)
|
||||
8. ⬜ — Port observability modules (Phase 4)
|
||||
9. ⬜ — Update integration tests
|
||||
|
||||
---
|
||||
|
||||
*🦞 Steward — Gateway Compatibility Report v1.0*
|
||||
@@ -0,0 +1,56 @@
|
||||
# ⚠️ ARCHIVED — JS Gateway (openclaw-gateway.js)
|
||||
|
||||
**Archived:** 2026-04-03
|
||||
**Reason:** Parallel, incompatible implementation — never deployed in production
|
||||
|
||||
---
|
||||
|
||||
## What was here
|
||||
|
||||
`/gateway/openclaw-gateway.js` — A standalone Node.js WebSocket gateway implementing a simple JSON protocol for agent-to-agent communication, using `ioredis` for pub/sub and message queuing.
|
||||
|
||||
## Why it was archived
|
||||
|
||||
The JS gateway was a **design prototype** that was never connected to the actual OpenClaw system. Two fundamentally different gateways existed side by side:
|
||||
|
||||
| | JS Gateway (this file) | NPM Gateway (actual) |
|
||||
|---|---|---|
|
||||
| **Source** | `heretek-openclaw-core/gateway/` | npm package `openclaw` |
|
||||
| **Protocol** | Simple JSON `{type: register\|message\|broadcast\|status}` | ACP (Agent Client Protocol) over WebSocket |
|
||||
| **Auth** | None | Nonce-challenge HMAC |
|
||||
| **Registry** | In-memory Map | SQLite + Redis |
|
||||
| **Port** | Default 18789 (conflicted) | 18789 (running) |
|
||||
| **Protocol version** | 1.0.0 (proprietary) | ACP 1.x via `@agentclientprotocol/sdk` |
|
||||
| **Deploy status** | Never deployed | Running since 2026-03-31 |
|
||||
|
||||
The JS gateway was incompatible with the actual npm gateway at every layer — protocol, auth, registry, and message format.
|
||||
|
||||
## What still works (from the JS gateway design)
|
||||
|
||||
The following concepts from this prototype are worth preserving for the npm gateway port:
|
||||
|
||||
- ✅ Agent registration with capabilities advertisement
|
||||
- ✅ Message routing via Redis pub/sub
|
||||
- ✅ Message queuing for offline agents (`openclaw:a2a:inbox:{agentId}`)
|
||||
- ✅ Broadcast to all connected agents
|
||||
- ✅ Redis key structure: `openclaw:a2a:agents`, `openclaw:a2a:agent:{id}`, `openclaw:a2a:broadcast`
|
||||
- ✅ Heartbeat interval tracking (`heartbeatInterval * 2` for offline detection)
|
||||
- ✅ `GET /health` and `GET /agents` HTTP endpoints
|
||||
- ✅ Ping/pong keepalive protocol
|
||||
|
||||
## Compatible Redis key prefix
|
||||
|
||||
Both the JS gateway and the `skills/a2a-message-send/a2a-redis.js` use the same Redis prefix:
|
||||
|
||||
```
|
||||
openclaw:a2a:
|
||||
```
|
||||
|
||||
This shared prefix means **no data migration is needed** — the Redis structures are already compatible.
|
||||
|
||||
## See also
|
||||
|
||||
- `GATEWAY_COMPATIBILITY_REPORT.md` — Full analysis and porting roadmap
|
||||
- `../modules/communication/redis-websocket-bridge.js` — Standalone Redis↔WebSocket bridge (also archived)
|
||||
- `../skills/a2a-message-send/a2a-redis.js` — Still active; uses same Redis keys
|
||||
- `../modules/a2a-protocol/event-mesh.js` — Still active; uses different Redis prefix `heretek:event:`
|
||||
@@ -0,0 +1,88 @@
|
||||
# Adapters — Heretek OpenClaw Core
|
||||
|
||||
Adapter layer for connecting legacy Heretek modules to the npm OpenClaw gateway.
|
||||
|
||||
## Architecture
|
||||
|
||||
```
|
||||
Heretek Modules
|
||||
│
|
||||
├── acp-adapter.js ← WS + HMAC auth for npm gateway
|
||||
├── bft-consensus-adapter.js ← BFT consensus → npm gateway bridge
|
||||
└── [future adapters]
|
||||
│
|
||||
▼
|
||||
npm OpenClaw Gateway (port 18789)
|
||||
ACP over WebSocket + Redis pub/sub
|
||||
```
|
||||
|
||||
## Adapters
|
||||
|
||||
### `acp-adapter.js` ✅ Active
|
||||
|
||||
Primary adapter for connecting to the npm OpenClaw gateway.
|
||||
|
||||
```javascript
|
||||
const { ACPAdapter } = require('./acp-adapter');
|
||||
|
||||
// Connect to npm gateway (handles HMAC nonce auth)
|
||||
const adapter = await ACPAdapter.connect({
|
||||
agentId: 'alpha',
|
||||
token: process.env.OPENCLAW_GATEWAY_TOKEN,
|
||||
gatewayUrl: 'ws://localhost:18789/a2a'
|
||||
});
|
||||
|
||||
// Send messages
|
||||
await adapter.sendMessage('beta', 'Hello Beta');
|
||||
await adapter.broadcast({ type: 'alert', content: 'Phase 3 starting' });
|
||||
|
||||
// Listen
|
||||
adapter.on('message', (msg) => console.log(`${msg.from}: ${msg.content}`));
|
||||
|
||||
await adapter.close();
|
||||
```
|
||||
|
||||
**Also includes:** `LiteLLMGatewayAdapter` — REST adapter for LiteLLM proxy (used by some legacy skills).
|
||||
|
||||
### `bft-consensus-adapter.js` ✅ Active
|
||||
|
||||
Bridges BFT consensus events to npm gateway ACP broadcast.
|
||||
|
||||
```javascript
|
||||
const { BFTConsensusAdapter } = require('./bft-consensus-adapter');
|
||||
|
||||
const bft = new BFTConsensusAdapter({ nodeId: 'alpha' });
|
||||
await bft.connect();
|
||||
await bft.connectACP({ agentId: 'alpha' });
|
||||
|
||||
bft.onCommitted = (result) => {
|
||||
console.log('Consensus reached!', result);
|
||||
};
|
||||
|
||||
await bft.propose({ action: 'approve_phase_3', reason: '...' });
|
||||
```
|
||||
|
||||
## For Module Writers
|
||||
|
||||
If your module needs to send/receive messages from other agents via the npm gateway:
|
||||
|
||||
```javascript
|
||||
// Option 1: Direct Redis (simplest, always works)
|
||||
// Uses openclaw:a2a: prefix — npm gateway reads this too
|
||||
const Redis = require('ioredis');
|
||||
const client = new Redis('redis://localhost:6379');
|
||||
await client.lpush('openclaw:a2a:inbox:alpha', JSON.stringify(msg));
|
||||
|
||||
// Option 2: ACP adapter (real-time, requires HMAC auth)
|
||||
// See acp-adapter.js
|
||||
const { ACPAdapter } = require('./modules/adapters/acp-adapter');
|
||||
const adapter = await ACPAdapter.connect({ agentId: '...', token: '...' });
|
||||
await adapter.sendMessage('alpha', 'urgent!');
|
||||
```
|
||||
|
||||
## Compatibility Notes
|
||||
|
||||
- Redis prefix `openclaw:a2a:` — shared with npm gateway, **no migration needed**
|
||||
- Redis prefix `heretek:event:` — Heretek-only, does not conflict
|
||||
- ACP adapter uses bundled `@agentclientprotocol/sdk` from npm openclaw package
|
||||
- HMAC auth: nonce is signed with gateway token from `openclaw.json` → `gateway.auth.token`
|
||||
@@ -0,0 +1,523 @@
|
||||
/**
|
||||
* Heretek OpenClaw — ACP Adapter
|
||||
* ==============================================================================
|
||||
* Adapter for connecting to the npm OpenClaw gateway using the ACP (Agent Client
|
||||
* Protocol) SDK. Replaces the archived JS gateway's simple JSON protocol.
|
||||
*
|
||||
* Architecture:
|
||||
* Heretek Module --> ACP Adapter --> npm OpenClaw Gateway (port 18789)
|
||||
* |
|
||||
* v
|
||||
* @agentclientprotocol/sdk
|
||||
* (bundled in npm openclaw)
|
||||
*
|
||||
* Auth:
|
||||
* The npm gateway uses HMAC nonce-challenge auth on WebSocket connect.
|
||||
* Token: from openclaw.json → gateway.auth.token
|
||||
*
|
||||
* WebSocket:
|
||||
* URL: ws://localhost:18789/a2a
|
||||
* Protocol: ACP over WebSocket (JSON-RPC 2.0)
|
||||
*
|
||||
* Usage:
|
||||
* const adapter = await ACPAdapter.connect({ agentId: 'alpha', token: '...' });
|
||||
* adapter.send({ type: 'message', to: 'beta', content: 'Hello' });
|
||||
* adapter.on('message', (msg) => console.log(msg));
|
||||
* await adapter.close();
|
||||
* ==============================================================================
|
||||
*/
|
||||
|
||||
const WebSocket = require('ws');
|
||||
const crypto = require('crypto');
|
||||
const EventEmitter = require('events');
|
||||
|
||||
// ACP SDK path (bundled in npm openclaw)
|
||||
const ACP_SDK_PATH = '/usr/lib/node_modules/openclaw/node_modules/@agentclientprotocol/sdk/dist';
|
||||
let ACP = null;
|
||||
|
||||
// Load ACP SDK dynamically (avoids hard dependency for non-WS usage)
|
||||
function loadACP() {
|
||||
if (ACP === null) {
|
||||
try {
|
||||
ACP = require(ACP_SDK_PATH);
|
||||
} catch (err) {
|
||||
console.warn('[ACP Adapter] ACP SDK not available at', ACP_SDK_PATH, err.message);
|
||||
ACP = null;
|
||||
}
|
||||
}
|
||||
return ACP;
|
||||
}
|
||||
|
||||
// ==============================================================================
|
||||
// ACP Adapter Class
|
||||
// ==============================================================================
|
||||
|
||||
class ACPAdapter extends EventEmitter {
|
||||
/**
|
||||
* Connect to the npm OpenClaw gateway
|
||||
* @param {Object} options - Connection options
|
||||
* @param {string} options.agentId - Agent ID to register as
|
||||
* @param {string} options.token - Gateway auth token (from openclaw.json)
|
||||
* @param {string} [options.gatewayUrl='ws://localhost:18789/a2a'] - Gateway WebSocket URL
|
||||
* @param {number} [options.timeout=10000] - Connection timeout ms
|
||||
* @returns {Promise<ACPAdapter>} Connected adapter instance
|
||||
*/
|
||||
static async connect(options = {}) {
|
||||
const adapter = new ACPAdapter(options);
|
||||
await adapter._connect();
|
||||
return adapter;
|
||||
}
|
||||
|
||||
constructor(options = {}) {
|
||||
super();
|
||||
this.options = {
|
||||
gatewayUrl: process.env.OPENCLAW_GATEWAY_URL || 'ws://localhost:18789/a2a',
|
||||
token: options.token || process.env.OPENCLAW_GATEWAY_TOKEN || '',
|
||||
agentId: options.agentId || 'unknown',
|
||||
timeout: options.timeout || 10000,
|
||||
...options
|
||||
};
|
||||
|
||||
this.ws = null;
|
||||
this.connected = false;
|
||||
this.authenticated = false;
|
||||
this.clientId = null;
|
||||
this.pendingRequests = new Map(); // correlationId -> { resolve, reject, timeout }
|
||||
this.messageCounter = 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Internal connect flow
|
||||
* @private
|
||||
*/
|
||||
async _connect() {
|
||||
return new Promise((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
reject(new Error(`ACP connect timeout after ${this.options.timeout}ms`));
|
||||
}, this.options.timeout);
|
||||
|
||||
try {
|
||||
this.ws = new WebSocket(this.options.gatewayUrl, {
|
||||
headers: this.options.token
|
||||
? { 'Authorization': `Bearer ${this.options.token}` }
|
||||
: {}
|
||||
});
|
||||
|
||||
this.ws.on('open', () => {
|
||||
clearTimeout(timer);
|
||||
console.log(`[ACP Adapter] Connected to ${this.options.gatewayUrl}`);
|
||||
});
|
||||
|
||||
this.ws.on('message', (data) => {
|
||||
this._handleMessage(data.toString());
|
||||
});
|
||||
|
||||
this.ws.on('error', (err) => {
|
||||
clearTimeout(timer);
|
||||
console.error('[ACP Adapter] WS error:', err.message);
|
||||
reject(err);
|
||||
});
|
||||
|
||||
this.ws.on('close', (code, reason) => {
|
||||
this.connected = false;
|
||||
this.authenticated = false;
|
||||
console.log(`[ACP Adapter] Disconnected: code=${code} reason=${reason}`);
|
||||
this._rejectAllPending(`Connection closed: ${code}`);
|
||||
this.emit('close', { code, reason });
|
||||
});
|
||||
|
||||
// Resolve once authenticated
|
||||
this.once('authenticated', () => {
|
||||
clearTimeout(timer);
|
||||
this.connected = true;
|
||||
this.authenticated = true;
|
||||
resolve(this);
|
||||
});
|
||||
|
||||
this.once('auth_failed', (err) => {
|
||||
clearTimeout(timer);
|
||||
reject(new Error(`ACP auth failed: ${err}`));
|
||||
});
|
||||
|
||||
} catch (err) {
|
||||
clearTimeout(timer);
|
||||
reject(err);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle incoming WebSocket message
|
||||
* @private
|
||||
*/
|
||||
_handleMessage(raw) {
|
||||
let msg;
|
||||
try {
|
||||
msg = JSON.parse(raw);
|
||||
} catch {
|
||||
console.warn('[ACP Adapter] Failed to parse message:', raw.slice(0, 100));
|
||||
return;
|
||||
}
|
||||
|
||||
// ACP uses connect.challenge for auth on the npm gateway
|
||||
if (msg.type === 'event' && msg.event === 'connect.challenge') {
|
||||
this._handleChallenge(msg.payload);
|
||||
return;
|
||||
}
|
||||
|
||||
// Auth success
|
||||
if (msg.type === 'auth.success' || msg.auth === true) {
|
||||
this.clientId = msg.clientId || this.options.agentId;
|
||||
this.emit('authenticated', msg);
|
||||
return;
|
||||
}
|
||||
|
||||
// Auth failure
|
||||
if (msg.type === 'auth.failed' || msg.type === 'auth_failed' || msg.error?.code === 401) {
|
||||
this.emit('auth_failed', msg.error || msg);
|
||||
return;
|
||||
}
|
||||
|
||||
// Response to our request
|
||||
if (msg.correlationId || msg.id) {
|
||||
const cid = msg.correlationId || msg.id;
|
||||
if (this.pendingRequests.has(cid)) {
|
||||
const pending = this.pendingRequests.get(cid);
|
||||
clearTimeout(pending.timeout);
|
||||
this.pendingRequests.delete(cid);
|
||||
if (msg.error) {
|
||||
pending.reject(new Error(msg.error.message || msg.error));
|
||||
} else {
|
||||
pending.resolve(msg);
|
||||
}
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// Inbound message from another agent
|
||||
if (msg.type === 'message' || msg.type === 'a2a.message') {
|
||||
this.emit('message', msg);
|
||||
return;
|
||||
}
|
||||
|
||||
// Broadcast
|
||||
if (msg.type === 'broadcast') {
|
||||
this.emit('broadcast', msg);
|
||||
return;
|
||||
}
|
||||
|
||||
// Agent event
|
||||
if (msg.type === 'agent.registered' || msg.type === 'agent.unregistered') {
|
||||
this.emit('agent.event', msg);
|
||||
return;
|
||||
}
|
||||
|
||||
// Default: forward as raw event
|
||||
this.emit('raw', msg);
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle nonce challenge from gateway
|
||||
* @private
|
||||
*/
|
||||
async _handleChallenge(payload) {
|
||||
const { nonce, ts } = payload;
|
||||
|
||||
if (!nonce) {
|
||||
// Gateway doesn't require auth — skip HMAC
|
||||
this._sendRaw({ type: 'auth.skip' });
|
||||
return;
|
||||
}
|
||||
|
||||
// Sign nonce with HMAC-SHA256 using gateway token
|
||||
const token = this.options.token;
|
||||
if (!token) {
|
||||
this.emit('auth_failed', 'No token available for HMAC signing');
|
||||
return;
|
||||
}
|
||||
|
||||
const hmac = crypto.createHmac('sha256', token);
|
||||
hmac.update(nonce);
|
||||
const signature = hmac.digest('base64');
|
||||
|
||||
this._sendRaw({
|
||||
type: 'auth.response',
|
||||
nonce,
|
||||
signature,
|
||||
agentId: this.options.agentId,
|
||||
ts
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Send raw JSON to WebSocket
|
||||
* @private
|
||||
*/
|
||||
_sendRaw(obj) {
|
||||
if (!this.ws || this.ws.readyState !== WebSocket.OPEN) {
|
||||
throw new Error('WebSocket not connected');
|
||||
}
|
||||
this.ws.send(JSON.stringify(obj));
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a message and wait for response
|
||||
* @param {Object} message - Message to send
|
||||
* @param {number} [timeout=30000] - Response timeout ms
|
||||
* @returns {Promise<Object>} Response
|
||||
*/
|
||||
async send(message, timeout = 30000) {
|
||||
if (!this.authenticated) {
|
||||
throw new Error('Not authenticated');
|
||||
}
|
||||
|
||||
const correlationId = `acp_${Date.now()}_${++this.messageCounter}`;
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
this.pendingRequests.delete(correlationId);
|
||||
reject(new Error(`ACP request timeout: ${correlationId}`));
|
||||
}, timeout);
|
||||
|
||||
this.pendingRequests.set(correlationId, { resolve, reject, timeout: timer });
|
||||
|
||||
this._sendRaw({
|
||||
...message,
|
||||
correlationId,
|
||||
from: this.options.agentId,
|
||||
timestamp: new Date().toISOString()
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a message to a specific agent
|
||||
* @param {string} to - Target agent ID
|
||||
* @param {Object|string} content - Message content
|
||||
* @param {Object} [options] - Additional options
|
||||
* @returns {Promise<Object>} Send result
|
||||
*/
|
||||
async sendMessage(to, content, options = {}) {
|
||||
return this.send({
|
||||
type: 'message',
|
||||
to,
|
||||
content: typeof content === 'string' ? content : JSON.stringify(content),
|
||||
...options
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Broadcast to all connected agents
|
||||
* @param {Object|string} content - Broadcast content
|
||||
* @returns {Promise<Object>} Broadcast result
|
||||
*/
|
||||
async broadcast(content) {
|
||||
return this.send({
|
||||
type: 'broadcast',
|
||||
content: typeof content === 'string' ? content : JSON.stringify(content)
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Ping an agent
|
||||
* @param {string} [agentId] - Target agent (if omitted, ping gateway)
|
||||
* @returns {Promise<Object>} Ping result
|
||||
*/
|
||||
async ping(agentId) {
|
||||
return this.send({
|
||||
type: 'ping',
|
||||
to: agentId || undefined,
|
||||
timestamp: Date.now()
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Request gateway status
|
||||
* @returns {Promise<Object>} Gateway status
|
||||
*/
|
||||
async getStatus() {
|
||||
return this.send({ type: 'status' });
|
||||
}
|
||||
|
||||
/**
|
||||
* Discover connected agents
|
||||
* @returns {Promise<Object>} Agent list
|
||||
*/
|
||||
async discover() {
|
||||
return this.send({ type: 'discover' });
|
||||
}
|
||||
|
||||
/**
|
||||
* Register this agent with the gateway
|
||||
* @param {Object} [metadata] - Agent metadata
|
||||
* @returns {Promise<Object>} Registration result
|
||||
*/
|
||||
async register(metadata = {}) {
|
||||
return this.send({
|
||||
type: 'register',
|
||||
agentId: this.options.agentId,
|
||||
metadata: {
|
||||
role: metadata.role || 'unknown',
|
||||
...metadata
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Close the connection
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
async close() {
|
||||
this._rejectAllPending('Connection closed');
|
||||
if (this.ws) {
|
||||
this.ws.close(1000, 'Normal close');
|
||||
this.ws = null;
|
||||
}
|
||||
this.connected = false;
|
||||
this.authenticated = false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Reject all pending requests
|
||||
* @private
|
||||
*/
|
||||
_rejectAllPending(reason) {
|
||||
for (const [id, pending] of this.pendingRequests) {
|
||||
clearTimeout(pending.timeout);
|
||||
pending.reject(new Error(reason));
|
||||
}
|
||||
this.pendingRequests.clear();
|
||||
}
|
||||
|
||||
// ======================================================================
|
||||
// Static helpers (for modules that only need Redis, not WS)
|
||||
// ======================================================================
|
||||
|
||||
/**
|
||||
* Get Redis client (from a2a-message-send pattern)
|
||||
* Returns a connected ioredis client.
|
||||
* @param {string} [url] - Redis URL
|
||||
* @returns {Promise<Redis>} ioredis client
|
||||
*/
|
||||
static async getRedisClient(url) {
|
||||
const Redis = require('ioredis');
|
||||
const redisUrl = url || process.env.REDIS_URL || 'redis://localhost:6379';
|
||||
const client = new Redis(redisUrl, {
|
||||
maxRetriesPerRequest: 3,
|
||||
lazyConnect: true
|
||||
});
|
||||
await client.ping();
|
||||
return client;
|
||||
}
|
||||
}
|
||||
|
||||
// ==============================================================================
|
||||
// LiteLLM A2A Gateway Adapter (REST-based)
|
||||
//
|
||||
// Some Heretek skills target the LiteLLM proxy A2A gateway via REST.
|
||||
// This adapter provides a typed interface for that.
|
||||
// ==============================================================================
|
||||
|
||||
class LiteLLMGatewayAdapter {
|
||||
/**
|
||||
* Create LiteLLM gateway adapter
|
||||
* @param {Object} options
|
||||
* @param {string} [options.host='localhost'] - LiteLLM host
|
||||
* @param {string} [options.port='4000'] - LiteLLM port
|
||||
* @param {string} [options.masterKey] - LiteLLM master key
|
||||
*/
|
||||
constructor(options = {}) {
|
||||
this.host = options.host || process.env.LITELLM_HOST || 'localhost';
|
||||
this.port = options.port || process.env.LITELLM_PORT || '4000';
|
||||
this.masterKey = options.masterKey || process.env.LITELLM_MASTER_KEY || '';
|
||||
this.baseUrl = `http://${this.host}:${this.port}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Make authenticated request to LiteLLM proxy
|
||||
* @private
|
||||
*/
|
||||
async _request(method, path, body) {
|
||||
const fetch = require('http').request || require('fetch-cookie')(require('node-fetch'));
|
||||
const url = `${this.baseUrl}${path}`;
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
try {
|
||||
const urlObj = new URL(url);
|
||||
const opts = {
|
||||
hostname: urlObj.hostname,
|
||||
port: urlObj.port,
|
||||
path: urlObj.pathname,
|
||||
method,
|
||||
headers: {
|
||||
'Authorization': `Bearer ${this.masterKey}`,
|
||||
'Content-Type': 'application/json'
|
||||
}
|
||||
};
|
||||
|
||||
const req = require('http').request(opts, (res) => {
|
||||
let data = '';
|
||||
res.on('data', chunk => data += chunk);
|
||||
res.on('end', () => {
|
||||
if (res.statusCode >= 400) {
|
||||
reject(new Error(`LiteLLM ${res.statusCode}: ${data}`));
|
||||
} else {
|
||||
try {
|
||||
resolve(JSON.parse(data));
|
||||
} catch {
|
||||
resolve(data);
|
||||
}
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
req.on('error', reject);
|
||||
if (body) req.write(JSON.stringify(body));
|
||||
req.end();
|
||||
} catch (err) {
|
||||
reject(err);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Register an agent with LiteLLM proxy
|
||||
* @param {Object} agentInfo
|
||||
* @returns {Promise<Object>}
|
||||
*/
|
||||
async registerAgent(agentInfo) {
|
||||
return this._request('POST', '/key/generate', {
|
||||
key_alias: `a2a-${agentInfo.agentId}`,
|
||||
duration: '30d',
|
||||
agent: agentInfo.agentId,
|
||||
agent_permissions: ['a2a:send', 'a2a:receive'],
|
||||
...agentInfo.metadata
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* List agents
|
||||
* @returns {Promise<Array>}
|
||||
*/
|
||||
async listAgents() {
|
||||
return this._request('GET', '/agents');
|
||||
}
|
||||
|
||||
/**
|
||||
* Get agent info
|
||||
* @param {string} agentId
|
||||
* @returns {Promise<Object>}
|
||||
*/
|
||||
async getAgent(agentId) {
|
||||
return this._request('GET', `/agents/${agentId}`);
|
||||
}
|
||||
}
|
||||
|
||||
// ==============================================================================
|
||||
// Exports
|
||||
// ==============================================================================
|
||||
|
||||
module.exports = {
|
||||
ACPAdapter,
|
||||
LiteLLMGatewayAdapter
|
||||
};
|
||||
@@ -0,0 +1,217 @@
|
||||
/**
|
||||
* Heretek OpenClaw — BFT Consensus Adapter
|
||||
* ==============================================================================
|
||||
* Adapter that wraps the existing BFT consensus implementation and publishes
|
||||
* consensus results to the npm OpenClaw gateway via the ACP adapter.
|
||||
*
|
||||
* **STATUS:** No protocol change needed — BFT uses Redis pub/sub only.
|
||||
* This adapter adds npm gateway notification when consensus is reached.
|
||||
*
|
||||
* BFT Protocol phases: PRE-PREPARE → PREPARE → COMMIT → REPLY
|
||||
* Redis keys used (already in use, no migration needed):
|
||||
* - bft:consensus (pub/sub channel)
|
||||
* - bft:prepare:{view}:{seq}:{digest}
|
||||
* - bft:commit:{view}:{seq}:{digest}
|
||||
* - bft:view-change:{view}
|
||||
*
|
||||
* Usage:
|
||||
* const { BFTConsensusAdapter } = require('./bft-consensus-adapter');
|
||||
* const bft = new BFTConsensusAdapter({ nodeId: 'alpha' });
|
||||
* await bft.connect();
|
||||
* await bft.propose({ decision: 'approve_phase_3', reason: '...' });
|
||||
* // When committed, result is published to npm gateway automatically
|
||||
* ==============================================================================
|
||||
*/
|
||||
|
||||
const { BFTConsensus } = require('../consensus/bft-consensus');
|
||||
const { Redis } = require('ioredis');
|
||||
|
||||
class BFTConsensusAdapter {
|
||||
/**
|
||||
* Create BFT consensus adapter
|
||||
* @param {Object} options
|
||||
* @param {string} options.nodeId - This node's agent ID
|
||||
* @param {string} [options.redisUrl] - Redis URL
|
||||
* @param {string} [options.gatewayToken] - npm gateway token (for ACP notification)
|
||||
* @param {string} [options.gatewayUrl] - npm gateway WS URL
|
||||
* @param {number} [options.clusterSize=4] - Cluster size (3f+1, default 4 for f=1)
|
||||
*/
|
||||
constructor(options = {}) {
|
||||
this.nodeId = options.nodeId || 'unknown';
|
||||
this.redisUrl = options.redisUrl || process.env.REDIS_URL || 'redis://localhost:6379';
|
||||
this.gatewayToken = options.gatewayToken || process.env.OPENCLAW_GATEWAY_TOKEN;
|
||||
this.gatewayUrl = options.gatewayUrl || process.env.OPENCLAW_GATEWAY_WS || 'ws://localhost:18789/a2a';
|
||||
this.clusterSize = options.clusterSize || 4;
|
||||
|
||||
this.bft = null;
|
||||
this.redis = null;
|
||||
this.acpAdapter = null;
|
||||
this.connected = false;
|
||||
|
||||
// Callbacks for consensus events
|
||||
this.onCommitted = options.onCommitted || (() => {});
|
||||
this.onViewChange = options.onViewChange || (() => {});
|
||||
}
|
||||
|
||||
/**
|
||||
* Connect BFT and Redis
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
async connect() {
|
||||
// Connect Redis
|
||||
this.redis = new Redis(this.redisUrl, { maxRetriesPerRequest: 3 });
|
||||
await this.redis.ping();
|
||||
console.log(`[BFT Adapter] Connected to Redis`);
|
||||
|
||||
// Initialize BFT consensus
|
||||
this.bft = new BFTConsensus({
|
||||
redisUrl: this.redisUrl,
|
||||
nodeId: this.nodeId,
|
||||
clusterSize: this.clusterSize
|
||||
});
|
||||
|
||||
// Subscribe to BFT consensus channel
|
||||
await this.bft.subscribe();
|
||||
console.log(`[BFT Adapter] Subscribed to bft:consensus channel`);
|
||||
|
||||
// Override consensus handlers to add gateway notification
|
||||
this._patchBFTHandlers();
|
||||
|
||||
this.connected = true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Patch BFT handlers to notify npm gateway on consensus events
|
||||
* @private
|
||||
*/
|
||||
_patchBFTHandlers() {
|
||||
const originalCommit = this.bft.handleCommit.bind(this.bft);
|
||||
const self = this;
|
||||
|
||||
this.bft.handleCommit = async function(msg) {
|
||||
const result = await originalCommit(msg);
|
||||
if (result && result.committed) {
|
||||
await self._notifyGateway('consensus.committed', {
|
||||
view: result.view,
|
||||
sequence: result.sequence,
|
||||
nodeId: self.nodeId,
|
||||
quorum: result.commits
|
||||
});
|
||||
self.onCommitted(result);
|
||||
}
|
||||
return result;
|
||||
};
|
||||
|
||||
const originalViewChange = this.bft.handleViewChange.bind(this.bft);
|
||||
this.bft.handleViewChange = async function(msg) {
|
||||
const result = await originalViewChange(msg);
|
||||
if (result && result.accepted) {
|
||||
await self._notifyGateway('consensus.view_change', {
|
||||
view: result.view,
|
||||
nodeId: self.nodeId
|
||||
});
|
||||
self.onViewChange(result);
|
||||
}
|
||||
return result;
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Notify npm gateway of BFT event
|
||||
* @private
|
||||
*/
|
||||
async _notifyGateway(eventType, payload) {
|
||||
if (!this.acpAdapter || !this.acpAdapter.authenticated) {
|
||||
// Fall back to Redis broadcast (npm gateway reads this too)
|
||||
try {
|
||||
await this.redis.publish('openclaw:a2a:broadcast', JSON.stringify({
|
||||
type: 'bft.event',
|
||||
event: eventType,
|
||||
source: this.nodeId,
|
||||
payload,
|
||||
timestamp: new Date().toISOString()
|
||||
}));
|
||||
} catch (err) {
|
||||
console.warn(`[BFT Adapter] Redis broadcast failed:`, err.message);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await this.acpAdapter.broadcast({
|
||||
type: 'bft.event',
|
||||
event: eventType,
|
||||
source: this.nodeId,
|
||||
payload
|
||||
});
|
||||
} catch (err) {
|
||||
console.warn(`[BFT Adapter] ACP broadcast failed:`, err.message);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Connect ACP adapter for real-time gateway notification
|
||||
* @param {Object} options
|
||||
* @param {string} options.agentId - Agent ID to register as
|
||||
* @param {string} [options.token] - Gateway token override
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
async connectACP(options = {}) {
|
||||
try {
|
||||
const { ACPAdapter } = require('./acp-adapter.js');
|
||||
this.acpAdapter = await ACPAdapter.connect({
|
||||
agentId: options.agentId || this.nodeId,
|
||||
token: options.token || this.gatewayToken,
|
||||
gatewayUrl: this.gatewayUrl
|
||||
});
|
||||
console.log(`[BFT Adapter] ACP connected`);
|
||||
} catch (err) {
|
||||
console.warn(`[BFT Adapter] ACP connection failed (continuing without):`, err.message);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Propose a decision for BFT consensus
|
||||
* @param {Object} decision - Decision object
|
||||
* @param {string} decision.action - Action being decided (e.g. 'approve_phase_3')
|
||||
* @param {string} decision.reason - Reason for the decision
|
||||
* @param {Object} [decision.metadata] - Additional metadata
|
||||
* @returns {Promise<Object>} Consensus result
|
||||
*/
|
||||
async propose(decision) {
|
||||
if (!this.connected) throw new Error('BFT Adapter not connected');
|
||||
console.log(`[BFT Adapter] Proposing: ${decision.action}`);
|
||||
return this.bft.propose(decision);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get BFT status
|
||||
* @returns {Object}
|
||||
*/
|
||||
getStatus() {
|
||||
return {
|
||||
connected: this.connected,
|
||||
nodeId: this.nodeId,
|
||||
clusterSize: this.clusterSize,
|
||||
bft: this.bft ? this.bft.getStatus() : null,
|
||||
acpConnected: !!(this.acpAdapter && this.acpAdapter.authenticated)
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Disconnect
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
async disconnect() {
|
||||
if (this.acpAdapter) {
|
||||
await this.acpAdapter.close();
|
||||
}
|
||||
if (this.redis) {
|
||||
await this.redis.quit();
|
||||
}
|
||||
this.connected = false;
|
||||
console.log(`[BFT Adapter] Disconnected`);
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = { BFTConsensusAdapter };
|
||||
@@ -0,0 +1,205 @@
|
||||
/**
|
||||
* Heretek OpenClaw — LiteLLM Agent Registry Adapter
|
||||
* ==============================================================================
|
||||
* Adapter for registering agents with the LiteLLM proxy A2A gateway via REST API.
|
||||
*
|
||||
* This is separate from the npm OpenClaw gateway (port 18789).
|
||||
* LiteLLM proxy typically runs on port 4000.
|
||||
*
|
||||
* The LiteLLM proxy has its own A2A gateway built in, accessible via:
|
||||
* POST /key/generate — create agent API key
|
||||
* GET /agents — list agents
|
||||
* GET /key/info — key metadata
|
||||
*
|
||||
* Usage:
|
||||
* const registry = new LiteLLMAgentRegistry({
|
||||
* host: 'localhost',
|
||||
* port: 4000,
|
||||
* masterKey: process.env.LITELLM_MASTER_KEY
|
||||
* });
|
||||
*
|
||||
* await registry.registerAgent({
|
||||
* agentId: 'alpha',
|
||||
* role: 'triad',
|
||||
* skills: ['deliberate', 'vote']
|
||||
* });
|
||||
*
|
||||
* const agents = await registry.listAgents();
|
||||
* ==============================================================================
|
||||
*/
|
||||
|
||||
const http = require('http');
|
||||
|
||||
class LiteLLMAgentRegistry {
|
||||
/**
|
||||
* Create LiteLLM agent registry adapter
|
||||
* @param {Object} options
|
||||
* @param {string} [options.host='localhost'] - LiteLLM host
|
||||
* @param {number} [options.port=4000] - LiteLLM port
|
||||
* @param {string} [options.masterKey] - LiteLLM master key
|
||||
* @param {string} [options.baseUrl] - Override base URL
|
||||
*/
|
||||
constructor(options = {}) {
|
||||
this.host = options.host || process.env.LITELLM_HOST || 'localhost';
|
||||
this.port = options.port || parseInt(process.env.LITELLM_PORT || '4000', 10);
|
||||
this.masterKey = options.masterKey || process.env.LITELLM_MASTER_KEY || '';
|
||||
this.baseUrl = options.baseUrl || `http://${this.host}:${this.port}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Make HTTP request to LiteLLM proxy
|
||||
* @private
|
||||
*/
|
||||
_request(method, path, body = null) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const url = new URL(path, this.baseUrl);
|
||||
const opts = {
|
||||
hostname: url.hostname,
|
||||
port: url.port,
|
||||
path: url.pathname + url.search,
|
||||
method,
|
||||
headers: {
|
||||
'Authorization': `Bearer ${this.masterKey}`,
|
||||
'Content-Type': 'application/json'
|
||||
},
|
||||
timeout: 10000
|
||||
};
|
||||
|
||||
const req = http.request(opts, (res) => {
|
||||
let data = '';
|
||||
res.on('data', chunk => data += chunk);
|
||||
res.on('end', () => {
|
||||
if (res.statusCode >= 400) {
|
||||
reject(new Error(`LiteLLM ${res.statusCode}: ${data.slice(0, 200)}`));
|
||||
} else {
|
||||
try { resolve(JSON.parse(data)); }
|
||||
catch { resolve(data); }
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
req.on('error', reject);
|
||||
req.on('timeout', () => { req.destroy(); reject(new Error(`Request timeout: ${path}`)); });
|
||||
if (body) req.write(JSON.stringify(body));
|
||||
req.end();
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Register an agent with LiteLLM proxy
|
||||
* Creates an API key with a2a:send and a2a:receive permissions.
|
||||
*
|
||||
* @param {Object} agentInfo
|
||||
* @param {string} agentInfo.agentId - Agent ID
|
||||
* @param {string} [agentInfo.role] - Agent role
|
||||
* @param {Array<string>} [agentInfo.skills] - Agent skills
|
||||
* @param {Object} [agentInfo.metadata] - Additional metadata
|
||||
* @returns {Promise<Object>} Registration result
|
||||
*/
|
||||
async registerAgent(agentInfo) {
|
||||
const keyAlias = `a2a-${agentInfo.agentId}`;
|
||||
|
||||
const payload = {
|
||||
key_alias: keyAlias,
|
||||
duration: '30d',
|
||||
agent: agentInfo.agentId,
|
||||
agent_permissions: ['a2a:send', 'a2a:receive'],
|
||||
metadata: {
|
||||
role: agentInfo.role || 'unknown',
|
||||
skills: agentInfo.skills || [],
|
||||
registered_at: new Date().toISOString(),
|
||||
...agentInfo.metadata
|
||||
}
|
||||
};
|
||||
|
||||
try {
|
||||
const result = await this._request('POST', '/key/generate', payload);
|
||||
console.log(`[LiteLLM Registry] Registered agent: ${agentInfo.agentId}`);
|
||||
return {
|
||||
success: true,
|
||||
agentId: agentInfo.agentId,
|
||||
keyAlias,
|
||||
result
|
||||
};
|
||||
} catch (error) {
|
||||
console.error(`[LiteLLM Registry] Failed to register ${agentInfo.agentId}:`, error.message);
|
||||
return { success: false, agentId: agentInfo.agentId, error: error.message };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Register multiple agents
|
||||
* @param {Array<Object>} agents - Array of agent info objects
|
||||
* @returns {Promise<Array<Object>>} Results
|
||||
*/
|
||||
async registerAgents(agents) {
|
||||
return Promise.all(agents.map(agent => this.registerAgent(agent)));
|
||||
}
|
||||
|
||||
/**
|
||||
* Register all known collective agents
|
||||
* @returns {Promise<Array<Object>>} Results
|
||||
*/
|
||||
async registerAllAgents() {
|
||||
const { KNOWN_AGENTS } = await import('../skills/a2a-message-send/a2a-redis.js');
|
||||
const agents = KNOWN_AGENTS.map(id => ({ agentId: id }));
|
||||
return this.registerAgents(agents);
|
||||
}
|
||||
|
||||
/**
|
||||
* List all agents registered with LiteLLM proxy
|
||||
* @returns {Promise<Array>}
|
||||
*/
|
||||
async listAgents() {
|
||||
try {
|
||||
return await this._request('GET', '/agents');
|
||||
} catch (error) {
|
||||
console.error('[LiteLLM Registry] listAgents error:', error.message);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get info about a specific key/agent
|
||||
* @param {string} keyAlias - Key alias (e.g., 'a2a-alpha')
|
||||
* @returns {Promise<Object>}
|
||||
*/
|
||||
async getKeyInfo(keyAlias) {
|
||||
try {
|
||||
return await this._request('GET', `/key/info?key=${encodeURIComponent(keyAlias)}`);
|
||||
} catch (error) {
|
||||
return { error: error.message };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete an agent's key
|
||||
* @param {string} keyAlias - Key alias to delete
|
||||
* @returns {Promise<Object>}
|
||||
*/
|
||||
async deleteAgent(keyAlias) {
|
||||
try {
|
||||
// LiteLLM doesn't expose key deletion via standard API
|
||||
// This is a placeholder for documentation
|
||||
console.warn(`[LiteLLM Registry] Key deletion not supported via API for: ${keyAlias}`);
|
||||
return { success: false, error: 'Key deletion not supported by LiteLLM proxy API' };
|
||||
} catch (error) {
|
||||
return { success: false, error: error.message };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get registry status
|
||||
* @returns {Promise<Object>}
|
||||
*/
|
||||
async getStatus() {
|
||||
try {
|
||||
await this._request('GET', '/key/info');
|
||||
return { reachable: true, host: this.host, port: this.port };
|
||||
} catch {
|
||||
return { reachable: false, host: this.host, port: this.port, error: 'Health check failed' };
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = { LiteLLMAgentRegistry };
|
||||
@@ -1,52 +1,33 @@
|
||||
---
|
||||
name: a2a-message-send
|
||||
description: Send agent-to-agent (A2A) messages via Redis pub/sub with optional ACP WebSocket delivery. Use when agents need to communicate, broadcast to the triad, or check agent health. Compatible with the npm OpenClaw gateway.
|
||||
---
|
||||
|
||||
# A2A Message Send Skill
|
||||
|
||||
**Version:** 1.0.0
|
||||
**Type:** Communication
|
||||
**Backend:** Redis
|
||||
|
||||
## Overview
|
||||
|
||||
Provides agent-to-agent (A2A) communication capabilities for the OpenClaw collective via Redis pub/sub messaging. This skill enables agents to send messages, broadcast to groups, manage inboxes, and perform health checks.
|
||||
Send messages between collective agents via Redis queues (npm gateway compatible) with optional real-time delivery via ACP WebSocket.
|
||||
|
||||
## Architecture
|
||||
|
||||
```
|
||||
┌─────────────┐ Redis ┌─────────────┐
|
||||
│ Agent A │ ◄───────────► │ Redis │
|
||||
│ (steward) │ Pub/Sub │ :6379 │
|
||||
├─────────────┤ │ Lists + │
|
||||
│ Agent B │ ◄───────────► │ Sets │
|
||||
│ (alpha) │ Redis └─────────────┘
|
||||
└─────────────┘ Commands
|
||||
Agent --> a2a-redis.js --> Redis (openclaw:a2a:inbox:{agentId})
|
||||
|
|
||||
+--> ACP Adapter (optional) --> npm Gateway (port 18789)
|
||||
|
||||
npm Gateway reads from same Redis keys for agent delivery.
|
||||
```
|
||||
|
||||
### Redis Data Structures
|
||||
**Redis prefix:** `openclaw:a2a:` (shared with npm gateway — no migration needed)
|
||||
|
||||
- `openclaw:a2a:inbox:{agentId}` - List storing messages for each agent
|
||||
- `openclaw:a2a:agents` - Set of registered agent IDs
|
||||
- `openclaw:a2a:agent:{agentId}` - Hash with agent metadata
|
||||
- `openclaw:a2a:broadcast` - Pub/sub channel for broadcasts
|
||||
- `openclaw:a2a:read:{agentId}` - Set of read message IDs
|
||||
|
||||
## Installation
|
||||
|
||||
### Prerequisites
|
||||
|
||||
- Redis server running and accessible
|
||||
- Node.js 18+
|
||||
- `ioredis` package installed
|
||||
|
||||
### Setup
|
||||
## Configuration
|
||||
|
||||
```bash
|
||||
# Install dependency
|
||||
npm install ioredis
|
||||
# Redis (required)
|
||||
REDIS_URL=redis://localhost:6379
|
||||
|
||||
# Set environment variables
|
||||
export REDIS_URL=redis://localhost:6379
|
||||
# Or
|
||||
export REDIS_HOST=localhost
|
||||
export REDIS_PORT=6379
|
||||
# ACP WebSocket (optional — for real-time delivery)
|
||||
OPENCLAW_GATEWAY_TOKEN=9b54947854ee05186fb5363d0ea113685794d08d4ab45f80
|
||||
OPENCLAW_GATEWAY_WS=ws://localhost:18789/a2a
|
||||
```
|
||||
|
||||
## Usage
|
||||
@@ -54,50 +35,45 @@ export REDIS_PORT=6379
|
||||
### Basic Messaging
|
||||
|
||||
```javascript
|
||||
const { sendMessage, getMessages } = require('./skills/a2a-message-send/a2a-redis.js');
|
||||
const { sendMessage, getMessages, connectACP } = require('./a2a-redis.js');
|
||||
|
||||
// Send a message from steward to alpha
|
||||
const result = await sendMessage('steward', 'alpha', 'Hello Alpha!');
|
||||
console.log(result);
|
||||
// { success: true, messageId: 'msg_...', from: 'steward', to: 'alpha', ... }
|
||||
// Send a message (always works — Redis queue)
|
||||
const result = await sendMessage('steward', 'alpha', 'Triad meeting now');
|
||||
// → { success: true, messageId: 'msg_1743...', from: 'steward', to: 'alpha' }
|
||||
|
||||
// Get messages from alpha's inbox
|
||||
// Get inbox messages
|
||||
const messages = await getMessages('alpha', 10);
|
||||
console.log(messages);
|
||||
// [{ messageId, from, to, content, timestamp, ... }, ...]
|
||||
|
||||
// Mark as read
|
||||
await markAsRead('alpha', 'msg_1743...');
|
||||
```
|
||||
|
||||
### Broadcasting
|
||||
### Real-Time via ACP
|
||||
|
||||
```javascript
|
||||
const { broadcast, broadcastToTriad, broadcastToAgents } = require('./a2a-redis.js');
|
||||
// Connect to npm gateway for real-time delivery
|
||||
const acp = await connectACP({ agentId: 'alpha' });
|
||||
|
||||
// Broadcast to all agents
|
||||
const result = await broadcast('steward', 'Meeting in 5 minutes!');
|
||||
// Now messages go via WebSocket when agent is online
|
||||
const result = await sendMessage('steward', 'alpha', 'Urgent!', { via: 'acp' });
|
||||
// → { success: true, acpDelivered: true }
|
||||
|
||||
// Broadcast to triad only
|
||||
const triadResult = await broadcastToTriad('coordinator', 'Triad deliberation needed');
|
||||
// Listen for incoming messages
|
||||
acp.on('message', (msg) => {
|
||||
console.log(`${msg.from}: ${msg.content}`);
|
||||
});
|
||||
|
||||
// Broadcast to specific agents
|
||||
const customResult = await broadcastToAgents('steward', ['alpha', 'beta', 'coder'], 'Task update');
|
||||
// Disconnect
|
||||
await disconnectACP();
|
||||
```
|
||||
|
||||
### Inbox Management
|
||||
### Broadcast to Triad
|
||||
|
||||
```javascript
|
||||
const { getMessages, countMessages, clearMessages, markAsRead, getUnreadMessages } = require('./a2a-redis.js');
|
||||
const { broadcastToTriad } = require('./a2a-redis.js');
|
||||
|
||||
// Count messages
|
||||
const { count } = await countMessages('alpha');
|
||||
|
||||
// Get unread messages only
|
||||
const unread = await getUnreadMessages('alpha', 10);
|
||||
|
||||
// Mark message as read
|
||||
await markAsRead('alpha', 'msg_123');
|
||||
|
||||
// Clear all messages
|
||||
await clearMessages('alpha');
|
||||
await broadcastToTriad('steward', 'Phase 3 deliberation starting');
|
||||
// → { success: true, sentTo: ['alpha', 'beta', 'charlie'], count: 3 }
|
||||
```
|
||||
|
||||
### Health Checks
|
||||
@@ -105,178 +81,66 @@ await clearMessages('alpha');
|
||||
```javascript
|
||||
const { pingAgent, pingTriad } = require('./a2a-redis.js');
|
||||
|
||||
// Ping a single agent
|
||||
const pingResult = await pingAgent('steward', 'alpha');
|
||||
console.log(pingResult);
|
||||
// { success: true, response: 'pong', latency: 5, target: 'alpha', registered: true }
|
||||
// Ping single agent
|
||||
const result = await pingAgent('steward', 'alpha');
|
||||
// → { success: true, response: 'pong', latency: 2, target: 'alpha', registered: true }
|
||||
|
||||
// Ping all triad members
|
||||
const triadPing = await pingTriad('steward');
|
||||
console.log(triadPing.responses);
|
||||
// { alpha: {...}, beta: {...}, charlie: {...} }
|
||||
const triad = await pingTriad('steward');
|
||||
// → { success: true, responses: { alpha: {...}, beta: {...}, charlie: {...} } }
|
||||
```
|
||||
|
||||
### Agent Registration
|
||||
|
||||
```javascript
|
||||
const { registerAgent, unregisterAgent, getRegisteredAgents } = require('./a2a-redis.js');
|
||||
const { registerAgent, getRegisteredAgents, unregisterAgent } = require('./a2a-redis.js');
|
||||
|
||||
// Register an agent
|
||||
await registerAgent('steward', { role: 'orchestrator', capabilities: ['coordinate', 'delegate'] });
|
||||
// Register as active
|
||||
await registerAgent('alpha', { role: 'triad', skills: ['deliberate', 'vote'] });
|
||||
|
||||
// Get all registered agents
|
||||
// List all registered agents
|
||||
const agents = await getRegisteredAgents();
|
||||
console.log(agents);
|
||||
// ['steward', 'alpha', 'beta', ...]
|
||||
// → ['steward', 'alpha', 'beta', 'charlie', ...]
|
||||
|
||||
// Unregister an agent
|
||||
await unregisterAgent('steward');
|
||||
// Unregister
|
||||
await unregisterAgent('alpha');
|
||||
```
|
||||
|
||||
## API Reference
|
||||
## Redis Key Reference
|
||||
|
||||
### Core Functions
|
||||
|
||||
| Function | Description | Parameters | Returns |
|
||||
|----------|-------------|------------|---------|
|
||||
| `sendMessage(from, to, content, options)` | Send message to agent | `from`: sender ID, `to`: recipient ID, `content`: message, `options`: {priority, type, metadata} | `{success, messageId, from, to, timestamp}` |
|
||||
| `getMessages(agentId, limit)` | Get messages from inbox | `agentId`: recipient ID, `limit`: max messages | `Array<Message>` |
|
||||
| `getUnreadMessages(agentId, limit)` | Get unread messages | `agentId`: recipient ID, `limit`: max messages | `Array<Message>` |
|
||||
| `markAsRead(agentId, messageId)` | Mark message as read | `agentId`: owner ID, `messageId`: ID to mark | `{success, agentId, messageId}` |
|
||||
| `countMessages(agentId)` | Count inbox messages | `agentId`: owner ID | `{count, agentId}` |
|
||||
| `clearMessages(agentId)` | Clear all messages | `agentId`: owner ID | `{success, agentId}` |
|
||||
|
||||
### Broadcast Functions
|
||||
|
||||
| Function | Description | Parameters | Returns |
|
||||
|----------|-------------|------------|---------|
|
||||
| `broadcast(from, content)` | Broadcast to all agents | `from`: sender ID, `content`: message | `{success, from, count, timestamp}` |
|
||||
| `broadcastToAll(from, content)` | Alias for broadcast | Same as broadcast | Same as broadcast |
|
||||
| `broadcastToAgents(from, agents, content)` | Broadcast to specific agents | `from`: sender ID, `agents`: array, `content`: message | `{success, from, sentTo, count}` |
|
||||
| `broadcastToTriad(from, content)` | Broadcast to triad | `from`: sender ID, `content`: message | `{success, from, recipients, count}` |
|
||||
|
||||
### Health Check Functions
|
||||
|
||||
| Function | Description | Parameters | Returns |
|
||||
|----------|-------------|------------|---------|
|
||||
| `pingAgent(from, to)` | Ping an agent | `from`: sender ID, `to`: target ID | `{success, response, latency, target, registered}` |
|
||||
| `pingTriad(from)` | Ping all triad members | `from`: sender ID | `{success, from, responses, timestamp}` |
|
||||
|
||||
### Validation Functions
|
||||
|
||||
| Function | Description | Parameters | Returns |
|
||||
|----------|-------------|------------|---------|
|
||||
| `validateMessage(message)` | Validate message format | `message`: message object | `{valid, errors}` |
|
||||
| `validateAgentId(agentId)` | Validate agent ID format | `agentId`: ID to validate | `boolean` |
|
||||
|
||||
### Agent Registration Functions
|
||||
|
||||
| Function | Description | Parameters | Returns |
|
||||
|----------|-------------|------------|---------|
|
||||
| `registerAgent(agentId, metadata)` | Register an agent | `agentId`: ID, `metadata`: optional info | `{success, agentId, timestamp}` |
|
||||
| `unregisterAgent(agentId)` | Unregister an agent | `agentId`: ID to remove | `{success, agentId}` |
|
||||
| `getRegisteredAgents()` | Get all registered agents | None | `Array<string>` |
|
||||
|
||||
### Connection Management
|
||||
|
||||
| Function | Description | Parameters | Returns |
|
||||
|----------|-------------|------------|---------|
|
||||
| `getRedisClient()` | Get Redis client instance | None | `Redis` client |
|
||||
| `closeRedisClient()` | Close Redis connection | None | `Promise<void>` |
|
||||
| Key | Type | Purpose |
|
||||
|-----|------|---------|
|
||||
| `openclaw:a2a:inbox:{agentId}` | List | Message queue per agent |
|
||||
| `openclaw:a2a:agents` | Set | Registered agent IDs |
|
||||
| `openclaw:a2a:agent:{agentId}` | Hash | Agent metadata |
|
||||
| `openclaw:a2a:broadcast` | Pub/Sub | Real-time broadcast channel |
|
||||
| `openclaw:a2a:read:{agentId}` | Set | Read message IDs |
|
||||
|
||||
## Message Format
|
||||
|
||||
```javascript
|
||||
{
|
||||
messageId: 'msg_1712000000000_abc123', // Unique message ID
|
||||
from: 'steward', // Sender agent ID
|
||||
to: 'alpha', // Recipient agent ID
|
||||
content: 'Hello!', // Message content (string or JSON)
|
||||
timestamp: '2026-04-01T12:00:00.000Z', // ISO 8601 timestamp
|
||||
priority: 'normal', // 'low', 'normal', 'high', 'urgent'
|
||||
type: 'task', // 'task', 'query', 'response', 'broadcast'
|
||||
inReplyTo: 'msg_...', // Original message ID (for responses)
|
||||
metadata: {} // Custom metadata
|
||||
messageId: 'msg_1743...',
|
||||
from: 'steward',
|
||||
to: 'alpha',
|
||||
content: 'Triad meeting now',
|
||||
timestamp: '2026-04-03T01:00:00.000Z',
|
||||
priority: 'normal',
|
||||
type: 'task'
|
||||
}
|
||||
```
|
||||
|
||||
## Known Agents
|
||||
## Migration Notes
|
||||
|
||||
The following agents are pre-registered in the OpenClaw collective:
|
||||
This skill was originally written for the archived `gateway/openclaw-gateway.js` (simple JSON WebSocket protocol). It has been migrated:
|
||||
|
||||
- **steward** - Orchestrator
|
||||
- **alpha, beta, charlie** - Triad (deliberation)
|
||||
- **examiner** - Interrogator
|
||||
- **explorer** - Scout
|
||||
- **sentinel, sentinel-prime** - Guardian
|
||||
- **coder** - Artisan
|
||||
- **dreamer** - Visionary
|
||||
- **empath** - Diplomat
|
||||
- **historian** - Archivist
|
||||
- **arbiter** - Adjudicator
|
||||
- **catalyst** - Accelerator
|
||||
- **chronos** - Timekeeper
|
||||
- **coordinator** - Integrator
|
||||
- **echo** - Communicator
|
||||
- **habit-forge** - Optimizer
|
||||
- **metis** - Strategist
|
||||
- **nexus** - Connector
|
||||
- **perceiver** - Sensor
|
||||
- **prism** - Analyzer
|
||||
- ✅ Redis key space (`openclaw:a2a:*`) is identical to npm gateway — **no data migration needed**
|
||||
- ✅ ACP adapter added for real-time WebSocket delivery when connected to npm gateway
|
||||
- ✅ Fallback to Redis queue always works (npm gateway uses same keys)
|
||||
- ⚠️ If using ACP real-time delivery, both agents must be ACP-connected
|
||||
|
||||
## Error Handling
|
||||
|
||||
All functions return a result object with a `success` flag. When `success` is `false`, an `error` property contains the error message.
|
||||
|
||||
```javascript
|
||||
const result = await sendMessage('steward', 'invalid-agent', 'Test');
|
||||
if (!result.success) {
|
||||
console.error('Send failed:', result.error);
|
||||
// Handle error
|
||||
}
|
||||
```
|
||||
|
||||
### Common Errors
|
||||
|
||||
- `Invalid sender agent ID` - Sender ID doesn't match expected format
|
||||
- `Invalid recipient agent ID` - Recipient ID doesn't match expected format
|
||||
- `Redis connection failed` - Cannot connect to Redis server
|
||||
- `Invalid message` - Message validation failed
|
||||
|
||||
## Testing
|
||||
|
||||
Run the test suite:
|
||||
## Dependencies
|
||||
|
||||
```bash
|
||||
# Run A2A message send tests
|
||||
node --test tests/skills/a2a-message-send.test.js
|
||||
|
||||
# Run integration tests
|
||||
npm run test:integration
|
||||
npm install ioredis ws
|
||||
```
|
||||
|
||||
## Integration with Agent Client
|
||||
|
||||
The `agent-client.js` library automatically uses this skill for A2A communication when Redis is available:
|
||||
|
||||
```javascript
|
||||
const AgentClient = require('./agents/lib/agent-client');
|
||||
const client = new AgentClient({
|
||||
agentId: 'steward',
|
||||
role: 'orchestrator',
|
||||
gatewayUrl: 'ws://localhost:18789'
|
||||
});
|
||||
|
||||
// Send message (uses Redis backend)
|
||||
await client.sendMessage('alpha', { task: 'Analyze this' });
|
||||
|
||||
// Broadcast
|
||||
await client.broadcast('Attention all agents!');
|
||||
```
|
||||
|
||||
## See Also
|
||||
|
||||
- [`../modules/communication/redis-websocket-bridge.js`](../modules/communication/redis-websocket-bridge.js) - Redis to WebSocket bridge
|
||||
- [`../gateway/openclaw-gateway.js`](../gateway/openclaw-gateway.js) - OpenClaw Gateway server
|
||||
- [`../../agents/lib/agent-client.js`](../../agents/lib/agent-client.js) - Agent client library
|
||||
- [`../../tests/skills/a2a-message-send.test.js`](../../tests/skills/a2a-message-send.test.js) - Test suite
|
||||
|
||||
@@ -1,31 +1,48 @@
|
||||
/**
|
||||
* Heretek OpenClaw — A2A Message Send Skill (Redis-based)
|
||||
* Heretek OpenClaw — A2A Message Send Skill (Redis + ACP Dual-Mode)
|
||||
* ==============================================================================
|
||||
* Provides agent-to-agent communication via Redis pub/sub messaging.
|
||||
*
|
||||
* Provides agent-to-agent communication via Redis pub/sub (queue-and-deliver)
|
||||
* with optional real-time delivery via ACP WebSocket adapter.
|
||||
*
|
||||
* **MIGRATED from:** `gateway/openclaw-gateway.js` JS protocol
|
||||
* **MIGRATION NOTES:**
|
||||
* - Redis key space (`openclaw:a2a:*`) is shared with npm gateway — no migration needed
|
||||
* - Real-time delivery now routes via ACP adapter when agent is WS-connected
|
||||
* - Fallback: Redis queue (works offline, used by npm gateway internally)
|
||||
*
|
||||
* Features:
|
||||
* - Send messages between agents via Redis lists
|
||||
* - Send messages between agents via Redis lists (compatible with npm gateway)
|
||||
* - Real-time delivery via ACP WebSocket when connected
|
||||
* - Broadcast to specific agents or all agents
|
||||
* - Message persistence in Redis
|
||||
* - Message persistence in Redis (inbox pattern)
|
||||
* - Inbox management (get, count, clear messages)
|
||||
* - Ping/pong health checks
|
||||
* - Message validation
|
||||
* - Priority messaging support
|
||||
*
|
||||
* Redis Structure:
|
||||
*
|
||||
* Redis Structure (npm gateway compatible):
|
||||
* - openclaw:a2a:inbox:{agentId} - List of messages for agent
|
||||
* - openclaw:a2a:agents - Set of registered agents
|
||||
* - openclaw:a2a:broadcast - Pub/sub channel for broadcasts
|
||||
*
|
||||
* - openclaw:a2a:agent:{agentId} - Agent metadata (hash)
|
||||
*
|
||||
* ACP Mode:
|
||||
* When an agent is connected via ACP WebSocket (using acp-adapter.js),
|
||||
* messages are delivered in real-time. When offline, messages queue
|
||||
* in Redis and are delivered on next connect.
|
||||
*
|
||||
* Usage:
|
||||
* const { sendMessage, broadcast, getMessages } = require('./a2a-redis.js');
|
||||
*
|
||||
* // Send message
|
||||
* const { sendMessage, broadcast, getMessages, connectACP } = require('./a2a-redis.js');
|
||||
*
|
||||
* // Basic send (Redis queue, works with npm gateway)
|
||||
* const result = await sendMessage('steward', 'alpha', 'Hello Alpha!');
|
||||
*
|
||||
*
|
||||
* // Real-time send via ACP WebSocket
|
||||
* await connectACP({ agentId: 'alpha', token: '...' });
|
||||
* const result = await sendMessage('steward', 'alpha', 'Urgent!', { via: 'acp' });
|
||||
*
|
||||
* // Broadcast to triad
|
||||
* const result = await broadcastToTriad('steward', 'Triad meeting in 5');
|
||||
*
|
||||
*
|
||||
* // Get messages
|
||||
* const messages = await getMessages('alpha', 10);
|
||||
* ==============================================================================
|
||||
@@ -57,6 +74,9 @@ const TRIAD_AGENTS = ['alpha', 'beta', 'charlie'];
|
||||
// Redis client singleton
|
||||
let redisClient = null;
|
||||
|
||||
// ACP adapter singleton (optional — for real-time delivery)
|
||||
let acpAdapter = null;
|
||||
|
||||
// ==============================================================================
|
||||
// Redis Connection
|
||||
// ==============================================================================
|
||||
@@ -71,7 +91,6 @@ async function getRedisClient() {
|
||||
}
|
||||
|
||||
try {
|
||||
// Try REDIS_URL first, then fall back to HOST/PORT
|
||||
const url = process.env.REDIS_URL;
|
||||
if (url) {
|
||||
redisClient = new Redis(url, {
|
||||
@@ -89,10 +108,8 @@ async function getRedisClient() {
|
||||
});
|
||||
}
|
||||
|
||||
// Test connection
|
||||
await redisClient.ping();
|
||||
console.log('[A2A Redis] Connected to Redis');
|
||||
|
||||
return redisClient;
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] Failed to connect to Redis:', error.message);
|
||||
@@ -112,83 +129,113 @@ async function closeRedisClient() {
|
||||
}
|
||||
|
||||
// ==============================================================================
|
||||
// Message Utilities
|
||||
// ACP Connection (optional real-time layer)
|
||||
// ==============================================================================
|
||||
|
||||
/**
|
||||
* Generate unique message ID
|
||||
* @returns {string} Unique message ID
|
||||
* Connect to npm gateway via ACP WebSocket for real-time message delivery.
|
||||
* When connected, messages are sent directly via WebSocket instead of Redis queue.
|
||||
*
|
||||
* @param {Object} options - ACP connection options
|
||||
* @param {string} options.agentId - Agent ID to register as
|
||||
* @param {string} [options.token] - Gateway auth token (from openclaw.json)
|
||||
* @param {string} [options.gatewayUrl] - Gateway WebSocket URL
|
||||
* @returns {Promise<ACPAdapter>} ACP adapter instance
|
||||
*/
|
||||
async function connectACP(options = {}) {
|
||||
// Lazy-load adapter to avoid hard dependency when only Redis needed
|
||||
let ACPAdapter;
|
||||
try {
|
||||
ACPAdapter = require('../modules/adapters/acp-adapter.js').ACPAdapter;
|
||||
} catch {
|
||||
throw new Error(
|
||||
'[A2A Redis] connectACP requires modules/adapters/acp-adapter.js. ' +
|
||||
'Run: cp modules/adapters/acp-adapter.js /root/heretek/heretek-openclaw-core/modules/adapters/acp-adapter.js'
|
||||
);
|
||||
}
|
||||
|
||||
const token = options.token || process.env.OPENCLAW_GATEWAY_TOKEN;
|
||||
const gatewayUrl = options.gatewayUrl || process.env.OPENCLAW_GATEWAY_WS || 'ws://localhost:18789/a2a';
|
||||
|
||||
acpAdapter = await ACPAdapter.connect({
|
||||
agentId: options.agentId,
|
||||
token,
|
||||
gatewayUrl
|
||||
});
|
||||
|
||||
// Forward ACP messages to Redis inbox for persistence
|
||||
acpAdapter.on('message', async (msg) => {
|
||||
console.log(`[A2A ACP] Real-time message from ${msg.from}:`, String(msg.content).slice(0, 80));
|
||||
// Also store in Redis inbox for durability
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
const inboxKey = `${A2A_PREFIX}:inbox:${options.agentId}`;
|
||||
await client.lpush(inboxKey, JSON.stringify({
|
||||
messageId: msg.messageId || `acp_${Date.now()}`,
|
||||
from: msg.from,
|
||||
to: options.agentId,
|
||||
content: msg.content,
|
||||
timestamp: msg.timestamp || new Date().toISOString(),
|
||||
via: 'acp'
|
||||
}));
|
||||
} catch (err) {
|
||||
console.error('[A2A ACP] Failed to store ACP message in Redis:', err.message);
|
||||
}
|
||||
});
|
||||
|
||||
console.log(`[A2A ACP] Connected as ${options.agentId}`);
|
||||
return acpAdapter;
|
||||
}
|
||||
|
||||
/**
|
||||
* Disconnect ACP adapter
|
||||
*/
|
||||
async function disconnectACP() {
|
||||
if (acpAdapter) {
|
||||
await acpAdapter.close();
|
||||
acpAdapter = null;
|
||||
console.log('[A2A ACP] Disconnected');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if ACP is connected
|
||||
* @returns {boolean}
|
||||
*/
|
||||
function isACPConnected() {
|
||||
return acpAdapter !== null && acpAdapter.connected;
|
||||
}
|
||||
|
||||
// ==============================================================================
|
||||
// Message Utilities
|
||||
// ==============================================================================
|
||||
|
||||
function generateMessageId() {
|
||||
return `msg_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate agent ID format
|
||||
* @param {string} agentId - Agent ID to validate
|
||||
* @returns {boolean} True if valid
|
||||
*/
|
||||
function validateAgentId(agentId) {
|
||||
if (!agentId || typeof agentId !== 'string') {
|
||||
return false;
|
||||
}
|
||||
// Valid agent IDs: lowercase letters, numbers, hyphens
|
||||
if (!agentId || typeof agentId !== 'string') return false;
|
||||
return /^[a-z][a-z0-9-]*$/.test(agentId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate message format
|
||||
* @param {Object} message - Message to validate
|
||||
* @returns {Object} Validation result with valid flag and errors
|
||||
*/
|
||||
function validateMessage(message) {
|
||||
const errors = [];
|
||||
|
||||
if (!message) {
|
||||
errors.push('Message is required');
|
||||
return { valid: false, errors };
|
||||
}
|
||||
|
||||
if (!message.from || !validateAgentId(message.from)) {
|
||||
errors.push('Invalid or missing sender (from)');
|
||||
}
|
||||
|
||||
if (!message.to && !message.broadcast) {
|
||||
errors.push('Message must have recipient (to) or be a broadcast');
|
||||
}
|
||||
|
||||
if (message.to && !validateAgentId(message.to)) {
|
||||
errors.push('Invalid recipient (to)');
|
||||
}
|
||||
|
||||
if (message.content === undefined || message.content === null) {
|
||||
errors.push('Message content is required');
|
||||
}
|
||||
|
||||
return {
|
||||
valid: errors.length === 0,
|
||||
errors
|
||||
};
|
||||
if (!message) { errors.push('Message is required'); return { valid: false, errors }; }
|
||||
if (!message.from || !validateAgentId(message.from)) errors.push('Invalid or missing sender (from)');
|
||||
if (!message.to && !message.broadcast && !message.via) errors.push('Message must have recipient (to) or be a broadcast');
|
||||
if (message.to && !validateAgentId(message.to)) errors.push('Invalid recipient (to)');
|
||||
if (message.content === undefined || message.content === null) errors.push('Message content is required');
|
||||
return { valid: errors.length === 0, errors };
|
||||
}
|
||||
|
||||
/**
|
||||
* Create message object
|
||||
* @param {string} from - Sender agent ID
|
||||
* @param {string} to - Recipient agent ID
|
||||
* @param {string|Object} content - Message content
|
||||
* @param {Object} options - Additional options
|
||||
* @returns {Object} Message object
|
||||
*/
|
||||
function createMessage(from, to, content, options = {}) {
|
||||
const messageId = generateMessageId();
|
||||
const timestamp = new Date().toISOString();
|
||||
|
||||
return {
|
||||
messageId,
|
||||
messageId: generateMessageId(),
|
||||
from,
|
||||
to,
|
||||
content: typeof content === 'string' ? content : JSON.stringify(content),
|
||||
timestamp,
|
||||
timestamp: new Date().toISOString(),
|
||||
priority: options.priority || 'normal',
|
||||
type: options.type || 'task',
|
||||
inReplyTo: options.inReplyTo,
|
||||
@@ -201,224 +248,176 @@ function createMessage(from, to, content, options = {}) {
|
||||
// ==============================================================================
|
||||
|
||||
/**
|
||||
* Send a message to another agent via Redis
|
||||
* Send a message to another agent
|
||||
*
|
||||
* Delivery priority:
|
||||
* 1. If ACP connected and agent is online → real-time WebSocket
|
||||
* 2. If ACP connected but agent is offline → Redis queue
|
||||
* 3. Always: Redis queue (for npm gateway compatibility)
|
||||
*
|
||||
* @param {string} from - Sender agent ID
|
||||
* @param {string} to - Recipient agent ID
|
||||
* @param {string|Object} content - Message content
|
||||
* @param {Object} options - Additional options (priority, type, etc.)
|
||||
* @param {Object} [options]
|
||||
* @param {string} [options.via='redis'] - 'acp' for real-time, 'redis' for queue-only
|
||||
* @param {string} [options.priority] - 'high', 'normal', 'low'
|
||||
* @returns {Promise<Object>} Send result with messageId, success flag
|
||||
*/
|
||||
async function sendMessage(from, to, content, options = {}) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
// Validate inputs
|
||||
if (!validateAgentId(from)) {
|
||||
throw new Error(`Invalid sender agent ID: ${from}`);
|
||||
}
|
||||
if (!validateAgentId(to)) {
|
||||
throw new Error(`Invalid recipient agent ID: ${to}`);
|
||||
}
|
||||
|
||||
// Create message
|
||||
|
||||
if (!validateAgentId(from)) throw new Error(`Invalid sender agent ID: ${from}`);
|
||||
if (!validateAgentId(to)) throw new Error(`Invalid recipient agent ID: ${to}`);
|
||||
|
||||
const message = createMessage(from, to, content, options);
|
||||
|
||||
// Validate message
|
||||
const validation = validateMessage(message);
|
||||
if (!validation.valid) {
|
||||
throw new Error(`Invalid message: ${validation.errors.join(', ')}`);
|
||||
}
|
||||
|
||||
// Store message in recipient's inbox (Redis list)
|
||||
if (!validation.valid) throw new Error(`Invalid message: ${validation.errors.join(', ')}`);
|
||||
|
||||
// Always store in Redis inbox (npm gateway compatible)
|
||||
const inboxKey = `${A2A_PREFIX}:inbox:${to}`;
|
||||
await client.lpush(inboxKey, JSON.stringify(message));
|
||||
|
||||
// Register sender and recipient in agents set
|
||||
|
||||
// Register agents in set
|
||||
await client.sadd(`${A2A_PREFIX}:agents`, from, to);
|
||||
|
||||
// Publish to broadcast channel for real-time delivery
|
||||
|
||||
// Publish to broadcast channel
|
||||
await client.publish(`${A2A_PREFIX}:broadcast`, JSON.stringify({
|
||||
...message,
|
||||
action: 'message'
|
||||
}));
|
||||
|
||||
console.log(`[A2A Redis] Message sent from ${from} to ${to}: ${message.messageId}`);
|
||||
|
||||
|
||||
// Try real-time delivery via ACP if connected
|
||||
let acpDelivered = false;
|
||||
if (acpAdapter && acpAdapter.authenticated) {
|
||||
try {
|
||||
// Check if recipient is connected via ACP (have pub/sub knowledge)
|
||||
// For now, always also send via ACP when connected
|
||||
await acpAdapter.sendMessage(to, message.content, {
|
||||
messageId: message.messageId,
|
||||
priority: message.priority,
|
||||
type: message.type
|
||||
});
|
||||
acpDelivered = true;
|
||||
} catch (err) {
|
||||
console.warn(`[A2A] ACP real-time delivery to ${to} failed:`, err.message);
|
||||
}
|
||||
}
|
||||
|
||||
console.log(`[A2A] ${from} → ${to}: ${message.messageId}${acpDelivered ? ' [ACP real-time]' : ' [Redis queued]'}`);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
messageId: message.messageId,
|
||||
from: message.from,
|
||||
to: message.to,
|
||||
timestamp: message.timestamp,
|
||||
priority: message.priority
|
||||
priority: message.priority,
|
||||
acpDelivered
|
||||
};
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] sendMessage error:', error.message);
|
||||
return {
|
||||
success: false,
|
||||
error: error.message,
|
||||
from,
|
||||
to
|
||||
};
|
||||
console.error('[A2A] sendMessage error:', error.message);
|
||||
return { success: false, error: error.message, from, to };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get messages from agent's inbox
|
||||
* @param {string} agentId - Agent ID to get messages for
|
||||
* @param {number} limit - Maximum number of messages to return
|
||||
* @returns {Promise<Array>} Array of messages
|
||||
* @param {string} agentId - Agent ID
|
||||
* @param {number} [limit=10] - Max messages
|
||||
* @returns {Promise<Array>}
|
||||
*/
|
||||
async function getMessages(agentId, limit = 10) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(agentId)) {
|
||||
throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
}
|
||||
|
||||
if (!validateAgentId(agentId)) throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
const inboxKey = `${A2A_PREFIX}:inbox:${agentId}`;
|
||||
const messages = await client.lrange(inboxKey, 0, limit - 1);
|
||||
|
||||
return messages.map(msg => {
|
||||
try {
|
||||
return JSON.parse(msg);
|
||||
} catch (e) {
|
||||
return { raw: msg, parseError: e.message };
|
||||
}
|
||||
try { return JSON.parse(msg); }
|
||||
catch { return { raw: msg, parseError: true }; }
|
||||
});
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] getMessages error:', error.message);
|
||||
console.error('[A2A] getMessages error:', error.message);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get unread messages (messages not yet marked as read)
|
||||
* @param {string} agentId - Agent ID to get messages for
|
||||
* @param {number} limit - Maximum number of messages to return
|
||||
* @returns {Promise<Array>} Array of unread messages
|
||||
* Get unread messages
|
||||
* @param {string} agentId - Agent ID
|
||||
* @param {number} [limit=10] - Max messages
|
||||
* @returns {Promise<Array>}
|
||||
*/
|
||||
async function getUnreadMessages(agentId, limit = 10) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(agentId)) {
|
||||
throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
}
|
||||
|
||||
if (!validateAgentId(agentId)) throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
const inboxKey = `${A2A_PREFIX}:inbox:${agentId}`;
|
||||
const readSetKey = `${A2A_PREFIX}:read:${agentId}`;
|
||||
|
||||
const messages = await client.lrange(inboxKey, 0, limit - 1);
|
||||
const readIds = await client.smembers(readSetKey);
|
||||
const readSet = new Set(readIds);
|
||||
|
||||
return messages
|
||||
.map(msg => {
|
||||
try {
|
||||
return JSON.parse(msg);
|
||||
} catch (e) {
|
||||
return null;
|
||||
}
|
||||
})
|
||||
.map(msg => { try { return JSON.parse(msg); } catch { return null; } })
|
||||
.filter(msg => msg && !readSet.has(msg.messageId));
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] getUnreadMessages error:', error.message);
|
||||
console.error('[A2A] getUnreadMessages error:', error.message);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark a message as read
|
||||
* Mark message as read
|
||||
* @param {string} agentId - Agent ID
|
||||
* @param {string} messageId - Message ID to mark as read
|
||||
* @returns {Promise<Object>} Result with success flag
|
||||
* @param {string} messageId - Message ID
|
||||
* @returns {Promise<Object>}
|
||||
*/
|
||||
async function markAsRead(agentId, messageId) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(agentId)) {
|
||||
throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
}
|
||||
|
||||
if (!validateAgentId(agentId)) throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
const readSetKey = `${A2A_PREFIX}:read:${agentId}`;
|
||||
await client.sadd(readSetKey, messageId);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
agentId,
|
||||
messageId
|
||||
};
|
||||
return { success: true, agentId, messageId };
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] markAsRead error:', error.message);
|
||||
return {
|
||||
success: false,
|
||||
error: error.message
|
||||
};
|
||||
return { success: false, error: error.message };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Count messages in agent's inbox
|
||||
* @param {string} agentId - Agent ID to count messages for
|
||||
* @returns {Promise<Object>} Result with count
|
||||
* Count inbox messages
|
||||
* @param {string} agentId - Agent ID
|
||||
* @returns {Promise<Object>}
|
||||
*/
|
||||
async function countMessages(agentId) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(agentId)) {
|
||||
throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
}
|
||||
|
||||
if (!validateAgentId(agentId)) throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
const inboxKey = `${A2A_PREFIX}:inbox:${agentId}`;
|
||||
const count = await client.llen(inboxKey);
|
||||
|
||||
return {
|
||||
count,
|
||||
agentId
|
||||
};
|
||||
return { count, agentId };
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] countMessages error:', error.message);
|
||||
return {
|
||||
count: 0,
|
||||
agentId,
|
||||
error: error.message
|
||||
};
|
||||
return { count: 0, agentId, error: error.message };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear all messages from agent's inbox
|
||||
* @param {string} agentId - Agent ID to clear inbox for
|
||||
* @returns {Promise<Object>} Result with success flag
|
||||
* Clear inbox
|
||||
* @param {string} agentId - Agent ID
|
||||
* @returns {Promise<Object>}
|
||||
*/
|
||||
async function clearMessages(agentId) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(agentId)) {
|
||||
throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
}
|
||||
|
||||
if (!validateAgentId(agentId)) throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
const inboxKey = `${A2A_PREFIX}:inbox:${agentId}`;
|
||||
const readSetKey = `${A2A_PREFIX}:read:${agentId}`;
|
||||
|
||||
await client.del(inboxKey);
|
||||
await client.del(readSetKey);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
agentId
|
||||
};
|
||||
return { success: true, agentId };
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] clearMessages error:', error.message);
|
||||
return {
|
||||
success: false,
|
||||
error: error.message
|
||||
};
|
||||
return { success: false, error: error.message };
|
||||
}
|
||||
}
|
||||
|
||||
@@ -427,152 +426,96 @@ async function clearMessages(agentId) {
|
||||
// ==============================================================================
|
||||
|
||||
/**
|
||||
* Broadcast message to specific agents
|
||||
* @param {string} from - Sender agent ID
|
||||
* @param {Array<string>} agents - Array of recipient agent IDs
|
||||
* @param {string|Object} content - Message content
|
||||
* @returns {Promise<Object>} Result with sentTo array
|
||||
* Broadcast to array of agents
|
||||
*/
|
||||
async function broadcastToAgents(from, agents, content) {
|
||||
try {
|
||||
const results = await Promise.all(
|
||||
agents.map(agent => sendMessage(from, agent, content))
|
||||
);
|
||||
|
||||
const sentTo = results
|
||||
.filter(r => r.success)
|
||||
.map(r => r.to);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
from,
|
||||
sentTo,
|
||||
count: sentTo.length,
|
||||
timestamp: new Date().toISOString()
|
||||
};
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] broadcastToAgents error:', error.message);
|
||||
return {
|
||||
success: false,
|
||||
error: error.message,
|
||||
from
|
||||
};
|
||||
}
|
||||
const results = await Promise.all(agents.map(agent => sendMessage(from, agent, content)));
|
||||
const sentTo = results.filter(r => r.success).map(r => r.to);
|
||||
return { success: true, from, sentTo, count: sentTo.length, timestamp: new Date().toISOString() };
|
||||
}
|
||||
|
||||
/**
|
||||
* Broadcast message to triad members (alpha, beta, charlie)
|
||||
* @param {string} from - Sender agent ID
|
||||
* @param {string|Object} content - Message content
|
||||
* @returns {Promise<Object>} Result with recipients array
|
||||
* Broadcast to triad (alpha, beta, charlie)
|
||||
*/
|
||||
async function broadcastToTriad(from, content) {
|
||||
return broadcastToAgents(from, TRIAD_AGENTS, content);
|
||||
}
|
||||
|
||||
/**
|
||||
* Broadcast message to all known agents
|
||||
* @param {string} from - Sender agent ID
|
||||
* @param {string|Object} content - Message content
|
||||
* @returns {Promise<Object>} Result with count
|
||||
* Broadcast to all known agents
|
||||
*/
|
||||
async function broadcastToAll(from, content) {
|
||||
return broadcastToAgents(from, KNOWN_AGENTS, content);
|
||||
}
|
||||
|
||||
/**
|
||||
* Broadcast message (alias for broadcastToAll)
|
||||
* @param {string} from - Sender agent ID
|
||||
* @param {string|Object} content - Message content
|
||||
* @returns {Promise<Object>} Result with count
|
||||
* Broadcast (alias for broadcastToAll)
|
||||
*/
|
||||
async function broadcast(from, content) {
|
||||
return broadcastToAll(from, content);
|
||||
}
|
||||
|
||||
// ==============================================================================
|
||||
// Ping/Health Check Functions
|
||||
// Health Check Functions
|
||||
// ==============================================================================
|
||||
|
||||
/**
|
||||
* Ping another agent (health check)
|
||||
* @param {string} from - Sender agent ID
|
||||
* @param {string} to - Target agent ID
|
||||
* @returns {Promise<Object>} Ping result with response and latency
|
||||
* Ping an agent (via Redis queue — always works, real-time if ACP connected)
|
||||
*/
|
||||
async function pingAgent(from, to) {
|
||||
const startTime = Date.now();
|
||||
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(from)) {
|
||||
throw new Error(`Invalid sender agent ID: ${from}`);
|
||||
}
|
||||
if (!validateAgentId(to)) {
|
||||
throw new Error(`Invalid target agent ID: ${to}`);
|
||||
}
|
||||
|
||||
// Send ping message
|
||||
if (!validateAgentId(from)) throw new Error(`Invalid sender: ${from}`);
|
||||
if (!validateAgentId(to)) throw new Error(`Invalid target: ${to}`);
|
||||
|
||||
const pingMessage = {
|
||||
messageId: generateMessageId(),
|
||||
from,
|
||||
to,
|
||||
type: 'ping',
|
||||
content: 'ping',
|
||||
timestamp: new Date().toISOString()
|
||||
};
|
||||
|
||||
|
||||
const inboxKey = `${A2A_PREFIX}:inbox:${to}`;
|
||||
await client.lpush(inboxKey, JSON.stringify({
|
||||
messageId: generateMessageId(),
|
||||
from,
|
||||
to,
|
||||
...pingMessage
|
||||
}));
|
||||
|
||||
await client.lpush(inboxKey, JSON.stringify(pingMessage));
|
||||
|
||||
const latency = Date.now() - startTime;
|
||||
|
||||
// Check if target agent is registered
|
||||
const isRegistered = await client.sismember(`${A2A_PREFIX}:agents`, to);
|
||||
|
||||
|
||||
// Try ACP ping if connected
|
||||
let acpPing = null;
|
||||
if (acpAdapter && acpAdapter.authenticated) {
|
||||
try {
|
||||
const pong = await acpAdapter.ping(to);
|
||||
acpPing = { success: true, latency: Date.now() - startTime };
|
||||
} catch {
|
||||
acpPing = { success: false };
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
success: true,
|
||||
response: 'pong',
|
||||
latency,
|
||||
target: to,
|
||||
registered: isRegistered === 1
|
||||
registered: isRegistered === 1,
|
||||
acpPing
|
||||
};
|
||||
} catch (error) {
|
||||
const latency = Date.now() - startTime;
|
||||
console.error('[A2A Redis] pingAgent error:', error.message);
|
||||
return {
|
||||
success: false,
|
||||
error: error.message,
|
||||
latency,
|
||||
target: to
|
||||
};
|
||||
return { success: false, error: error.message, latency: Date.now() - startTime, target: to };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Ping all triad members
|
||||
* @param {string} from - Sender agent ID
|
||||
* @returns {Promise<Object>} Ping results for each triad member
|
||||
* Ping triad members
|
||||
*/
|
||||
async function pingTriad(from) {
|
||||
const results = await Promise.all(
|
||||
TRIAD_AGENTS.map(agent => pingAgent(from, agent))
|
||||
);
|
||||
|
||||
const results = await Promise.all(TRIAD_AGENTS.map(agent => pingAgent(from, agent)));
|
||||
const responses = {};
|
||||
TRIAD_AGENTS.forEach((agent, index) => {
|
||||
responses[agent] = results[index];
|
||||
});
|
||||
|
||||
return {
|
||||
success: true,
|
||||
from,
|
||||
responses,
|
||||
timestamp: new Date().toISOString()
|
||||
};
|
||||
TRIAD_AGENTS.forEach((agent, i) => { responses[agent] = results[i]; });
|
||||
return { success: true, from, responses, timestamp: new Date().toISOString() };
|
||||
}
|
||||
|
||||
// ==============================================================================
|
||||
@@ -580,23 +523,13 @@ async function pingTriad(from) {
|
||||
// ==============================================================================
|
||||
|
||||
/**
|
||||
* Register an agent in the A2A system
|
||||
* @param {string} agentId - Agent ID to register
|
||||
* @param {Object} metadata - Optional agent metadata
|
||||
* @returns {Promise<Object>} Registration result
|
||||
* Register agent in A2A system
|
||||
*/
|
||||
async function registerAgent(agentId, metadata = {}) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(agentId)) {
|
||||
throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
}
|
||||
|
||||
// Add to agents set
|
||||
if (!validateAgentId(agentId)) throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
await client.sadd(`${A2A_PREFIX}:agents`, agentId);
|
||||
|
||||
// Store agent metadata
|
||||
const agentKey = `${A2A_PREFIX}:agent:${agentId}`;
|
||||
await client.hset(agentKey, {
|
||||
id: agentId,
|
||||
@@ -605,73 +538,58 @@ async function registerAgent(agentId, metadata = {}) {
|
||||
status: 'active',
|
||||
...metadata
|
||||
});
|
||||
|
||||
console.log(`[A2A Redis] Agent registered: ${agentId}`);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
agentId,
|
||||
timestamp: new Date().toISOString()
|
||||
};
|
||||
console.log(`[A2A] Agent registered: ${agentId}`);
|
||||
return { success: true, agentId, timestamp: new Date().toISOString() };
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] registerAgent error:', error.message);
|
||||
return {
|
||||
success: false,
|
||||
error: error.message
|
||||
};
|
||||
return { success: false, error: error.message };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get list of registered agents
|
||||
* @returns {Promise<Array>} Array of agent IDs
|
||||
* Get registered agents
|
||||
*/
|
||||
async function getRegisteredAgents() {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
const agents = await client.smembers(`${A2A_PREFIX}:agents`);
|
||||
return agents;
|
||||
return await client.smembers(`${A2A_PREFIX}:agents`);
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] getRegisteredAgents error:', error.message);
|
||||
console.error('[A2A] getRegisteredAgents error:', error.message);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Unregister an agent
|
||||
* @param {string} agentId - Agent ID to unregister
|
||||
* @returns {Promise<Object>} Unregistration result
|
||||
* Unregister agent
|
||||
*/
|
||||
async function unregisterAgent(agentId) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(agentId)) {
|
||||
throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
}
|
||||
|
||||
// Remove from agents set
|
||||
if (!validateAgentId(agentId)) throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
await client.srem(`${A2A_PREFIX}:agents`, agentId);
|
||||
|
||||
// Remove agent metadata
|
||||
const agentKey = `${A2A_PREFIX}:agent:${agentId}`;
|
||||
await client.del(agentKey);
|
||||
|
||||
console.log(`[A2A Redis] Agent unregistered: ${agentId}`);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
agentId
|
||||
};
|
||||
await client.del(`${A2A_PREFIX}:agent:${agentId}`);
|
||||
console.log(`[A2A] Agent unregistered: ${agentId}`);
|
||||
return { success: true, agentId };
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] unregisterAgent error:', error.message);
|
||||
return {
|
||||
success: false,
|
||||
error: error.message
|
||||
};
|
||||
return { success: false, error: error.message };
|
||||
}
|
||||
}
|
||||
|
||||
// ==============================================================================
|
||||
// ACP Status
|
||||
// ==============================================================================
|
||||
|
||||
/**
|
||||
* Get ACP connection status
|
||||
*/
|
||||
function getACPStatus() {
|
||||
if (!acpAdapter) return { connected: false };
|
||||
return {
|
||||
connected: acpAdapter.connected,
|
||||
authenticated: acpAdapter.authenticated,
|
||||
clientId: acpAdapter.clientId
|
||||
};
|
||||
}
|
||||
|
||||
// ==============================================================================
|
||||
// Exports
|
||||
// ==============================================================================
|
||||
@@ -684,34 +602,36 @@ module.exports = {
|
||||
markAsRead,
|
||||
countMessages,
|
||||
clearMessages,
|
||||
|
||||
|
||||
// Broadcast
|
||||
broadcast,
|
||||
broadcastToAll,
|
||||
broadcastToAgents,
|
||||
broadcastToTriad,
|
||||
|
||||
|
||||
// Health checks
|
||||
pingAgent,
|
||||
pingTriad,
|
||||
|
||||
|
||||
// Validation
|
||||
validateMessage,
|
||||
validateAgentId,
|
||||
|
||||
|
||||
// Agent registration
|
||||
registerAgent,
|
||||
unregisterAgent,
|
||||
getRegisteredAgents,
|
||||
|
||||
// Utilities
|
||||
createMessage,
|
||||
generateMessageId,
|
||||
|
||||
|
||||
// ACP real-time layer (optional)
|
||||
connectACP,
|
||||
disconnectACP,
|
||||
isACPConnected,
|
||||
getACPStatus,
|
||||
|
||||
// Connection management
|
||||
getRedisClient,
|
||||
closeRedisClient,
|
||||
|
||||
|
||||
// Constants
|
||||
KNOWN_AGENTS,
|
||||
TRIAD_AGENTS,
|
||||
|
||||
@@ -0,0 +1,719 @@
|
||||
/**
|
||||
* Heretek OpenClaw — A2A Message Send Skill (Redis-based)
|
||||
* ==============================================================================
|
||||
* Provides agent-to-agent communication via Redis pub/sub messaging.
|
||||
*
|
||||
* Features:
|
||||
* - Send messages between agents via Redis lists
|
||||
* - Broadcast to specific agents or all agents
|
||||
* - Message persistence in Redis
|
||||
* - Inbox management (get, count, clear messages)
|
||||
* - Ping/pong health checks
|
||||
* - Message validation
|
||||
* - Priority messaging support
|
||||
*
|
||||
* Redis Structure:
|
||||
* - openclaw:a2a:inbox:{agentId} - List of messages for agent
|
||||
* - openclaw:a2a:agents - Set of registered agents
|
||||
* - openclaw:a2a:broadcast - Pub/sub channel for broadcasts
|
||||
*
|
||||
* Usage:
|
||||
* const { sendMessage, broadcast, getMessages } = require('./a2a-redis.js');
|
||||
*
|
||||
* // Send message
|
||||
* const result = await sendMessage('steward', 'alpha', 'Hello Alpha!');
|
||||
*
|
||||
* // Broadcast to triad
|
||||
* const result = await broadcastToTriad('steward', 'Triad meeting in 5');
|
||||
*
|
||||
* // Get messages
|
||||
* const messages = await getMessages('alpha', 10);
|
||||
* ==============================================================================
|
||||
*/
|
||||
|
||||
const Redis = require('ioredis');
|
||||
|
||||
// ==============================================================================
|
||||
// Configuration
|
||||
// ==============================================================================
|
||||
const REDIS_URL = process.env.REDIS_URL || 'redis://localhost:6379';
|
||||
const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
|
||||
const REDIS_PORT = process.env.REDIS_PORT || 6379;
|
||||
const A2A_PREFIX = 'openclaw:a2a';
|
||||
|
||||
// Known agents in the OpenClaw collective
|
||||
const KNOWN_AGENTS = [
|
||||
'steward', 'alpha', 'beta', 'charlie',
|
||||
'examiner', 'explorer', 'sentinel', 'coder',
|
||||
'dreamer', 'empath', 'historian', 'arbiter',
|
||||
'catalyst', 'chronos', 'coordinator', 'echo',
|
||||
'habit-forge', 'metis', 'nexus', 'perceiver',
|
||||
'prism', 'sentinel-prime'
|
||||
];
|
||||
|
||||
// Triad members for deliberation
|
||||
const TRIAD_AGENTS = ['alpha', 'beta', 'charlie'];
|
||||
|
||||
// Redis client singleton
|
||||
let redisClient = null;
|
||||
|
||||
// ==============================================================================
|
||||
// Redis Connection
|
||||
// ==============================================================================
|
||||
|
||||
/**
|
||||
* Get or create Redis client
|
||||
* @returns {Promise<Redis>} Redis client instance
|
||||
*/
|
||||
async function getRedisClient() {
|
||||
if (redisClient) {
|
||||
return redisClient;
|
||||
}
|
||||
|
||||
try {
|
||||
// Try REDIS_URL first, then fall back to HOST/PORT
|
||||
const url = process.env.REDIS_URL;
|
||||
if (url) {
|
||||
redisClient = new Redis(url, {
|
||||
maxRetriesPerRequest: 3,
|
||||
retryDelayOnFailover: 100,
|
||||
lazyConnect: true
|
||||
});
|
||||
} else {
|
||||
redisClient = new Redis({
|
||||
host: REDIS_HOST,
|
||||
port: REDIS_PORT,
|
||||
maxRetriesPerRequest: 3,
|
||||
retryDelayOnFailover: 100,
|
||||
lazyConnect: true
|
||||
});
|
||||
}
|
||||
|
||||
// Test connection
|
||||
await redisClient.ping();
|
||||
console.log('[A2A Redis] Connected to Redis');
|
||||
|
||||
return redisClient;
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] Failed to connect to Redis:', error.message);
|
||||
throw new Error(`Redis connection failed: ${error.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Close Redis client connection
|
||||
*/
|
||||
async function closeRedisClient() {
|
||||
if (redisClient) {
|
||||
await redisClient.quit();
|
||||
redisClient = null;
|
||||
console.log('[A2A Redis] Redis connection closed');
|
||||
}
|
||||
}
|
||||
|
||||
// ==============================================================================
|
||||
// Message Utilities
|
||||
// ==============================================================================
|
||||
|
||||
/**
|
||||
* Generate unique message ID
|
||||
* @returns {string} Unique message ID
|
||||
*/
|
||||
function generateMessageId() {
|
||||
return `msg_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate agent ID format
|
||||
* @param {string} agentId - Agent ID to validate
|
||||
* @returns {boolean} True if valid
|
||||
*/
|
||||
function validateAgentId(agentId) {
|
||||
if (!agentId || typeof agentId !== 'string') {
|
||||
return false;
|
||||
}
|
||||
// Valid agent IDs: lowercase letters, numbers, hyphens
|
||||
return /^[a-z][a-z0-9-]*$/.test(agentId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate message format
|
||||
* @param {Object} message - Message to validate
|
||||
* @returns {Object} Validation result with valid flag and errors
|
||||
*/
|
||||
function validateMessage(message) {
|
||||
const errors = [];
|
||||
|
||||
if (!message) {
|
||||
errors.push('Message is required');
|
||||
return { valid: false, errors };
|
||||
}
|
||||
|
||||
if (!message.from || !validateAgentId(message.from)) {
|
||||
errors.push('Invalid or missing sender (from)');
|
||||
}
|
||||
|
||||
if (!message.to && !message.broadcast) {
|
||||
errors.push('Message must have recipient (to) or be a broadcast');
|
||||
}
|
||||
|
||||
if (message.to && !validateAgentId(message.to)) {
|
||||
errors.push('Invalid recipient (to)');
|
||||
}
|
||||
|
||||
if (message.content === undefined || message.content === null) {
|
||||
errors.push('Message content is required');
|
||||
}
|
||||
|
||||
return {
|
||||
valid: errors.length === 0,
|
||||
errors
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Create message object
|
||||
* @param {string} from - Sender agent ID
|
||||
* @param {string} to - Recipient agent ID
|
||||
* @param {string|Object} content - Message content
|
||||
* @param {Object} options - Additional options
|
||||
* @returns {Object} Message object
|
||||
*/
|
||||
function createMessage(from, to, content, options = {}) {
|
||||
const messageId = generateMessageId();
|
||||
const timestamp = new Date().toISOString();
|
||||
|
||||
return {
|
||||
messageId,
|
||||
from,
|
||||
to,
|
||||
content: typeof content === 'string' ? content : JSON.stringify(content),
|
||||
timestamp,
|
||||
priority: options.priority || 'normal',
|
||||
type: options.type || 'task',
|
||||
inReplyTo: options.inReplyTo,
|
||||
metadata: options.metadata || {}
|
||||
};
|
||||
}
|
||||
|
||||
// ==============================================================================
|
||||
// Core Messaging Functions
|
||||
// ==============================================================================
|
||||
|
||||
/**
|
||||
* Send a message to another agent via Redis
|
||||
* @param {string} from - Sender agent ID
|
||||
* @param {string} to - Recipient agent ID
|
||||
* @param {string|Object} content - Message content
|
||||
* @param {Object} options - Additional options (priority, type, etc.)
|
||||
* @returns {Promise<Object>} Send result with messageId, success flag
|
||||
*/
|
||||
async function sendMessage(from, to, content, options = {}) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
// Validate inputs
|
||||
if (!validateAgentId(from)) {
|
||||
throw new Error(`Invalid sender agent ID: ${from}`);
|
||||
}
|
||||
if (!validateAgentId(to)) {
|
||||
throw new Error(`Invalid recipient agent ID: ${to}`);
|
||||
}
|
||||
|
||||
// Create message
|
||||
const message = createMessage(from, to, content, options);
|
||||
|
||||
// Validate message
|
||||
const validation = validateMessage(message);
|
||||
if (!validation.valid) {
|
||||
throw new Error(`Invalid message: ${validation.errors.join(', ')}`);
|
||||
}
|
||||
|
||||
// Store message in recipient's inbox (Redis list)
|
||||
const inboxKey = `${A2A_PREFIX}:inbox:${to}`;
|
||||
await client.lpush(inboxKey, JSON.stringify(message));
|
||||
|
||||
// Register sender and recipient in agents set
|
||||
await client.sadd(`${A2A_PREFIX}:agents`, from, to);
|
||||
|
||||
// Publish to broadcast channel for real-time delivery
|
||||
await client.publish(`${A2A_PREFIX}:broadcast`, JSON.stringify({
|
||||
...message,
|
||||
action: 'message'
|
||||
}));
|
||||
|
||||
console.log(`[A2A Redis] Message sent from ${from} to ${to}: ${message.messageId}`);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
messageId: message.messageId,
|
||||
from: message.from,
|
||||
to: message.to,
|
||||
timestamp: message.timestamp,
|
||||
priority: message.priority
|
||||
};
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] sendMessage error:', error.message);
|
||||
return {
|
||||
success: false,
|
||||
error: error.message,
|
||||
from,
|
||||
to
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get messages from agent's inbox
|
||||
* @param {string} agentId - Agent ID to get messages for
|
||||
* @param {number} limit - Maximum number of messages to return
|
||||
* @returns {Promise<Array>} Array of messages
|
||||
*/
|
||||
async function getMessages(agentId, limit = 10) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(agentId)) {
|
||||
throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
}
|
||||
|
||||
const inboxKey = `${A2A_PREFIX}:inbox:${agentId}`;
|
||||
const messages = await client.lrange(inboxKey, 0, limit - 1);
|
||||
|
||||
return messages.map(msg => {
|
||||
try {
|
||||
return JSON.parse(msg);
|
||||
} catch (e) {
|
||||
return { raw: msg, parseError: e.message };
|
||||
}
|
||||
});
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] getMessages error:', error.message);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get unread messages (messages not yet marked as read)
|
||||
* @param {string} agentId - Agent ID to get messages for
|
||||
* @param {number} limit - Maximum number of messages to return
|
||||
* @returns {Promise<Array>} Array of unread messages
|
||||
*/
|
||||
async function getUnreadMessages(agentId, limit = 10) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(agentId)) {
|
||||
throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
}
|
||||
|
||||
const inboxKey = `${A2A_PREFIX}:inbox:${agentId}`;
|
||||
const readSetKey = `${A2A_PREFIX}:read:${agentId}`;
|
||||
|
||||
const messages = await client.lrange(inboxKey, 0, limit - 1);
|
||||
const readIds = await client.smembers(readSetKey);
|
||||
const readSet = new Set(readIds);
|
||||
|
||||
return messages
|
||||
.map(msg => {
|
||||
try {
|
||||
return JSON.parse(msg);
|
||||
} catch (e) {
|
||||
return null;
|
||||
}
|
||||
})
|
||||
.filter(msg => msg && !readSet.has(msg.messageId));
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] getUnreadMessages error:', error.message);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark a message as read
|
||||
* @param {string} agentId - Agent ID
|
||||
* @param {string} messageId - Message ID to mark as read
|
||||
* @returns {Promise<Object>} Result with success flag
|
||||
*/
|
||||
async function markAsRead(agentId, messageId) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(agentId)) {
|
||||
throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
}
|
||||
|
||||
const readSetKey = `${A2A_PREFIX}:read:${agentId}`;
|
||||
await client.sadd(readSetKey, messageId);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
agentId,
|
||||
messageId
|
||||
};
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] markAsRead error:', error.message);
|
||||
return {
|
||||
success: false,
|
||||
error: error.message
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Count messages in agent's inbox
|
||||
* @param {string} agentId - Agent ID to count messages for
|
||||
* @returns {Promise<Object>} Result with count
|
||||
*/
|
||||
async function countMessages(agentId) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(agentId)) {
|
||||
throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
}
|
||||
|
||||
const inboxKey = `${A2A_PREFIX}:inbox:${agentId}`;
|
||||
const count = await client.llen(inboxKey);
|
||||
|
||||
return {
|
||||
count,
|
||||
agentId
|
||||
};
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] countMessages error:', error.message);
|
||||
return {
|
||||
count: 0,
|
||||
agentId,
|
||||
error: error.message
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear all messages from agent's inbox
|
||||
* @param {string} agentId - Agent ID to clear inbox for
|
||||
* @returns {Promise<Object>} Result with success flag
|
||||
*/
|
||||
async function clearMessages(agentId) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(agentId)) {
|
||||
throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
}
|
||||
|
||||
const inboxKey = `${A2A_PREFIX}:inbox:${agentId}`;
|
||||
const readSetKey = `${A2A_PREFIX}:read:${agentId}`;
|
||||
|
||||
await client.del(inboxKey);
|
||||
await client.del(readSetKey);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
agentId
|
||||
};
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] clearMessages error:', error.message);
|
||||
return {
|
||||
success: false,
|
||||
error: error.message
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
// ==============================================================================
|
||||
// Broadcast Functions
|
||||
// ==============================================================================
|
||||
|
||||
/**
|
||||
* Broadcast message to specific agents
|
||||
* @param {string} from - Sender agent ID
|
||||
* @param {Array<string>} agents - Array of recipient agent IDs
|
||||
* @param {string|Object} content - Message content
|
||||
* @returns {Promise<Object>} Result with sentTo array
|
||||
*/
|
||||
async function broadcastToAgents(from, agents, content) {
|
||||
try {
|
||||
const results = await Promise.all(
|
||||
agents.map(agent => sendMessage(from, agent, content))
|
||||
);
|
||||
|
||||
const sentTo = results
|
||||
.filter(r => r.success)
|
||||
.map(r => r.to);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
from,
|
||||
sentTo,
|
||||
count: sentTo.length,
|
||||
timestamp: new Date().toISOString()
|
||||
};
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] broadcastToAgents error:', error.message);
|
||||
return {
|
||||
success: false,
|
||||
error: error.message,
|
||||
from
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Broadcast message to triad members (alpha, beta, charlie)
|
||||
* @param {string} from - Sender agent ID
|
||||
* @param {string|Object} content - Message content
|
||||
* @returns {Promise<Object>} Result with recipients array
|
||||
*/
|
||||
async function broadcastToTriad(from, content) {
|
||||
return broadcastToAgents(from, TRIAD_AGENTS, content);
|
||||
}
|
||||
|
||||
/**
|
||||
* Broadcast message to all known agents
|
||||
* @param {string} from - Sender agent ID
|
||||
* @param {string|Object} content - Message content
|
||||
* @returns {Promise<Object>} Result with count
|
||||
*/
|
||||
async function broadcastToAll(from, content) {
|
||||
return broadcastToAgents(from, KNOWN_AGENTS, content);
|
||||
}
|
||||
|
||||
/**
|
||||
* Broadcast message (alias for broadcastToAll)
|
||||
* @param {string} from - Sender agent ID
|
||||
* @param {string|Object} content - Message content
|
||||
* @returns {Promise<Object>} Result with count
|
||||
*/
|
||||
async function broadcast(from, content) {
|
||||
return broadcastToAll(from, content);
|
||||
}
|
||||
|
||||
// ==============================================================================
|
||||
// Ping/Health Check Functions
|
||||
// ==============================================================================
|
||||
|
||||
/**
|
||||
* Ping another agent (health check)
|
||||
* @param {string} from - Sender agent ID
|
||||
* @param {string} to - Target agent ID
|
||||
* @returns {Promise<Object>} Ping result with response and latency
|
||||
*/
|
||||
async function pingAgent(from, to) {
|
||||
const startTime = Date.now();
|
||||
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(from)) {
|
||||
throw new Error(`Invalid sender agent ID: ${from}`);
|
||||
}
|
||||
if (!validateAgentId(to)) {
|
||||
throw new Error(`Invalid target agent ID: ${to}`);
|
||||
}
|
||||
|
||||
// Send ping message
|
||||
const pingMessage = {
|
||||
type: 'ping',
|
||||
content: 'ping',
|
||||
timestamp: new Date().toISOString()
|
||||
};
|
||||
|
||||
const inboxKey = `${A2A_PREFIX}:inbox:${to}`;
|
||||
await client.lpush(inboxKey, JSON.stringify({
|
||||
messageId: generateMessageId(),
|
||||
from,
|
||||
to,
|
||||
...pingMessage
|
||||
}));
|
||||
|
||||
const latency = Date.now() - startTime;
|
||||
|
||||
// Check if target agent is registered
|
||||
const isRegistered = await client.sismember(`${A2A_PREFIX}:agents`, to);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
response: 'pong',
|
||||
latency,
|
||||
target: to,
|
||||
registered: isRegistered === 1
|
||||
};
|
||||
} catch (error) {
|
||||
const latency = Date.now() - startTime;
|
||||
console.error('[A2A Redis] pingAgent error:', error.message);
|
||||
return {
|
||||
success: false,
|
||||
error: error.message,
|
||||
latency,
|
||||
target: to
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Ping all triad members
|
||||
* @param {string} from - Sender agent ID
|
||||
* @returns {Promise<Object>} Ping results for each triad member
|
||||
*/
|
||||
async function pingTriad(from) {
|
||||
const results = await Promise.all(
|
||||
TRIAD_AGENTS.map(agent => pingAgent(from, agent))
|
||||
);
|
||||
|
||||
const responses = {};
|
||||
TRIAD_AGENTS.forEach((agent, index) => {
|
||||
responses[agent] = results[index];
|
||||
});
|
||||
|
||||
return {
|
||||
success: true,
|
||||
from,
|
||||
responses,
|
||||
timestamp: new Date().toISOString()
|
||||
};
|
||||
}
|
||||
|
||||
// ==============================================================================
|
||||
// Agent Registration
|
||||
// ==============================================================================
|
||||
|
||||
/**
|
||||
* Register an agent in the A2A system
|
||||
* @param {string} agentId - Agent ID to register
|
||||
* @param {Object} metadata - Optional agent metadata
|
||||
* @returns {Promise<Object>} Registration result
|
||||
*/
|
||||
async function registerAgent(agentId, metadata = {}) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(agentId)) {
|
||||
throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
}
|
||||
|
||||
// Add to agents set
|
||||
await client.sadd(`${A2A_PREFIX}:agents`, agentId);
|
||||
|
||||
// Store agent metadata
|
||||
const agentKey = `${A2A_PREFIX}:agent:${agentId}`;
|
||||
await client.hset(agentKey, {
|
||||
id: agentId,
|
||||
registeredAt: new Date().toISOString(),
|
||||
lastSeen: new Date().toISOString(),
|
||||
status: 'active',
|
||||
...metadata
|
||||
});
|
||||
|
||||
console.log(`[A2A Redis] Agent registered: ${agentId}`);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
agentId,
|
||||
timestamp: new Date().toISOString()
|
||||
};
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] registerAgent error:', error.message);
|
||||
return {
|
||||
success: false,
|
||||
error: error.message
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get list of registered agents
|
||||
* @returns {Promise<Array>} Array of agent IDs
|
||||
*/
|
||||
async function getRegisteredAgents() {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
const agents = await client.smembers(`${A2A_PREFIX}:agents`);
|
||||
return agents;
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] getRegisteredAgents error:', error.message);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Unregister an agent
|
||||
* @param {string} agentId - Agent ID to unregister
|
||||
* @returns {Promise<Object>} Unregistration result
|
||||
*/
|
||||
async function unregisterAgent(agentId) {
|
||||
try {
|
||||
const client = await getRedisClient();
|
||||
|
||||
if (!validateAgentId(agentId)) {
|
||||
throw new Error(`Invalid agent ID: ${agentId}`);
|
||||
}
|
||||
|
||||
// Remove from agents set
|
||||
await client.srem(`${A2A_PREFIX}:agents`, agentId);
|
||||
|
||||
// Remove agent metadata
|
||||
const agentKey = `${A2A_PREFIX}:agent:${agentId}`;
|
||||
await client.del(agentKey);
|
||||
|
||||
console.log(`[A2A Redis] Agent unregistered: ${agentId}`);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
agentId
|
||||
};
|
||||
} catch (error) {
|
||||
console.error('[A2A Redis] unregisterAgent error:', error.message);
|
||||
return {
|
||||
success: false,
|
||||
error: error.message
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
// ==============================================================================
|
||||
// Exports
|
||||
// ==============================================================================
|
||||
|
||||
module.exports = {
|
||||
// Core messaging
|
||||
sendMessage,
|
||||
getMessages,
|
||||
getUnreadMessages,
|
||||
markAsRead,
|
||||
countMessages,
|
||||
clearMessages,
|
||||
|
||||
// Broadcast
|
||||
broadcast,
|
||||
broadcastToAll,
|
||||
broadcastToAgents,
|
||||
broadcastToTriad,
|
||||
|
||||
// Health checks
|
||||
pingAgent,
|
||||
pingTriad,
|
||||
|
||||
// Validation
|
||||
validateMessage,
|
||||
validateAgentId,
|
||||
|
||||
// Agent registration
|
||||
registerAgent,
|
||||
unregisterAgent,
|
||||
getRegisteredAgents,
|
||||
|
||||
// Utilities
|
||||
createMessage,
|
||||
generateMessageId,
|
||||
|
||||
// Connection management
|
||||
getRedisClient,
|
||||
closeRedisClient,
|
||||
|
||||
// Constants
|
||||
KNOWN_AGENTS,
|
||||
TRIAD_AGENTS,
|
||||
A2A_PREFIX
|
||||
};
|
||||
Reference in New Issue
Block a user