Add OpenClaw bridge runtime and provisioning support

CONTRIBUTING.md

@@ -17,7 +17,7 @@ Requires Node 22+, pnpm 9+, and a Go toolchain.
 pnpm typecheck
 pnpm test
 pnpm build
-go test ./...   # run from packages/pickle/native
+pnpm test:go    # runs Pickle's Go tests with the goolm build tag
 ```
 
 ## Release

package.json

@@ -1,5 +1,5 @@
 {
-  "name": "pickle-monorepo",
+  "name": "@beeper/openclaw",
   "private": true,
   "type": "module",
   "packageManager": "pnpm@10.25.0",
@@ -15,7 +15,7 @@
     "smoke:cloudflare": "node scripts/smoke-cloudflare-worker.mjs",
     "smoke:consumer": "node scripts/package-consumer-smoke.mjs",
     "smoke:package-consumer": "node scripts/package-consumer-smoke.mjs",
-    "test:go": "cd packages/pickle/native && go test -tags goolm ./...",
+    "test:go": "pnpm --filter @beeper/pickle test:go",
     "test:e2e": "pnpm build && pnpm --dir e2e test",
     "test:e2e:adapter": "pnpm build && pnpm --dir e2e test:adapter",
     "test:e2e:browser:serve": "pnpm --dir e2e test:browser:serve",

packages/bridge/src/appservice-websocket.test.ts

@@ -16,13 +16,13 @@ afterEach(async () => {
 });
 
 describe("AppserviceWebsocket", () => {
-  it("connects to as_sync, dispatches transactions, and acknowledges them", async () => {
+  it("connects to as_sync, forwards transactions, and acknowledges them", async () => {
     const httpServer = createServer();
     const wsServer = new WebSocketServer({ server: httpServer });
     servers.push(wsServer, httpServer);
     await new Promise<void>((resolve) => httpServer.listen(0, "127.0.0.1", resolve));
     const homeserver = `http://127.0.0.1:${(httpServer.address() as AddressInfo).port}/_hungryserv/alice`;
-    const dispatch = vi.fn(async () => {});
+    const handleTransaction = vi.fn(async () => {});
     const connected = new Promise<void>((resolve, reject) => {
       wsServer.on("connection", (socket, request) => {
         try {
@@ -55,19 +55,21 @@ describe("AppserviceWebsocket", () => {
       });
     });
     const websocket = createWebsocket(homeserver, {
-      dispatch,
+      handleTransaction,
       log: (() => {}) as BridgeLogger,
     });
     websockets.push(websocket);
 
     websocket.start();
     await connected;
 
-    expect(dispatch).toHaveBeenCalledWith(expect.objectContaining({
-      eventId: "$event",
-      kind: "message",
-      roomId: "!room:example",
-      text: "hi",
+    expect(handleTransaction).toHaveBeenCalledWith(expect.objectContaining({
+      events: [expect.objectContaining({
+        event_id: "$event",
+        room_id: "!room:example",
+        type: "m.room.message",
+      })],
+      txn_id: "txn-1",
     }));
   });
 
@@ -147,7 +149,6 @@ describe("AppserviceWebsocket", () => {
     servers.push(wsServer, httpServer);
     await new Promise<void>((resolve) => httpServer.listen(0, "127.0.0.1", resolve));
     const homeserver = `http://127.0.0.1:${(httpServer.address() as AddressInfo).port}/_hungryserv/alice`;
-    const dispatch = vi.fn(async () => {});
     const handleTransaction = vi.fn(async () => {});
     const connected = new Promise<void>((resolve, reject) => {
       wsServer.on("connection", (socket) => {
@@ -183,7 +184,6 @@ describe("AppserviceWebsocket", () => {
       });
     });
     const websocket = createWebsocket(homeserver, {
-      dispatch,
       handleTransaction,
       log: (() => {}) as BridgeLogger,
     });
@@ -192,11 +192,6 @@ describe("AppserviceWebsocket", () => {
     websocket.start();
     await connected;
 
-    expect(dispatch).toHaveBeenCalledWith(expect.objectContaining({
-      eventId: "$proxied",
-      kind: "message",
-      text: "proxied",
-    }));
     expect(handleTransaction).toHaveBeenCalledWith(expect.objectContaining({
       events: [expect.objectContaining({ event_id: "$proxied" })],
       txn_id: "txn-2",
@@ -327,7 +322,6 @@ function createWebsocket(
         url: "",
       },
     },
-    dispatch: vi.fn(async () => {}),
     log: (() => {}) as BridgeLogger,
     ...overrides,
   });

packages/bridge/src/appservice-websocket.ts

@@ -1,10 +1,9 @@
 import WebSocket from "ws";
-import type { MatrixAppserviceInitOptions, MatrixClientEvent } from "@beeper/pickle";
+import type { MatrixAppserviceInitOptions } from "@beeper/pickle";
 import type { BridgeLogger } from "./types";
 
 export interface AppserviceWebsocketOptions {
   appservice: MatrixAppserviceInitOptions;
-  dispatch(event: MatrixClientEvent): Promise<unknown>;
   handleHTTPProxy?(request: HTTPProxyRequest): Promise<HTTPProxyResponse | null>;
   handleTransaction?(transaction: Record<string, unknown>): Promise<unknown>;
   log: BridgeLogger;
@@ -40,7 +39,6 @@ export class AppserviceWebsocket {
   };
 
   readonly #appservice: MatrixAppserviceInitOptions;
-  readonly #dispatch: (event: MatrixClientEvent) => Promise<unknown>;
   readonly #handleProxy: ((request: HTTPProxyRequest) => Promise<HTTPProxyResponse | null>) | undefined;
   readonly #handleTransaction: ((transaction: Record<string, unknown>) => Promise<unknown>) | undefined;
   readonly #log: BridgeLogger;
@@ -61,7 +59,6 @@ export class AppserviceWebsocket {
 
   constructor(options: AppserviceWebsocketOptions) {
     this.#appservice = options.appservice;
-    this.#dispatch = options.dispatch;
     this.#handleProxy = options.handleHTTPProxy;
     this.#handleTransaction = options.handleTransaction;
     this.#log = options.log;
@@ -201,7 +198,19 @@ export class AppserviceWebsocket {
   }
 
   async #handleMessage(data: WebSocket.RawData): Promise<void> {
-    const message = JSON.parse(data.toString()) as WebsocketMessage;
+    const raw = data.toString();
+    if (!raw.trim()) {
+      this.#log("warn", "appservice_websocket_empty_message");
+      return;
+    }
+    let message: WebsocketMessage;
+    try {
+      message = JSON.parse(raw) as WebsocketMessage;
+    } catch (error: unknown) {
+      const messageText = error instanceof Error ? error.message : String(error);
+      this.#log("error", "appservice_websocket_invalid_json", { error: messageText, size: raw.length });
+      return;
+    }
     this.#log("debug", "appservice_websocket_message", {
       command: message.command ?? "transaction",
       eventCount: message.events?.length,
@@ -220,16 +229,6 @@ export class AppserviceWebsocket {
       if (message.command === "response" || message.command === "error") return;
       if (!message.command || message.command === "transaction") {
         await this.#handleTransaction?.(message as Record<string, unknown>);
-        for (const raw of message.events ?? []) {
-          const event = rawMatrixEvent(raw);
-          this.#log("debug", "appservice_websocket_transaction_event", {
-            eventId: raw.event_id,
-            roomId: raw.room_id,
-            sender: raw.sender,
-            type: raw.type,
-          });
-          if (event) await this.#dispatch(event);
-        }
         this.#send(messageResponse(message, true, { txn_id: message.txn_id }));
         return;
       }
@@ -270,10 +269,6 @@ export class AppserviceWebsocket {
         txnId: transactionMatch[1],
       });
       await this.#handleTransaction?.(transaction);
-      for (const raw of events) {
-        const event = rawMatrixEvent(raw as RawMatrixEvent);
-        if (event) await this.#dispatch(event);
-      }
       return jsonHTTPResponse(200, {});
     }
     if (method === "GET" && /^\/?_matrix\/app\/v1\/users\//.test(path)) {
@@ -324,7 +319,7 @@ interface WebsocketRequest {
 interface WebsocketMessage {
   command?: string;
   data?: unknown;
-  events?: RawMatrixEvent[];
+  events?: unknown[];
   id?: number;
   status?: string;
   to_device?: unknown;
@@ -346,19 +341,6 @@ export interface HTTPProxyResponse {
   status: number;
 }
 
-interface RawMatrixEvent {
-  [key: string]: unknown;
-  content?: Record<string, unknown>;
-  event_id?: string;
-  origin_server_ts?: number;
-  redacts?: string;
-  room_id?: string;
-  sender?: string;
-  state_key?: string;
-  type?: string;
-  unsigned?: Record<string, unknown>;
-}
-
 function messageResponse(message: WebsocketMessage, ok: boolean, data: unknown): WebsocketRequest | null {
   if (message.id === undefined || message.id === null || message.command === "response" || message.command === "error") return null;
   return {
@@ -400,86 +382,10 @@ function eventCount(events: unknown): number | undefined {
   return Array.isArray(events) && events.length > 0 ? events.length : undefined;
 }
 
-function rawMatrixEvent(raw: RawMatrixEvent): MatrixClientEvent | null {
-  const type = raw.type ?? "";
-  const content = raw.content ?? {};
-  const roomId = raw.room_id;
-  const eventId = raw.event_id;
-  const senderId = raw.sender;
-  const sender = senderId ? { isMe: false, userId: senderId } : undefined;
-  if (type === "m.room.message" && roomId && eventId && sender) {
-    return stripUndefined({
-      attachments: [],
-      class: "message",
-      content,
-      edited: false,
-      encrypted: false,
-      eventId,
-      kind: "message",
-      messageType: stringValue(content.msgtype) ?? "m.text",
-      raw,
-      roomId,
-      sender,
-      text: stringValue(content.body) ?? "",
-      timestamp: raw.origin_server_ts,
-      type,
-      unsigned: raw.unsigned,
-    }) as MatrixClientEvent;
-  }
-  if (type === "m.reaction" && roomId && eventId && sender) {
-    const relates = objectValue(content["m.relates_to"]);
-    return stripUndefined({
-      added: true,
-      class: "message",
-      content,
-      eventId,
-      key: stringValue(relates?.key) ?? "",
-      kind: "reaction",
-      raw,
-      relatesTo: stringValue(relates?.event_id) ?? "",
-      roomId,
-      sender,
-      timestamp: raw.origin_server_ts,
-      type,
-      unsigned: raw.unsigned,
-    }) as MatrixClientEvent;
-  }
-  if (type === "m.room.redaction" && roomId) {
-    return genericEvent("redaction", raw, content);
-  }
-  if (type === "m.typing") {
-    return genericEvent("typing", raw, content);
-  }
-  return genericEvent("raw", raw, content);
-}
-
-function genericEvent(kind: "raw" | "redaction" | "typing", raw: RawMatrixEvent, content: Record<string, unknown>): MatrixClientEvent {
-  const event = {
-    class: kind === "typing" ? "ephemeral" : "unknown",
-    content,
-    eventId: raw.event_id,
-    kind,
-    raw,
-    roomId: raw.room_id,
-    sender: raw.sender ? { isMe: false, userId: raw.sender } : undefined,
-    timestamp: raw.origin_server_ts,
-    type: raw.type ?? "",
-    unsigned: raw.unsigned,
-  };
-  return stripUndefined(event) as MatrixClientEvent;
-}
-
 function objectValue(value: unknown): Record<string, unknown> | undefined {
   return value && typeof value === "object" ? value as Record<string, unknown> : undefined;
 }
 
 function stringValue(value: unknown): string | undefined {
   return typeof value === "string" ? value : undefined;
 }
-
-function stripUndefined<T extends Record<string, unknown>>(value: T): T {
-  for (const key of Object.keys(value)) {
-    if (value[key] === undefined) delete value[key];
-  }
-  return value;
-}

packages/bridge/src/beeper.test.ts

@@ -50,7 +50,7 @@ describe("Beeper bridge manager helpers", () => {
       }
       expect(String(url)).toBe("https://api.example/bridgebox/alice/bridge/sh-dummy/bridge_state");
       expect(init?.method).toBe("POST");
-      expect(init?.headers).toMatchObject({ authorization: "Bearer token" });
+      expect(init?.headers).toMatchObject({ authorization: "Bearer as" });
       expect(JSON.parse(String(init?.body))).toEqual({
         info: {},
         isSelfHosted: true,
@@ -110,6 +110,31 @@ describe("Beeper bridge manager helpers", () => {
       id: "sh-dummy",
     });
   });
+
+  it("refuses to post bridge state without an appservice token", async () => {
+    const fetch = vi.fn(async (url: URL) => {
+      if (String(url) === "https://api.example/whoami") {
+        return jsonResponse({
+          user: { bridges: {} },
+          userInfo: { username: "alice" },
+        });
+      }
+      return jsonResponse({
+        hs_token: "hs",
+        id: "sh-dummy",
+        namespaces: { user_ids: [{ exclusive: true, regex: "@dummy_.*:beeper.local" }] },
+        sender_localpart: "dummybot",
+        url: "websocket",
+      });
+    });
+
+    await expect(createBeeperAppServiceInit({
+      baseDomain: "example",
+      bridge: "sh-dummy",
+      fetch: fetch as never,
+      token: "token",
+    })).rejects.toThrow("missing as_token");
+  });
 });
 
 function jsonResponse(data: unknown): Response {

packages/bridge/src/beeper.ts

@@ -112,19 +112,22 @@ export class BeeperBridgeManagerClient {
       self_hosted: options.selfHosted ?? true,
     }));
     if (options.postState !== false) {
+      if (!registration.asToken) {
+        throw new Error(`Beeper appservice registration for ${options.bridge} did not include an appservice token`);
+      }
       const stateOptions: PostBridgeStateOptions = {
         bridge: options.bridge,
         isSelfHosted: options.selfHosted ?? true,
         reason: "SELF_HOST_REGISTERED",
         stateEvent: bridgeStateEvent(options),
       };
       if (options.bridgeType !== undefined) stateOptions.bridgeType = options.bridgeType;
-      await this.postBridgeState(stateOptions);
+      await this.postBridgeState(stateOptions, registration.asToken);
     }
     return registration;
   }
 
-  async postBridgeState(options: PostBridgeStateOptions): Promise<void> {
+  async postBridgeState(options: PostBridgeStateOptions, token?: string): Promise<void> {
     const whoami = await this.whoami();
     const username = this.#username ?? whoami.userInfo.username;
     await this.#request("api", "POST", `/bridgebox/${encodeURIComponent(username)}/bridge/${encodeURIComponent(options.bridge)}/bridge_state`, {
@@ -133,7 +136,7 @@ export class BeeperBridgeManagerClient {
       isSelfHosted: options.isSelfHosted ?? true,
       reason: options.reason,
       stateEvent: options.stateEvent,
-    });
+    }, undefined, token);
   }
 
   async createAppService(options: CreateAppServiceOptions): Promise<RegisteredAppService> {