diff --git a/src/bridge-openclaw/index.js b/src/bridge-openclaw/index.js new file mode 100644 index 0000000..91ef32a --- /dev/null +++ b/src/bridge-openclaw/index.js @@ -0,0 +1,742 @@ +// Spatial Office — OpenClaw Gateway Bridge +// +// Connects to the OpenClaw Gateway WebSocket (ws://127.0.0.1:18789) +// as a backend client, subscribes to agent events, and emits +// state diffs that the spatial-office state engine consumes. +// +// Architecture: +// bridge → WS → OpenClaw GW → agent events/presence/sessions +// bridge → EventEmitter → state engine → WS server → browser +// +// Protocol: OpenClaw Gateway Protocol v3 (WS + JSON) +// Docs: /gateway/protocol + +import { EventEmitter } from "node:events"; +import { WebSocket } from "ws"; +import { z } from "zod"; +import { + AgentStatus, + SpatialAgent, + AgentConfig, + OfficeEvent, + DEFAULT_POSITIONS, + DEFAULT_ZONES, + getZonePosition, +} from "./types.js"; + +// ── Constants ────────────────────────────────────────────── +const GW_URL = + process.env.OPENCLAW_GATEWAY_URL || "ws://127.0.0.1:18789"; +const GW_TOKEN = + process.env.OPENCLAW_GATEWAY_TOKEN || + process.env.OPENCLAW_SHARED_TOKEN || + "b2faf38189449f8712d87832ab9e464f66b8ad22b78131f0"; +const RECONNECT_BASE_MS = 1_000; +const RECONNECT_MAX_MS = 30_000; +const AGENT_POLL_INTERVAL_MS = 10_000; // poll session state every 10s +const AGENT_IDLE_THRESHOLD_MS = 5 * 60_000; // 5 min → idle +const PING_INTERVAL_MS = 25_000; + +// ── Zone Assignment Logic ────────────────────────────────── + +/** + * Map agent activity context → spatial zone. + * + * Rules (from the spec): + * - Agent messaging another agent → collaboration pod + * - Agent running a long task → solo work area + * - Agent processing coach feedback → coaching corner + * - Agent idle → coffee bar / social area + * - Multiple agents messaging each other → same collaboration pod + * - Default: zone based on department affinity + */ +function assignZone(agentId, department, activityState, context = "") { + // Offline → stays wherever it was (we don't move offline agents) + if (activityState === "offline") { + return null; // caller should keep current zone + } + + // Idle → coffee bar + if (activityState === "idle") { + return "coffee-bar"; + } + + // Blocked → coaching corner (coach feedback / stuck) + if (activityState === "blocked") { + return "coaching-corner"; + } + + // Active: determine zone from context + const ctx = context.toLowerCase(); + + // Inter-agent messaging → collaboration pods + if ( + ctx.includes("message") || + ctx.includes("sent") || + ctx.includes("dispatch") || + ctx.includes("collaborat") || + ctx.includes("review") || + ctx.includes("pair") + ) { + // Engineering dept agents → Pod A, content/market/presentations → Pod B + const engDepts = ["engineering", "senior-engineering", "trust"]; + if (engDepts.includes(department)) { + return "pod-a"; + } + return "pod-b"; + } + + // Coach feedback, learning, memory maintenance → coaching corner + if ( + ctx.includes("coach") || + ctx.includes("memory") || + ctx.includes("learning") || + ctx.includes("self-improve") || + ctx.includes("feedback") || + ctx.includes("heartbeat") + ) { + return "coaching-corner"; + } + + // Long tasks, building, coding, writing → solo work + if ( + ctx.includes("build") || + ctx.includes("code") || + ctx.includes("write") || + ctx.includes("implement") || + ctx.includes("analyze") || + ctx.includes("research") + ) { + return departmentAffinityZone(department); + } + + // Department affinity default + return departmentAffinityZone(department); +} + +function departmentAffinityZone(department) { + const soloWest = ["cos", "engineering", "senior-engineering", "trust"]; + const soloEast = ["market", "health", "content", "presentations", "vault"]; + + if (soloWest.includes(department)) return "solo-west"; + if (soloEast.includes(department)) return "solo-east"; + return "solo-west"; // fallback +} + +// ── Agent Registry (from OpenClaw config) ───────────────── + +/** + * Parse agent entries from OpenClaw's agents.list config. + * Returns SpatialAgentConfig[] suitable for initializing the bridge. + */ +export function buildAgentRegistry(agentEntries) { + const registry = {}; + for (const entry of agentEntries) { + const id = entry.id || entry.name; + if (!id) continue; + + // Extract identity name + const displayName = entry.identity?.name || entry.name || id; + + // Infer department from id + const department = inferDepartment(id); + + const defaultPos = DEFAULT_POSITIONS[id] || { x: 0, y: 0 }; + + registry[id] = { + id, + name: displayName, + department, + state: "offline", // starts offline until we see activity + currentZone: departmentAffinityZone(department), + currentTask: null, + lastActivity: 0, + position: { ...defaultPos }, + }; + } + return registry; +} + +function inferDepartment(agentId) { + // Map known agent IDs to departments + const known = { + main: "cos", + cos: "cos", + engineering: "engineering", + "senior-engineering": "senior-engineering", + trust: "trust", + market: "market", + health: "health", + content: "content", + presentations: "presentations", + vault: "vault", + }; + return known[agentId] || "cos"; +} + +// ── Gateway WebSocket Client ─────────────────────────────── + +let requestId = 0; +function nextReqId() { + return `spatial-office-${++requestId}`; +} + +export class OpenClawBridge extends EventEmitter { + /** + * @param {object} opts + * @param {string} [opts.gatewayUrl] — WS URL for the OpenClaw Gateway + * @param {string} [opts.gatewayToken] — shared gateway auth token + * @param {Array} [opts.agentEntries] — agent config entries from openclaw.json + */ + constructor(opts = {}) { + super(); + this.gatewayUrl = opts.gatewayUrl || GW_URL; + this.gatewayToken = opts.gatewayToken || GW_TOKEN; + this.agentEntries = opts.agentEntries || []; + + // Internal state + this.ws = null; + this.connected = false; + this.pendingRequests = new Map(); + this.reconnectTimer = null; + this.pingTimer = null; + this.pollTimer = null; + this.backoffMs = RECONNECT_BASE_MS; + this.dying = false; + + // Bridge state + this.agents = buildAgentRegistry(this.agentEntries); + this.zones = JSON.parse(JSON.stringify(DEFAULT_ZONES)); + this.events = []; + this.maxEvents = 500; + + // Agent activity tracking: last seen activity per agent + this.lastAgentActivity = new Map(); // agentId → timestamp + this.agentSessions = new Map(); // agentId → { sessionKey, context } + } + + // ── Lifecycle ────────────────────────────────────────── + + /** Start the bridge: connect to Gateway and begin polling */ + start() { + this.dying = false; + this._connect(); + } + + /** Graceful shutdown */ + async stop() { + this.dying = true; + this._clearTimers(); + if (this.ws) { + this.ws.close(1000, "spatial-office shutting down"); + this.ws = null; + } + this.connected = false; + } + + // ── Connection Management ────────────────────────────── + + _connect() { + if (this.dying) return; + if (this.ws) { + try { this.ws.close(); } catch (_) { /* ignore */ } + this.ws = null; + } + + this.emit("status", { state: "connecting", url: this.gatewayUrl }); + + const ws = new WebSocket(this.gatewayUrl); + this.ws = ws; + + ws.on("open", () => this._onOpen()); + ws.on("message", (data) => this._onMessage(data)); + ws.on("close", (code, reason) => this._onClose(code, reason)); + ws.on("error", (err) => this._onError(err)); + } + + _onOpen() { + this.connected = true; + this.backoffMs = RECONNECT_BASE_MS; + this.emit("status", { state: "connected" }); + + // Send connect handshake (Gateway Protocol v3) + this._send({ + type: "req", + id: nextReqId(), + method: "connect", + params: { + minProtocol: 3, + maxProtocol: 3, + client: { + id: "spatial-office", + version: "0.1.0", + platform: "linux", + mode: "backend", + }, + role: "operator", + scopes: ["operator.read"], + auth: { token: this.gatewayToken }, + }, + }); + } + + _onMessage(data) { + let msg; + try { + msg = JSON.parse(data.toString()); + } catch { + this.emit("error", new Error(`Unparseable GW message: ${data.slice(0, 200)}`)); + return; + } + + // Route by message type + if (msg.type === "event") { + this._handleEvent(msg); + } else if (msg.type === "res") { + this._handleResponse(msg); + } else if (msg.type === "req") { + // Server requests (unlikely for our role, but handle gracefully) + this.emit("warn", `Unexpected server request: ${msg.method}`); + } + } + + _onClose(code, reason) { + this.connected = false; + this._clearTimers(); + this.emit("status", { state: "disconnected", code, reason: reason?.toString() }); + + // Mark all agents as offline + for (const agent of Object.values(this.agents)) { + const prev = agent.state; + agent.state = "offline"; + if (prev !== "offline") { + this._emitAgentChange(agent.id, { state: "offline" }); + } + } + + // Reconnect with backoff (unless dying) + if (!this.dying) { + const delay = Math.min(this.backoffMs, RECONNECT_MAX_MS); + this.backoffMs = Math.min(this.backoffMs * 2, RECONNECT_MAX_MS); + this.reconnectTimer = setTimeout(() => this._connect(), delay); + } + } + + _onError(err) { + this.emit("error", err); + } + + // ── Protocol Handlers ────────────────────────────────── + + _handleEvent(msg) { + const { event, payload } = msg; + + switch (event) { + case "tick": + // Periodic keepalive — update gateway uptime + this.emit("tick", payload); + break; + + case "presence": + // System presence snapshot + this._handlePresence(payload); + break; + + case "health": + // Gateway health + this.emit("health", payload); + break; + + case "session.message": + case "session.tool": + // Transcript/tool events for subscribed sessions + this._handleSessionEvent(msg); + break; + + case "sessions.changed": + // Session index changed → poll for updates + this._pollAgentSessions(); + break; + + default: + // Unknown events are ignored + break; + } + } + + _handleResponse(msg) { + const { id, ok, payload, error } = msg; + const pending = this.pendingRequests.get(id); + if (pending) { + this.pendingRequests.delete(id); + if (ok) { + pending.resolve(payload); + } else { + pending.reject(error || new Error("RPC failed")); + } + } + + // Handle connect response specially + if (ok && payload?.type === "hello-ok") { + this._onHelloOk(payload); + } + } + + _onHelloOk(hello) { + this.emit("status", { + state: "authenticated", + serverVersion: hello.server?.version, + connId: hello.server?.connId, + }); + + // Start post-connect setup + this._startHeartbeat(); + this._pollSystemPresence(); + this._pollAgentSessions(); + + // Start periodic agent session polling + this.pollTimer = setInterval( + () => this._pollAgentSessions(), + AGENT_POLL_INTERVAL_MS + ); + } + + // ── Presence Handling ────────────────────────────────── + + _handlePresence(presence) { + if (!presence) return; + + // Count connected clients + const clientIds = Object.keys(presence); + this.emit("meta", { + connectedClients: clientIds.length, + scottPresent: clientIds.length > 0, + }); + } + + async _pollSystemPresence() { + try { + const result = await this._rpc("system-presence", {}); + this._handlePresence(result); + } catch (err) { + this.emit("warn", `system-presence failed: ${err.message}`); + } + } + + // ── Agent Session Polling ────────────────────────────── + + async _pollAgentSessions() { + try { + const result = await this._rpc("sessions.list", {}); + if (!result?.sessions) return; + + const now = Date.now(); + const sessionsByAgent = new Map(); + + // Group sessions by agent + for (const session of result.sessions) { + const agentId = session.agentId || session.agent || "main"; + if (!sessionsByAgent.has(agentId)) { + sessionsByAgent.set(agentId, []); + } + sessionsByAgent.get(agentId).push(session); + } + + // Track which agents we've seen in this poll + const seenAgents = new Set(); + + for (const [agentId, sessions] of sessionsByAgent) { + seenAgents.add(agentId); + const latest = sessions[0]; // sessions.list returns most-recent first + const isActive = latest.status === "active" || latest.status === "working"; + + // Update activity tracker + this.lastAgentActivity.set(agentId, now); + + // Determine agent state and context + let state; + let taskContext = ""; + + if (isActive) { + state = "active"; + // Extract task context from session metadata + if (latest.lastMessage) { + taskContext = typeof latest.lastMessage === "string" + ? latest.lastMessage.slice(0, 200) + : ""; + } + if (latest.title) { + taskContext = latest.title; + } + if (latest.currentTool) { + taskContext = `Running: ${latest.currentTool}`; + } + } else { + // Check if agent is blocked (error status) + if (latest.status === "error" || latest.status === "blocked") { + state = "blocked"; + taskContext = latest.error || "blocked"; + } else { + state = "idle"; + } + } + + // Get or create agent entry + if (!this.agents[agentId]) { + this.agents[agentId] = { + id: agentId, + name: agentId, + department: inferDepartment(agentId), + state: "offline", + currentZone: departmentAffinityZone(inferDepartment(agentId)), + currentTask: null, + lastActivity: 0, + position: DEFAULT_POSITIONS[agentId] || { x: 0, y: 0 }, + }; + } + + const agent = this.agents[agentId]; + const prevState = agent.state; + + // Mark as online (was offline previously) + if (prevState === "offline") { + this._emitAgentChange(agentId, { state: "online" }); + } + + // Assign zone based on activity context + const newZone = assignZone(agentId, agent.department, state, taskContext); + const zoneChanged = newZone && newZone !== agent.currentZone; + + // Update agent state + agent.state = state; + agent.lastActivity = now; + agent.currentTask = taskContext || null; + + if (newZone) { + // Update zone occupancy + if (zoneChanged) { + this._updateZoneOccupancy(agentId, agent.currentZone, newZone); + agent.currentZone = newZone; + } + + // Position agent within zone + const existingPositions = Object.fromEntries( + Object.entries(this.agents) + .filter(([id]) => id !== agentId && this.agents[id].currentZone === newZone) + .map(([id, a]) => [id, a.position]) + ); + const pos = getZonePosition(newZone, agentId, existingPositions); + agent.position = pos; + } + + if (prevState !== state) { + this._emitAgentChange(agentId, { state, previousState: prevState }); + } + if (zoneChanged) { + this._emitAgentChange(agentId, { + zone: newZone, + previousZone: agent.currentZone, // note: currentZone already updated + }); + // Actually use the old zone for the event + const oldZone = this.zones[agent.currentZone] + ? agent.currentZone + : newZone; + this._emitAgentChange(agentId, { + zone: newZone, + previousZone: oldZone, + }); + } + } + + // Check for agents that haven't been seen recently → mark idle or offline + for (const [agentId, agent] of Object.entries(this.agents)) { + const lastSeen = this.lastAgentActivity.get(agentId) || 0; + const timeSinceActivity = now - lastSeen; + + if (timeSinceActivity > AGENT_IDLE_THRESHOLD_MS && agent.state !== "offline") { + if (agent.state !== "idle") { + agent.state = "idle"; + agent.currentTask = null; + + // Move to coffee bar when going idle + const oldZone = agent.currentZone; + const newZone = "coffee-bar"; + this._updateZoneOccupancy(agentId, oldZone, newZone); + agent.currentZone = newZone; + const pos = getZonePosition(newZone, agentId, {}); + agent.position = pos; + + this._emitAgentChange(agentId, { + state: "idle", + zone: newZone, + previousZone: oldZone, + }); + } + } + } + } catch (err) { + this.emit("warn", `sessions.list failed: ${err.message}`); + } + } + + _handleSessionEvent(msg) { + const { payload } = msg; + if (!payload?.agentId) return; + + const agentId = payload.agentId; + this.lastAgentActivity.set(agentId, Date.now()); + + // If agent was offline, mark as active + if (this.agents[agentId]?.state === "offline") { + this._emitAgentChange(agentId, { state: "active" }); + } + } + + // ── Zone Occupancy ───────────────────────────────────── + + _updateZoneOccupancy(agentId, oldZoneId, newZoneId) { + if (oldZoneId && this.zones[oldZoneId]) { + this.zones[oldZoneId].occupants = this.zones[oldZoneId].occupants.filter( + (id) => id !== agentId + ); + } + if (newZoneId && this.zones[newZoneId]) { + if (!this.zones[newZoneId].occupants.includes(agentId)) { + this.zones[newZoneId].occupants.push(agentId); + } + } + } + + // ── Event Emission ───────────────────────────────────── + + _emitAgentChange(agentId, changes) { + const agent = this.agents[agentId]; + if (!agent) return; + + const event = { + type: "agentStateChanged", + agentId, + changes, + agent: { ...agent }, + timestamp: Date.now(), + }; + + // Add to event log + this.events.push({ + type: "agentStateChanged", + agentId, + zoneId: agent.currentZone, + payload: changes, + timestamp: Date.now(), + }); + if (this.events.length > this.maxEvents) { + this.events.shift(); + } + + this.emit("agentStateChanged", event); + } + + // ── RPC Helpers ──────────────────────────────────────── + + _send(msg) { + if (this.ws && this.ws.readyState === WebSocket.OPEN) { + this.ws.send(JSON.stringify(msg)); + } + } + + /** + * Send an RPC request and wait for the response. + * @param {string} method + * @param {object} params + * @param {number} [timeoutMs=15000] + * @returns {Promise} + */ + _rpc(method, params = {}, timeoutMs = 15_000) { + return new Promise((resolve, reject) => { + const id = nextReqId(); + const timer = setTimeout(() => { + this.pendingRequests.delete(id); + reject(new Error(`RPC timeout: ${method} (${id})`)); + }, timeoutMs); + + this.pendingRequests.set(id, { + resolve: (result) => { + clearTimeout(timer); + resolve(result); + }, + reject: (err) => { + clearTimeout(timer); + reject(err); + }, + }); + + this._send({ type: "req", id, method, params }); + }); + } + + // ── Heartbeat / Keepalive ────────────────────────────── + + _startHeartbeat() { + this._clearTimers(); + this.pingTimer = setInterval(() => { + // Send a lightweight ping via tick subscription or system-presence + // The tick event from the gateway is enough — we just track it + if (this.ws?.readyState !== WebSocket.OPEN) { + this._onClose(1006, "ping timeout"); + return; + } + this._pollSystemPresence().catch(() => {}); + }, PING_INTERVAL_MS); + } + + _clearTimers() { + if (this.pingTimer) { + clearInterval(this.pingTimer); + this.pingTimer = null; + } + if (this.pollTimer) { + clearInterval(this.pollTimer); + this.pollTimer = null; + } + if (this.reconnectTimer) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + } + } + + // ── Public API ───────────────────────────────────────── + + /** Get current full state snapshot */ + getState() { + return { + agents: { ...this.agents }, + zones: { ...this.zones }, + events: [...this.events], + meta: { + gatewayUptime: this.connected ? 1 : 0, + connectedClients: 0, + scottPresent: false, + lastUpdate: Date.now(), + }, + }; + } + + /** Get a single agent's current state */ + getAgent(agentId) { + return this.agents[agentId] || null; + } + + /** Get all zone states */ + getZones() { + return { ...this.zones }; + } + + /** Get recent events (last N) */ + getEvents(limit = 50) { + return this.events.slice(-limit); + } +} + +// ── Default Export ─────────────────────────────────────── + +export default OpenClawBridge;