bridge: add broker long-poll v2 pull signing

Author: benvinegar

AGENTS.md

@@ -150,6 +150,7 @@ tmux new-window -n baudbot 'sudo -u baudbot_agent ~/runtime/start.sh'
 - After a successful broker registration, always restart to load new keys:
   - `sudo baudbot restart`
 - The runtime starts `broker-bridge.mjs` automatically when `SLACK_BROKER_*` vars are present.
+- Optional tuning: set `SLACK_BROKER_WAIT_SECONDS` (default `20`, max `25`) to enable broker inbox long-polling; set `0` for legacy short-poll behavior.
 - Quick troubleshooting when Slack replies stop:
   - `sudo -u baudbot_agent tmux ls` (check `slack-bridge` session exists)
   - `sudo baudbot attach --tmux slack-bridge` (bridge logs)

CONFIGURATION.md

@@ -106,6 +106,7 @@ Set by `sudo baudbot broker register` when using brokered Slack OAuth flow.
 | `SLACK_BROKER_SIGNING_PUBLIC_KEY` | Broker Ed25519 public signing key (base64) |
 | `SLACK_BROKER_POLL_INTERVAL_MS` | Inbox poll interval in milliseconds (default: `3000`) |
 | `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 legacy short-poll, max `25`) |
 | `SLACK_BROKER_DEDUPE_TTL_MS` | Dedupe cache TTL in milliseconds (default: `1200000`) |
 
 ### Kernel (Cloud Browsers)
@@ -190,6 +191,7 @@ SLACK_BROKER_URL=https://broker.example.com
 SLACK_BROKER_WORKSPACE_ID=T0123ABCD
 SLACK_BROKER_POLL_INTERVAL_MS=3000
 SLACK_BROKER_MAX_MESSAGES=10
+SLACK_BROKER_WAIT_SECONDS=20
 SLACK_BROKER_DEDUPE_TTL_MS=1200000
 
 # Experimental features (required for email)

README.md

@@ -98,6 +98,8 @@ sudo baudbot broker register \
   --registration-token <token-from-dashboard-callback>
 ```
 
+Broker pull mode uses long-polling by default (`SLACK_BROKER_WAIT_SECONDS=20`, max `25`; set `0` for legacy short-poll).
+
 Need to rotate/update a key later?
 
 ```bash

bin/config.sh

@@ -451,6 +451,7 @@ else
     SLACK_BROKER_SIGNING_PUBLIC_KEY \
     SLACK_BROKER_POLL_INTERVAL_MS \
     SLACK_BROKER_MAX_MESSAGES \
+    SLACK_BROKER_WAIT_SECONDS \
     SLACK_BROKER_DEDUPE_TTL_MS
 
   prompt_secret "SLACK_BOT_TOKEN" \
@@ -613,6 +614,7 @@ ordered_keys=(
   SLACK_BROKER_SIGNING_PUBLIC_KEY
   SLACK_BROKER_POLL_INTERVAL_MS
   SLACK_BROKER_MAX_MESSAGES
+  SLACK_BROKER_WAIT_SECONDS
   SLACK_BROKER_DEDUPE_TTL_MS
   BAUDBOT_AGENT_USER
   BAUDBOT_AGENT_HOME

slack-bridge/broker-bridge.mjs

@@ -25,15 +25,30 @@ import {
 import {
   canonicalizeEnvelope,
   canonicalizeOutbound,
+  canonicalizeOutboundV2,
   canonicalizeSendRequest,
 } from "./crypto.mjs";
 
 const SOCKET_DIR = path.join(homedir(), ".pi", "session-control");
 const AGENT_TIMEOUT_MS = 120_000;
-const API_PORT = parseInt(process.env.BRIDGE_API_PORT || "7890", 10);
-const POLL_INTERVAL_MS = parseInt(process.env.SLACK_BROKER_POLL_INTERVAL_MS || "3000", 10);
-const MAX_MESSAGES = parseInt(process.env.SLACK_BROKER_MAX_MESSAGES || "10", 10);
-const DEDUPE_TTL_MS = parseInt(process.env.SLACK_BROKER_DEDUPE_TTL_MS || String(20 * 60 * 1000), 10);
+
+function clampInt(value, min, max, fallback) {
+  const parsed = Number.parseInt(String(value ?? ""), 10);
+  if (!Number.isFinite(parsed)) return fallback;
+  return Math.min(max, Math.max(min, parsed));
+}
+
+const API_PORT = clampInt(process.env.BRIDGE_API_PORT || "7890", 1, 65535, 7890);
+const POLL_INTERVAL_MS = clampInt(process.env.SLACK_BROKER_POLL_INTERVAL_MS || "3000", 0, 60_000, 3000);
+const MAX_MESSAGES = clampInt(process.env.SLACK_BROKER_MAX_MESSAGES || "10", 1, 100, 10);
+const MAX_WAIT_SECONDS = 25;
+const BROKER_WAIT_SECONDS = clampInt(process.env.SLACK_BROKER_WAIT_SECONDS || "20", 0, MAX_WAIT_SECONDS, 20);
+const DEDUPE_TTL_MS = clampInt(
+  process.env.SLACK_BROKER_DEDUPE_TTL_MS || String(20 * 60 * 1000),
+  1_000,
+  7 * 24 * 60 * 60 * 1000,
+  20 * 60 * 1000,
+);
 const MAX_BACKOFF_MS = 30_000;
 const BROKER_HEALTH_PATH = path.join(homedir(), ".pi", "agent", "broker-health.json");
 
@@ -357,6 +372,19 @@ function signRequest(action, timestamp, payloadField) {
   return toBase64(sig);
 }
 
+function signPullRequest(timestamp, maxMessages, waitSeconds) {
+  if (waitSeconds <= 0) {
+    return signRequest("inbox.pull", timestamp, String(maxMessages));
+  }
+
+  const canonical = canonicalizeOutboundV2(workspaceId, "inbox.pull.v2", timestamp, {
+    max_messages: maxMessages,
+    wait_seconds: waitSeconds,
+  });
+  const sig = sodium.crypto_sign_detached(canonical, cryptoState.serverSignSecretKey);
+  return toBase64(sig);
+}
+
 async function brokerFetch(pathname, body) {
   const url = `${brokerBaseUrl}${pathname}`;
   const response = await fetch(url, {
@@ -389,14 +417,17 @@ async function brokerFetch(pathname, body) {
 
 async function pullInbox() {
   const timestamp = Math.floor(Date.now() / 1000);
-  const signature = signRequest("inbox.pull", timestamp, String(MAX_MESSAGES));
+  const signature = signPullRequest(timestamp, MAX_MESSAGES, BROKER_WAIT_SECONDS);
 
-  const payload = await brokerFetch("/api/inbox/pull", {
+  const body = {
     workspace_id: workspaceId,
     max_messages: MAX_MESSAGES,
+    ...(BROKER_WAIT_SECONDS > 0 ? { wait_seconds: BROKER_WAIT_SECONDS } : {}),
     timestamp,
     signature,
-  });
+  };
+
+  const payload = await brokerFetch("/api/inbox/pull", body);
 
   return Array.isArray(payload.messages) ? payload.messages : [];
 }
@@ -918,7 +949,9 @@ async function startPollLoop() {
       }
 
       backoffMs = POLL_INTERVAL_MS;
-      await sleep(POLL_INTERVAL_MS);
+      if (BROKER_WAIT_SECONDS <= 0) {
+        await sleep(POLL_INTERVAL_MS);
+      }
     } catch (err) {
       if (!pollSucceeded) {
         markHealth("poll", false, err);
@@ -960,7 +993,10 @@ async function startPollLoop() {
   logInfo(`   outbound mode: ${outboundMode} ${outboundMode === "direct" ? "(using SLACK_BOT_TOKEN)" : "(via broker)"}`);
   logInfo(`   broker: ${brokerBaseUrl}`);
   logInfo(`   workspace: ${workspaceId}`);
-  logInfo(`   poll interval: ${POLL_INTERVAL_MS}ms, max messages: ${MAX_MESSAGES}`);
+  logInfo(
+    `   poll mode: ${BROKER_WAIT_SECONDS > 0 ? `long-poll (${BROKER_WAIT_SECONDS}s)` : "short-poll"}, ` +
+    `interval: ${POLL_INTERVAL_MS}ms, max messages: ${MAX_MESSAGES}`,
+  );
   logInfo(`   allowed users: ${ALLOWED_USERS.length || "all"}`);
   logInfo(`   pi socket: ${socketPath || "(not found — will retry on message)"}`);
   await startPollLoop();

slack-bridge/crypto.mjs

@@ -46,6 +46,23 @@ export function canonicalizeOutbound(workspace, action, timestamp, encryptedBody
   return utf8Bytes(`${workspace}|${action}|${timestamp}|${encryptedBody}`);
 }
 
+/**
+ * Construct canonical bytes for v2 outbound request signing.
+ *
+ * Uses deterministic JSON serialization (sorted keys) to match broker
+ * json-stable-stringify canonicalization.
+ */
+export function canonicalizeOutboundV2(workspace, action, timestamp, payload) {
+  return utf8Bytes(
+    stableStringify({
+      workspace_id: workspace,
+      action,
+      timestamp,
+      payload,
+    }),
+  );
+}
+
 /**
  * Construct canonical bytes for /api/send request signing.
  *

slack-bridge/crypto.test.mjs

@@ -10,6 +10,7 @@ import {
   stableStringify,
   canonicalizeEnvelope,
   canonicalizeOutbound,
+  canonicalizeOutboundV2,
   canonicalizeSendRequest,
 } from "./crypto.mjs";
 
@@ -151,6 +152,31 @@ describe("canonicalizeOutbound", () => {
   });
 });
 
+// ── canonicalizeOutboundV2 ──────────────────────────────────────────────────
+
+describe("canonicalizeOutboundV2", () => {
+  it("produces stable JSON payload for inbox.pull.v2", () => {
+    const result = decode(
+      canonicalizeOutboundV2("T123", "inbox.pull.v2", 1700000000, {
+        max_messages: 10,
+        wait_seconds: 20,
+      }),
+    );
+    assert.equal(
+      result,
+      '{"action":"inbox.pull.v2","payload":{"max_messages":10,"wait_seconds":20},"timestamp":1700000000,"workspace_id":"T123"}',
+    );
+  });
+
+  it("returns Uint8Array", () => {
+    const result = canonicalizeOutboundV2("T123", "inbox.pull.v2", 1700000000, {
+      max_messages: 10,
+      wait_seconds: 20,
+    });
+    assert.ok(result instanceof Uint8Array);
+  });
+});
+
 // ── canonicalizeSendRequest ─────────────────────────────────────────────────
 
 describe("canonicalizeSendRequest", () => {