bridge: adopt protocol-versioned inbox pull/ack signing

Author: benvinegar

CONFIGURATION.md

@@ -106,7 +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_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`) |
 
 ### Kernel (Cloud Browsers)

README.md

@@ -98,7 +98,7 @@ 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).
+Broker pull mode uses long-polling by default (`SLACK_BROKER_WAIT_SECONDS=20`, max `25`; set `0` for immediate short-poll behavior).
 
 Need to rotate/update a key later?
 

slack-bridge/broker-bridge.mjs

@@ -24,8 +24,7 @@ import {
 } from "./security.mjs";
 import {
   canonicalizeEnvelope,
-  canonicalizeOutbound,
-  canonicalizeOutboundV2,
+  canonicalizeProtocolRequest,
   canonicalizeSendRequest,
 } from "./crypto.mjs";
 
@@ -50,6 +49,7 @@ const DEDUPE_TTL_MS = clampInt(
   20 * 60 * 1000,
 );
 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");
 
 function ts() {
@@ -366,23 +366,23 @@ function getThreadId(channel, threadTs) {
   return id;
 }
 
-function signRequest(action, timestamp, payloadField) {
-  const canonical = canonicalizeOutbound(workspaceId, action, timestamp, payloadField);
+function signProtocolRequest(action, timestamp, payload) {
+  const canonical = canonicalizeProtocolRequest(
+    workspaceId,
+    INBOX_PROTOCOL_VERSION,
+    action,
+    timestamp,
+    payload,
+  );
   const sig = sodium.crypto_sign_detached(canonical, cryptoState.serverSignSecretKey);
   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, {
+  return signProtocolRequest("inbox.pull", timestamp, {
     max_messages: maxMessages,
     wait_seconds: waitSeconds,
   });
-  const sig = sodium.crypto_sign_detached(canonical, cryptoState.serverSignSecretKey);
-  return toBase64(sig);
 }
 
 async function brokerFetch(pathname, body) {
@@ -421,8 +421,9 @@ async function pullInbox() {
 
   const body = {
     workspace_id: workspaceId,
+    protocol_version: INBOX_PROTOCOL_VERSION,
     max_messages: MAX_MESSAGES,
-    ...(BROKER_WAIT_SECONDS > 0 ? { wait_seconds: BROKER_WAIT_SECONDS } : {}),
+    wait_seconds: BROKER_WAIT_SECONDS,
     timestamp,
     signature,
   };
@@ -435,11 +436,11 @@ async function pullInbox() {
 async function ackInbox(messageIds) {
   if (messageIds.length === 0) return;
   const timestamp = Math.floor(Date.now() / 1000);
-  const joined = messageIds.join(",");
-  const signature = signRequest("inbox.ack", timestamp, joined);
+  const signature = signProtocolRequest("inbox.ack", timestamp, { message_ids: messageIds });
 
   await brokerFetch("/api/inbox/ack", {
     workspace_id: workspaceId,
+    protocol_version: INBOX_PROTOCOL_VERSION,
     message_ids: messageIds,
     timestamp,
     signature,
@@ -993,6 +994,7 @@ async function startPollLoop() {
   logInfo(`   outbound mode: ${outboundMode} ${outboundMode === "direct" ? "(using SLACK_BOT_TOKEN)" : "(via broker)"}`);
   logInfo(`   broker: ${brokerBaseUrl}`);
   logInfo(`   workspace: ${workspaceId}`);
+  logInfo(`   inbox protocol: ${INBOX_PROTOCOL_VERSION}`);
   logInfo(
     `   poll mode: ${BROKER_WAIT_SECONDS > 0 ? `long-poll (${BROKER_WAIT_SECONDS}s)` : "short-poll"}, ` +
     `interval: ${POLL_INTERVAL_MS}ms, max messages: ${MAX_MESSAGES}`,

slack-bridge/crypto.mjs

@@ -47,15 +47,16 @@ export function canonicalizeOutbound(workspace, action, timestamp, encryptedBody
 }
 
 /**
- * Construct canonical bytes for v2 outbound request signing.
+ * Construct canonical bytes for protocol-versioned inbox pull/ack signing.
  *
  * Uses deterministic JSON serialization (sorted keys) to match broker
  * json-stable-stringify canonicalization.
  */
-export function canonicalizeOutboundV2(workspace, action, timestamp, payload) {
+export function canonicalizeProtocolRequest(workspace, protocolVersion, action, timestamp, payload) {
   return utf8Bytes(
     stableStringify({
       workspace_id: workspace,
+      protocol_version: protocolVersion,
       action,
       timestamp,
       payload,

slack-bridge/crypto.test.mjs

@@ -10,7 +10,7 @@ import {
   stableStringify,
   canonicalizeEnvelope,
   canonicalizeOutbound,
-  canonicalizeOutboundV2,
+  canonicalizeProtocolRequest,
   canonicalizeSendRequest,
 } from "./crypto.mjs";
 
@@ -152,26 +152,25 @@ describe("canonicalizeOutbound", () => {
   });
 });
 
-// ── canonicalizeOutboundV2 ──────────────────────────────────────────────────
+// ── canonicalizeProtocolRequest ─────────────────────────────────────────────
 
-describe("canonicalizeOutboundV2", () => {
-  it("produces stable JSON payload for inbox.pull.v2", () => {
+describe("canonicalizeProtocolRequest", () => {
+  it("produces stable JSON payload for protocol-versioned inbox.pull", () => {
     const result = decode(
-      canonicalizeOutboundV2("T123", "inbox.pull.v2", 1700000000, {
+      canonicalizeProtocolRequest("T123", "2026-02-1", "inbox.pull", 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"}',
+      '{"action":"inbox.pull","payload":{"max_messages":10,"wait_seconds":20},"protocol_version":"2026-02-1","timestamp":1700000000,"workspace_id":"T123"}',
     );
   });
 
   it("returns Uint8Array", () => {
-    const result = canonicalizeOutboundV2("T123", "inbox.pull.v2", 1700000000, {
-      max_messages: 10,
-      wait_seconds: 20,
+    const result = canonicalizeProtocolRequest("T123", "2026-02-1", "inbox.ack", 1700000000, {
+      message_ids: ["m1", "m2"],
     });
     assert.ok(result instanceof Uint8Array);
   });

test/broker-bridge.integration.test.mjs

@@ -9,8 +9,7 @@ import { tmpdir } from "node:os";
 import sodium from "libsodium-wrappers-sumo";
 import {
   canonicalizeEnvelope,
-  canonicalizeOutbound,
-  canonicalizeOutboundV2,
+  canonicalizeProtocolRequest,
 } from "../slack-bridge/crypto.mjs";
 
 function b64(bytes = 32, fill = 1) {
@@ -54,6 +53,8 @@ describe("broker pull bridge semi-integration", () => {
   });
 
   it("acks poison messages from broker to avoid infinite retry loops", async () => {
+    await sodium.ready;
+
     let pullCount = 0;
     let ackPayload = null;
 
@@ -162,7 +163,16 @@ describe("broker pull bridge semi-integration", () => {
     await Promise.race([ackWait, bridgeExited]);
 
     expect(ackPayload.workspace_id).toBe("T123BROKER");
+    expect(ackPayload.protocol_version).toBe("2026-02-1");
     expect(ackPayload.message_ids).toContain("m-poison-1");
+
+    const signKeypair = sodium.crypto_sign_seed_keypair(new Uint8Array(Buffer.alloc(32, 13)));
+    const canonical = canonicalizeProtocolRequest("T123BROKER", "2026-02-1", "inbox.ack", ackPayload.timestamp, {
+      message_ids: ackPayload.message_ids,
+    });
+    const sigBytes = new Uint8Array(Buffer.from(ackPayload.signature, "base64"));
+    const valid = sodium.crypto_sign_verify_detached(sigBytes, canonical, signKeypair.publicKey);
+    expect(valid).toBe(true);
   });
 
   it("forwards user messages to agent in fire-and-forget mode without get_message/turn_end RPCs", async () => {
@@ -352,7 +362,7 @@ describe("broker pull bridge semi-integration", () => {
     expect(sendPayloads.some((payload) => payload.action === "reactions.add")).toBe(false);
   });
 
-  it("uses inbox.pull.v2 signatures with wait_seconds by default", async () => {
+  it("uses protocol-versioned inbox.pull signatures with wait_seconds by default", async () => {
     await sodium.ready;
 
     const workspaceId = "T123BROKER";
@@ -423,10 +433,11 @@ describe("broker pull bridge semi-integration", () => {
     await waitFor(() => pullPayload !== null, 10_000, 50, "timeout waiting for inbox pull request");
 
     expect(pullPayload.workspace_id).toBe(workspaceId);
+    expect(pullPayload.protocol_version).toBe("2026-02-1");
     expect(pullPayload.max_messages).toBe(10);
     expect(pullPayload.wait_seconds).toBe(20);
 
-    const canonical = canonicalizeOutboundV2(workspaceId, "inbox.pull.v2", pullPayload.timestamp, {
+    const canonical = canonicalizeProtocolRequest(workspaceId, "2026-02-1", "inbox.pull", pullPayload.timestamp, {
       max_messages: 10,
       wait_seconds: 20,
     });
@@ -437,7 +448,7 @@ describe("broker pull bridge semi-integration", () => {
     bridge.kill("SIGTERM");
   });
 
-  it("falls back to legacy inbox.pull signature when SLACK_BROKER_WAIT_SECONDS=0", async () => {
+  it("uses protocol-versioned inbox.pull signature with wait_seconds=0", async () => {
     await sodium.ready;
 
     const workspaceId = "T123BROKER";
@@ -506,13 +517,17 @@ describe("broker pull bridge semi-integration", () => {
     });
     children.push(bridge);
 
-    await waitFor(() => pullPayload !== null, 10_000, 50, "timeout waiting for legacy inbox pull request");
+    await waitFor(() => pullPayload !== null, 10_000, 50, "timeout waiting for protocol inbox pull request");
 
     expect(pullPayload.workspace_id).toBe(workspaceId);
+    expect(pullPayload.protocol_version).toBe("2026-02-1");
     expect(pullPayload.max_messages).toBe(10);
-    expect(Object.prototype.hasOwnProperty.call(pullPayload, "wait_seconds")).toBe(false);
+    expect(pullPayload.wait_seconds).toBe(0);
 
-    const canonical = canonicalizeOutbound(workspaceId, "inbox.pull", pullPayload.timestamp, "10");
+    const canonical = canonicalizeProtocolRequest(workspaceId, "2026-02-1", "inbox.pull", pullPayload.timestamp, {
+      max_messages: 10,
+      wait_seconds: 0,
+    });
     const sigBytes = new Uint8Array(Buffer.from(pullPayload.signature, "base64"));
     const valid = sodium.crypto_sign_verify_detached(sigBytes, canonical, signKeypair.publicKey);
     expect(valid).toBe(true);
@@ -592,10 +607,11 @@ describe("broker pull bridge semi-integration", () => {
     await waitFor(() => pullPayload !== null, 10_000, 50, "timeout waiting for clamped inbox pull request");
 
     expect(pullPayload.workspace_id).toBe(workspaceId);
+    expect(pullPayload.protocol_version).toBe("2026-02-1");
     expect(pullPayload.max_messages).toBe(100);
     expect(pullPayload.wait_seconds).toBe(20);
 
-    const canonical = canonicalizeOutboundV2(workspaceId, "inbox.pull.v2", pullPayload.timestamp, {
+    const canonical = canonicalizeProtocolRequest(workspaceId, "2026-02-1", "inbox.pull", pullPayload.timestamp, {
       max_messages: 100,
       wait_seconds: 20,
     });