Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .env.schema
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions CONFIGURATION.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
72 changes: 68 additions & 4 deletions slack-bridge/broker-bridge.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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 = [];
Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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

Expand All @@ -953,7 +1017,7 @@ async function startPollLoop() {
const messages = await pullInbox();
pollSucceeded = true;
markHealth("poll", true);
pollCount++;
brokerPollCount++;
const ackIds = [];

if (messages.length > 0) {
Expand All @@ -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();
}

Expand Down
12 changes: 12 additions & 0 deletions test/broker-bridge.integration.test.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down