fix: disambiguate gateway event dedupe across reconnects

src/lib/controlplane/contracts.ts

@@ -16,6 +16,7 @@ export type ControlPlaneDomainEvent =
       type: "gateway.event";
       event: string;
       seq: number | null;
+      connectionEpoch?: string | null;
       payload: unknown;
       asOf: string;
     };

src/lib/controlplane/openclaw-adapter.ts

@@ -1,3 +1,5 @@
+import { randomUUID } from "node:crypto";
+
 import { WebSocket } from "ws";
 
 import type {
@@ -14,21 +16,22 @@ const REQUEST_TIMEOUT_MS = 15_000;
 const INITIAL_RECONNECT_DELAY_MS = 1_000;
 const MAX_RECONNECT_DELAY_MS = 15_000;
 const CONNECT_PROTOCOL = 3;
+const CONNECT_CLIENT_ID = "gateway-client";
+const CONNECT_CLIENT_MODE = "backend";
 
 const DEFAULT_METHOD_ALLOWLIST = new Set<string>([
   "status",
   "chat.send",
   "chat.abort",
+  "agents.create",
   "agents.update",
   "agents.delete",
   "agents.list",
-  "agents.create",
   "sessions.list",
   "sessions.preview",
   "sessions.patch",
   "sessions.reset",
   "config.get",
-  "config.patch",
   "config.set",
   "exec.approval.resolve",
   "exec.approvals.get",
@@ -100,6 +103,7 @@ export class OpenClawGatewayAdapter {
   private reconnectAttempt = 0;
   private stopping = false;
   private nextRequestNumber = 1;
+  private connectionEpoch: string | null = null;
   private pending = new Map<string, PendingRequest>();
   private loadSettings: () => ControlPlaneGatewaySettings;
   private createWebSocket: (url: string, opts: { origin: string }) => WebSocket;
@@ -145,6 +149,7 @@ export class OpenClawGatewayAdapter {
     const ws = this.ws;
     this.ws = null;
     this.connectRequestId = null;
+    this.connectionEpoch = null;
     if (ws && ws.readyState === WebSocket.OPEN) {
       await new Promise<void>((resolve) => {
         ws.once("close", () => resolve());
@@ -194,6 +199,7 @@ export class OpenClawGatewayAdapter {
 
   private async connect(): Promise<void> {
     const settings = this.loadSettings();
+    this.connectionEpoch = randomUUID();
     const ws = this.createWebSocket(settings.url, { origin: resolveOriginForUpstream(settings.url) });
     this.ws = ws;
     this.connectRequestId = null;
@@ -230,6 +236,7 @@ export class OpenClawGatewayAdapter {
             type: "gateway.event",
             event: parsed.event,
             seq: typeof parsed.seq === "number" ? parsed.seq : null,
+            connectionEpoch: this.connectionEpoch,
             payload: parsed.payload,
             asOf: new Date().toISOString(),
           });
@@ -258,6 +265,8 @@ export class OpenClawGatewayAdapter {
           settle(() => reject(new Error("Control-plane gateway connection closed during connect.")));
           return;
         }
+        this.rejectPending("Control-plane gateway connection closed.");
+        this.connectionEpoch = null;
         this.updateStatus("reconnecting", "gateway_closed");
         this.scheduleReconnect();
       });
@@ -269,6 +278,7 @@ export class OpenClawGatewayAdapter {
         }
       });
     }).catch((err) => {
+      this.connectionEpoch = null;
       this.updateStatus("error", err instanceof Error ? err.message : "connect_error");
       this.scheduleReconnect();
       throw err;
@@ -294,27 +304,38 @@ export class OpenClawGatewayAdapter {
     if (!ws || ws.readyState !== WebSocket.OPEN || this.connectRequestId) return;
     const id = String(this.nextRequestNumber++);
     this.connectRequestId = id;
-    ws.send(
-      JSON.stringify({
-        type: "req",
-        id,
-        method: "connect",
-        params: {
-          minProtocol: CONNECT_PROTOCOL,
-          maxProtocol: CONNECT_PROTOCOL,
-          client: {
-            id: "openclaw-studio-controlplane",
-            version: "dev",
-            platform: "node",
-            mode: "operator",
+    try {
+      ws.send(
+        JSON.stringify({
+          type: "req",
+          id,
+          method: "connect",
+          params: {
+            minProtocol: CONNECT_PROTOCOL,
+            maxProtocol: CONNECT_PROTOCOL,
+            client: {
+              id: CONNECT_CLIENT_ID,
+              version: "dev",
+              platform: "node",
+              mode: CONNECT_CLIENT_MODE,
+            },
+            role: "operator",
+            scopes: ["operator.admin", "operator.approvals", "operator.pairing"],
+            caps: [],
+            auth: { token },
           },
-          role: "operator",
-          scopes: ["operator.admin", "operator.approvals", "operator.pairing"],
-          caps: [],
-          auth: { token },
-        },
-      })
-    );
+        })
+      );
+    } catch (err) {
+      this.connectRequestId = null;
+      const reason = err instanceof Error ? err.message : "connect_send_failed";
+      this.updateStatus("error", reason);
+      try {
+        ws.close(1011, "connect send failed");
+      } catch (closeErr) {
+        console.error("Failed to close gateway socket after connect-send failure.", closeErr);
+      }
+    }
   }
 
   private parseFrame(raw: string): GatewayEventFrame | GatewayResponseFrame | null {

src/lib/controlplane/outbox.ts

@@ -6,8 +6,14 @@ export const deriveControlPlaneEventKey = (event: ControlPlaneDomainEvent): stri
   if (event.type === "runtime.status") {
     return ["runtime.status", event.status, safeString(event.reason), event.asOf].join(":");
   }
+  const connectionEpoch = safeString(event.connectionEpoch).trim();
   if (typeof event.seq === "number" && Number.isFinite(event.seq)) {
-    return ["gateway.event", event.event, "seq", String(event.seq)].join(":");
+    if (connectionEpoch) {
+      return ["gateway.event", event.event, "epoch", connectionEpoch, "seq", String(event.seq)].join(
+        ":"
+      );
+    }
+    return ["gateway.event", event.event, "seq", String(event.seq), safeString(event.asOf)].join(":");
   }
-  return ["gateway.event", event.event, "", event.asOf].join(":");
+  return ["gateway.event", event.event, connectionEpoch, event.asOf].join(":");
 };

tests/unit/controlPlaneProjectionStore.test.ts

@@ -4,6 +4,7 @@ import fs from "node:fs";
 import os from "node:os";
 import path from "node:path";
 
+import Database from "better-sqlite3";
 import { afterEach, describe, expect, it } from "vitest";
 
 import { SQLiteControlPlaneProjectionStore } from "@/lib/controlplane/projection-store";
@@ -48,6 +49,7 @@ describe("SQLiteControlPlaneProjectionStore", () => {
       type: "gateway.event" as const,
       event: "runtime.delta",
       seq: 42,
+      connectionEpoch: "conn-1",
       payload: { content: "a" },
       asOf: "2026-02-28T02:01:00.000Z",
     };
@@ -55,6 +57,7 @@ describe("SQLiteControlPlaneProjectionStore", () => {
       type: "gateway.event" as const,
       event: "runtime.final",
       seq: 43,
+      connectionEpoch: "conn-1",
       payload: { content: "b" },
       asOf: "2026-02-28T02:01:02.000Z",
     };
@@ -79,4 +82,209 @@ describe("SQLiteControlPlaneProjectionStore", () => {
 
     store.close();
   });
+
+  it("stores same event+seq as distinct rows when connection epoch changes", () => {
+    const store = new SQLiteControlPlaneProjectionStore(makeDbPath());
+    const first = store.applyDomainEvent({
+      type: "gateway.event",
+      event: "agent",
+      seq: 1,
+      connectionEpoch: "conn-1",
+      payload: { state: "running" },
+      asOf: "2026-02-28T02:01:00.000Z",
+    });
+    const second = store.applyDomainEvent({
+      type: "gateway.event",
+      event: "agent",
+      seq: 1,
+      connectionEpoch: "conn-2",
+      payload: { state: "idle" },
+      asOf: "2026-02-28T02:01:10.000Z",
+    });
+
+    expect(first.id).toBe(1);
+    expect(second.id).toBe(2);
+
+    const replay = store.readOutboxAfter(0, 10);
+    expect(replay.map((entry) => entry.id)).toEqual([1, 2]);
+
+    store.close();
+  });
+
+  it("reads outbox pages before a cursor in ascending order", () => {
+    const store = new SQLiteControlPlaneProjectionStore(makeDbPath());
+    for (let index = 1; index <= 5; index += 1) {
+      store.applyDomainEvent({
+        type: "gateway.event",
+        event: "runtime.delta",
+        seq: index,
+        payload: { index },
+        asOf: `2026-02-28T02:01:0${index}.000Z`,
+      });
+    }
+
+    const newestTwo = store.readOutboxBefore(6, 2);
+    expect(newestTwo.map((entry) => entry.id)).toEqual([4, 5]);
+
+    const olderTwo = store.readOutboxBefore(4, 2);
+    expect(olderTwo.map((entry) => entry.id)).toEqual([2, 3]);
+
+    store.close();
+  });
+
+  it("reads agent outbox pages before a cursor in ascending order and normalizes casing", () => {
+    const store = new SQLiteControlPlaneProjectionStore(makeDbPath());
+    store.applyDomainEvent({
+      type: "gateway.event",
+      event: "runtime.delta",
+      seq: 1,
+      payload: { sessionKey: "Agent:Alpha:Main", text: "a" },
+      asOf: "2026-02-28T02:01:01.000Z",
+    });
+    store.applyDomainEvent({
+      type: "gateway.event",
+      event: "runtime.delta",
+      seq: 2,
+      payload: { sessionKey: "agent:beta:main", text: "b" },
+      asOf: "2026-02-28T02:01:02.000Z",
+    });
+    store.applyDomainEvent({
+      type: "gateway.event",
+      event: "runtime.delta",
+      seq: 3,
+      payload: { agentId: "ALPHA", text: "c" },
+      asOf: "2026-02-28T02:01:03.000Z",
+    });
+    store.applyDomainEvent({
+      type: "gateway.event",
+      event: "runtime.delta",
+      seq: 4,
+      payload: { sessionKey: "agent:alpha:main", text: "d" },
+      asOf: "2026-02-28T02:01:04.000Z",
+    });
+
+    const newestTwo = store.readAgentOutboxBefore("ALPHA", 5, 2);
+    expect(newestTwo.map((entry) => entry.id)).toEqual([3, 4]);
+
+    const olderOne = store.readAgentOutboxBefore("alpha", 3, 10);
+    expect(olderOne.map((entry) => entry.id)).toEqual([1]);
+
+    store.close();
+  });
+
+  it("backfills legacy outbox rows into agent index and marks non-agent rows", () => {
+    const dbPath = makeDbPath();
+    const db = new Database(dbPath);
+    db.exec(`
+      CREATE TABLE runtime_projection (
+        id INTEGER PRIMARY KEY CHECK (id = 1),
+        status TEXT NOT NULL,
+        reason TEXT,
+        as_of TEXT,
+        updated_at TEXT NOT NULL
+      );
+      CREATE TABLE outbox (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        event_type TEXT NOT NULL,
+        event_json TEXT NOT NULL,
+        created_at TEXT NOT NULL
+      );
+      CREATE TABLE processed_events (
+        event_key TEXT PRIMARY KEY,
+        outbox_id INTEGER,
+        created_at TEXT NOT NULL,
+        FOREIGN KEY (outbox_id) REFERENCES outbox(id) ON DELETE SET NULL
+      );
+      CREATE INDEX idx_outbox_id ON outbox(id);
+    `);
+    db.prepare("INSERT INTO outbox (event_type, event_json, created_at) VALUES (?, ?, ?)").run(
+      "gateway.event",
+      JSON.stringify({
+        type: "gateway.event",
+        event: "runtime.delta",
+        seq: 10,
+        payload: { sessionKey: "Agent:Alpha:Main" },
+        asOf: "2026-02-28T02:01:01.000Z",
+      }),
+      "2026-02-28T02:01:01.000Z"
+    );
+    db.prepare("INSERT INTO outbox (event_type, event_json, created_at) VALUES (?, ?, ?)").run(
+      "runtime.status",
+      JSON.stringify({
+        type: "runtime.status",
+        status: "connected",
+        reason: null,
+        asOf: "2026-02-28T02:01:02.000Z",
+      }),
+      "2026-02-28T02:01:02.000Z"
+    );
+    db.prepare("INSERT INTO outbox (event_type, event_json, created_at) VALUES (?, ?, ?)").run(
+      "gateway.event",
+      JSON.stringify({
+        type: "gateway.event",
+        event: "runtime.delta",
+        seq: 11,
+        payload: { sessionKey: "agent:beta:main" },
+        asOf: "2026-02-28T02:01:03.000Z",
+      }),
+      "2026-02-28T02:01:03.000Z"
+    );
+    db.pragma("user_version = 1");
+    db.close();
+
+    const store = new SQLiteControlPlaneProjectionStore(dbPath);
+    const firstBackfill = store.backfillAgentOutboxBefore(4, 10);
+    expect(firstBackfill.scannedRows).toBe(3);
+    expect(firstBackfill.updatedRows).toBe(3);
+
+    const alphaRows = store.readAgentOutboxBefore("alpha", 4, 10);
+    expect(alphaRows.map((entry) => entry.id)).toEqual([1]);
+
+    const betaRows = store.readAgentOutboxBefore("beta", 4, 10);
+    expect(betaRows.map((entry) => entry.id)).toEqual([3]);
+
+    const secondBackfill = store.backfillAgentOutboxBefore(4, 10);
+    expect(secondBackfill.scannedRows).toBe(0);
+    expect(secondBackfill.exhausted).toBe(true);
+
+    store.close();
+
+    const verifyDb = new Database(dbPath, { readonly: true });
+    const runtimeStatusRow = verifyDb
+      .prepare("SELECT agent_id FROM outbox WHERE id = 2")
+      .get() as { agent_id: string | null };
+    expect(runtimeStatusRow.agent_id).toBe("");
+    verifyDb.close();
+  });
+
+  it("repairs missing agent index on dedupe replay", () => {
+    const dbPath = makeDbPath();
+    const event = {
+      type: "gateway.event" as const,
+      event: "runtime.delta",
+      seq: 99,
+      payload: { sessionKey: "agent:alpha:main", text: "dedupe" },
+      asOf: "2026-02-28T02:01:00.000Z",
+    };
+
+    const firstStore = new SQLiteControlPlaneProjectionStore(dbPath);
+    firstStore.applyDomainEvent(event);
+    firstStore.close();
+
+    const mutateDb = new Database(dbPath);
+    mutateDb.prepare("UPDATE outbox SET agent_id = NULL WHERE id = 1").run();
+    mutateDb.close();
+
+    const secondStore = new SQLiteControlPlaneProjectionStore(dbPath);
+    const deduped = secondStore.applyDomainEvent(event);
+    expect(deduped.id).toBe(1);
+    secondStore.close();
+
+    const verifyDb = new Database(dbPath, { readonly: true });
+    const repaired = verifyDb.prepare("SELECT agent_id FROM outbox WHERE id = 1").get() as {
+      agent_id: string | null;
+    };
+    expect(repaired.agent_id).toBe("alpha");
+    verifyDb.close();
+  });
 });

tests/unit/openclawAdapter.test.ts

@@ -0,0 +1,218 @@
+// @vitest-environment node
+
+import { afterEach, describe, expect, it } from "vitest";
+import { EventEmitter } from "node:events";
+import { WebSocket, WebSocketServer } from "ws";
+
+import { OpenClawGatewayAdapter } from "@/lib/controlplane/openclaw-adapter";
+import type { ControlPlaneDomainEvent } from "@/lib/controlplane/contracts";
+
+const closeWebSocketServer = (server: WebSocketServer) =>
+  new Promise<void>((resolve) => server.close(() => resolve()));
+
+const waitForCondition = async (predicate: () => boolean, timeoutMs: number = 3_000) => {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (predicate()) return;
+    await new Promise((resolve) => setTimeout(resolve, 10));
+  }
+  throw new Error("Condition not met before timeout.");
+};
+
+describe("OpenClawGatewayAdapter", () => {
+  let upstream: WebSocketServer | null = null;
+
+  afterEach(async () => {
+    if (upstream) {
+      await closeWebSocketServer(upstream);
+      upstream = null;
+    }
+  });
+
+  it("rejects in-flight requests immediately when the socket closes", async () => {
+    upstream = new WebSocketServer({ port: 0 });
+    const address = upstream.address();
+    if (!address || typeof address === "string") {
+      throw new Error("expected upstream server to provide a numeric port");
+    }
+    const upstreamUrl = `ws://127.0.0.1:${address.port}`;
+    let observedConnectClientId: string | null = null;
+    let observedConnectClientMode: string | null = null;
+
+    upstream.on("connection", (ws) => {
+      ws.send(JSON.stringify({ type: "event", event: "connect.challenge", payload: {} }));
+      ws.on("message", (raw) => {
+        const parsed = JSON.parse(String(raw ?? "")) as {
+          id?: string;
+          method?: string;
+          params?: {
+            client?: { id?: string; mode?: string };
+          };
+        };
+        if (parsed?.method === "connect") {
+          observedConnectClientId = parsed.params?.client?.id ?? null;
+          observedConnectClientMode = parsed.params?.client?.mode ?? null;
+          ws.send(
+            JSON.stringify({
+              type: "res",
+              id: parsed.id,
+              ok: true,
+              payload: { type: "hello-ok", protocol: 3 },
+            })
+          );
+          return;
+        }
+        if (parsed?.method === "status") {
+          ws.close(1011, "upstream closed");
+        }
+      });
+    });
+
+    const adapter = new OpenClawGatewayAdapter({
+      loadSettings: () => ({ url: upstreamUrl, token: "tkn" }),
+    });
+
+    await adapter.start();
+    const startedAt = Date.now();
+    await expect(adapter.request("status", {})).rejects.toThrow(
+      "Control-plane gateway connection closed."
+    );
+    expect(Date.now() - startedAt).toBeLessThan(2_000);
+    expect(observedConnectClientId).toBe("gateway-client");
+    expect(observedConnectClientMode).toBe("backend");
+
+    await adapter.stop();
+  });
+
+  it("fails connect gracefully when sending the connect request throws", async () => {
+    class ThrowingConnectSocket extends EventEmitter {
+      readyState: number = WebSocket.OPEN;
+
+      close() {
+        if (this.readyState === WebSocket.CLOSED) return;
+        this.readyState = WebSocket.CLOSED;
+        this.emit("close");
+      }
+
+      terminate() {
+        this.close();
+      }
+
+      send(raw: string) {
+        const parsed = JSON.parse(raw) as { method?: string };
+        if (parsed.method === "connect") {
+          throw new Error("connect send failed");
+        }
+      }
+    }
+
+    const socket = new ThrowingConnectSocket();
+    const adapter = new OpenClawGatewayAdapter({
+      loadSettings: () => ({ url: "ws://127.0.0.1:9", token: "tkn" }),
+      createWebSocket: () => socket as unknown as WebSocket,
+    });
+
+    setTimeout(() => {
+      socket.emit("message", JSON.stringify({ type: "event", event: "connect.challenge", payload: {} }));
+    }, 0);
+
+    await expect(adapter.start()).rejects.toThrow(
+      "Control-plane gateway connection closed during connect."
+    );
+
+    await adapter.stop();
+  });
+
+  it("emits gateway events with unique connection epochs across reconnect cycles", async () => {
+    upstream = new WebSocketServer({ port: 0 });
+    const address = upstream.address();
+    if (!address || typeof address === "string") {
+      throw new Error("expected upstream server to provide a numeric port");
+    }
+    const upstreamUrl = `ws://127.0.0.1:${address.port}`;
+    let acceptedConnections = 0;
+
+    upstream.on("connection", (ws) => {
+      acceptedConnections += 1;
+      const connectionIndex = acceptedConnections;
+      ws.send(JSON.stringify({ type: "event", event: "connect.challenge", payload: {} }));
+      ws.on("message", (raw) => {
+        const parsed = JSON.parse(String(raw ?? "")) as { id?: string; method?: string };
+        if (parsed?.method !== "connect" || !parsed.id) return;
+        ws.send(
+          JSON.stringify({
+            type: "res",
+            id: parsed.id,
+            ok: true,
+            payload: { type: "hello-ok", protocol: 3 },
+          })
+        );
+        ws.send(
+          JSON.stringify({
+            type: "event",
+            event: "agent",
+            seq: 1,
+            payload: { connectionIndex },
+          })
+        );
+      });
+    });
+
+    const observedEvents: ControlPlaneDomainEvent[] = [];
+    const adapter = new OpenClawGatewayAdapter({
+      loadSettings: () => ({ url: upstreamUrl, token: "tkn" }),
+      onDomainEvent: (event) => {
+        observedEvents.push(event);
+      },
+    });
+
+    await adapter.start();
+    await waitForCondition(() =>
+      observedEvents.some(
+        (event) =>
+          event.type === "gateway.event" &&
+          event.event === "agent" &&
+          typeof (event.payload as { connectionIndex?: unknown })?.connectionIndex === "number" &&
+          (event.payload as { connectionIndex?: number }).connectionIndex === 1
+      )
+    );
+
+    await adapter.stop();
+
+    await adapter.start();
+    await waitForCondition(() =>
+      observedEvents.some(
+        (event) =>
+          event.type === "gateway.event" &&
+          event.event === "agent" &&
+          typeof (event.payload as { connectionIndex?: unknown })?.connectionIndex === "number" &&
+          (event.payload as { connectionIndex?: number }).connectionIndex === 2
+      )
+    );
+
+    const firstGatewayEvent = observedEvents.find(
+      (event) =>
+        event.type === "gateway.event" &&
+        (event.payload as { connectionIndex?: number })?.connectionIndex === 1
+    );
+    const secondGatewayEvent = observedEvents.find(
+      (event) =>
+        event.type === "gateway.event" &&
+        (event.payload as { connectionIndex?: number })?.connectionIndex === 2
+    );
+
+    expect(firstGatewayEvent?.type).toBe("gateway.event");
+    expect(secondGatewayEvent?.type).toBe("gateway.event");
+    if (!firstGatewayEvent || firstGatewayEvent.type !== "gateway.event") {
+      throw new Error("Expected first gateway event.");
+    }
+    if (!secondGatewayEvent || secondGatewayEvent.type !== "gateway.event") {
+      throw new Error("Expected second gateway event.");
+    }
+    expect(firstGatewayEvent.connectionEpoch).toBeTruthy();
+    expect(secondGatewayEvent.connectionEpoch).toBeTruthy();
+    expect(firstGatewayEvent.connectionEpoch).not.toBe(secondGatewayEvent.connectionEpoch);
+
+    await adapter.stop();
+  });
+});