From 8028b1d1c385b3ec82e71392605bc0d03c357ef1 Mon Sep 17 00:00:00 2001 From: Ben Vinegar Date: Mon, 23 Feb 2026 16:23:23 -0500 Subject: [PATCH] bridge: include inbox.pull observability meta --- .env.schema | 5 ++ CONFIGURATION.md | 1 + slack-bridge/broker-bridge.mjs | 72 +++++++++++++++++++++++-- test/broker-bridge.integration.test.mjs | 12 +++++ 4 files changed, 86 insertions(+), 4 deletions(-) diff --git a/.env.schema b/.env.schema index 899c38b..363d272 100644 --- a/.env.schema +++ b/.env.schema @@ -148,6 +148,11 @@ SLACK_BROKER_ACCESS_TOKEN_EXPIRES_AT= # @sensitive=false @type=string SLACK_BROKER_ACCESS_TOKEN_SCOPES= +# Optional agent version override used in broker observability metadata. +# If unset, broker bridge falls back to ~/.pi/agent/baudbot-version.json (if present). +# @sensitive=false @type=string +BAUDBOT_AGENT_VERSION= + # Broker pull cadence in milliseconds (default: 3000) # @sensitive=false @type=number SLACK_BROKER_POLL_INTERVAL_MS=3000 diff --git a/CONFIGURATION.md b/CONFIGURATION.md index 1bb9e4d..df95dab 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -111,6 +111,7 @@ Set by `sudo baudbot broker register` when using brokered Slack OAuth flow. | `SLACK_BROKER_MAX_MESSAGES` | Max leased messages per poll request (default: `10`) | | `SLACK_BROKER_WAIT_SECONDS` | Long-poll wait window for `/api/inbox/pull` (default: `20`, set `0` for immediate short-poll, max `25`) | | `SLACK_BROKER_DEDUPE_TTL_MS` | Dedupe cache TTL in milliseconds (default: `1200000`) | +| `BAUDBOT_AGENT_VERSION` | Optional override for broker observability `meta.agent_version` (otherwise read from `~/.pi/agent/baudbot-version.json` when available) | ### Kernel (Cloud Browsers) diff --git a/slack-bridge/broker-bridge.mjs b/slack-bridge/broker-bridge.mjs index 07a5b58..55268d8 100755 --- a/slack-bridge/broker-bridge.mjs +++ b/slack-bridge/broker-bridge.mjs @@ -9,7 +9,7 @@ import * as net from "node:net"; import * as fs from "node:fs"; import * as path from "node:path"; -import { homedir } from "node:os"; +import { homedir, uptime as getSystemUptimeSeconds } from "node:os"; import { createServer } from "node:http"; import sodium from "libsodium-wrappers-sumo"; import { @@ -52,6 +52,7 @@ const DEDUPE_TTL_MS = clampInt( const MAX_BACKOFF_MS = 30_000; const INBOX_PROTOCOL_VERSION = "2026-02-1"; const BROKER_HEALTH_PATH = path.join(homedir(), ".pi", "agent", "broker-health.json"); +const BAUDBOT_VERSION_PATH = path.join(homedir(), ".pi", "agent", "baudbot-version.json"); const LOG_BUFFER_MAX_LINES = 1000; const logLineBuffer = []; @@ -153,6 +154,8 @@ let cryptoState = null; const dedupe = new Map(); let brokerTokenExpiryFormatWarned = false; +let brokerPollCount = 0; +const bridgeStartedAtMs = Date.now(); const brokerHealth = { started_at: new Date().toISOString(), @@ -185,6 +188,67 @@ const brokerHealth = { }, }; +function readAgentVersion() { + const explicitVersion = String(process.env.BAUDBOT_AGENT_VERSION || "").trim(); + if (explicitVersion) return explicitVersion; + + try { + const raw = fs.readFileSync(BAUDBOT_VERSION_PATH, "utf8"); + const parsed = JSON.parse(raw); + if (typeof parsed.short === "string" && parsed.short.trim()) return parsed.short.trim(); + if (typeof parsed.sha === "string" && parsed.sha.trim()) return parsed.sha.trim(); + } catch { + // Ignore read/parse failures. Observability metadata falls back to "unknown". + } + + return "unknown"; +} + +const agentVersion = readAgentVersion(); + +function countActivePiSessions() { + try { + const entries = fs.readdirSync(SOCKET_DIR, { withFileTypes: true }); + let activeSessions = 0; + let activeDevAgents = 0; + + for (const entry of entries) { + const name = entry.name; + if (name.endsWith(".sock")) { + activeSessions += 1; + } + if (/^dev-agent-.+\.alias$/.test(name) && (entry.isFile() || entry.isSymbolicLink())) { + activeDevAgents += 1; + } + } + + return { activeSessions, activeDevAgents }; + } catch { + return { activeSessions: 0, activeDevAgents: 0 }; + } +} + +function buildPullMeta(maxMessages, waitSeconds) { + const { activeSessions, activeDevAgents } = countActivePiSessions(); + const bridgeUptimeHours = Math.max(0, (Date.now() - bridgeStartedAtMs) / (1000 * 60 * 60)); + const systemUptimeHours = Math.max(0, getSystemUptimeSeconds() / (60 * 60)); + + return { + agent_version: agentVersion, + bridge_uptime_hours: bridgeUptimeHours, + system_uptime_hours: systemUptimeHours, + heartbeat_runs: brokerPollCount, + heartbeat_consecutive_errors: brokerHealth.poll.consecutive_failures, + heartbeat_last_ok_at: brokerHealth.poll.last_ok_at, + active_sessions: activeSessions, + active_dev_agents: activeDevAgents, + outbound_mode: outboundMode, + poll_count: brokerPollCount + 1, + max_messages: maxMessages, + wait_seconds: waitSeconds, + }; +} + function trimError(err) { const msg = err instanceof Error ? err.message : String(err || "unknown error"); return msg.slice(0, 400); @@ -496,6 +560,7 @@ async function pullInbox() { wait_seconds: BROKER_WAIT_SECONDS, timestamp, signature, + meta: buildPullMeta(MAX_MESSAGES, BROKER_WAIT_SECONDS), }; const inboxPullResponseBody = await brokerFetch("/api/inbox/pull", inboxPullRequestBody); @@ -941,7 +1006,6 @@ function startApiServer() { async function startPollLoop() { let backoffMs = POLL_INTERVAL_MS; - let pollCount = 0; let lastStatusLog = Date.now(); const STATUS_LOG_INTERVAL_MS = 60_000; // log a status line every 60s even when idle @@ -953,7 +1017,7 @@ async function startPollLoop() { const messages = await pullInbox(); pollSucceeded = true; markHealth("poll", true); - pollCount++; + brokerPollCount++; const ackIds = []; if (messages.length > 0) { @@ -962,7 +1026,7 @@ async function startPollLoop() { // Periodic idle status log so you know the bridge is alive if (messages.length === 0 && Date.now() - lastStatusLog >= STATUS_LOG_INTERVAL_MS) { - logInfo(`💤 idle — ${pollCount} polls since start, dedupe cache: ${dedupe.size} entries`); + logInfo(`💤 idle — ${brokerPollCount} polls since start, dedupe cache: ${dedupe.size} entries`); lastStatusLog = Date.now(); } diff --git a/test/broker-bridge.integration.test.mjs b/test/broker-bridge.integration.test.mjs index 54f1eaa..798a508 100644 --- a/test/broker-bridge.integration.test.mjs +++ b/test/broker-bridge.integration.test.mjs @@ -569,6 +569,18 @@ describe("broker pull bridge semi-integration", () => { expect(pullPayload.protocol_version).toBe("2026-02-1"); expect(pullPayload.max_messages).toBe(10); expect(pullPayload.wait_seconds).toBe(20); + expect(pullPayload.meta).toBeTruthy(); + expect(pullPayload.meta.outbound_mode).toBe("broker"); + expect(pullPayload.meta.poll_count).toBeGreaterThanOrEqual(1); + expect(pullPayload.meta.max_messages).toBe(10); + expect(pullPayload.meta.wait_seconds).toBe(20); + expect(typeof pullPayload.meta.bridge_uptime_hours).toBe("number"); + expect(typeof pullPayload.meta.system_uptime_hours).toBe("number"); + expect(typeof pullPayload.meta.active_sessions).toBe("number"); + expect(typeof pullPayload.meta.active_dev_agents).toBe("number"); + expect(typeof pullPayload.meta.agent_version).toBe("string"); + expect(pullPayload.meta.heartbeat_runs).toBeGreaterThanOrEqual(0); + expect(pullPayload.meta.heartbeat_consecutive_errors).toBeGreaterThanOrEqual(0); const canonical = canonicalizeProtocolRequest(workspaceId, "2026-02-1", "inbox.pull", pullPayload.timestamp, { max_messages: 10,