fix: subscribe ClickClack from startup tail cursor

extensions/clickclack/src/gateway.test.ts

@@ -20,6 +20,7 @@ const mocks = vi.hoisted(() => ({
   client: {
     me: vi.fn(),
     events: vi.fn(),
+    eventPage: vi.fn(),
     websocket: vi.fn(),
     channelMessages: vi.fn(),
     directMessages: vi.fn(),
@@ -79,6 +80,19 @@ function createGatewayContext(
   };
 }
 
+function createBacklogEvent(index: number, type = "channel.updated") {
+  return {
+    id: `evt-${index}`,
+    cursor: `cursor-${index}`,
+    type,
+    workspace_id: "workspace-1",
+    channel_id: "chan-1",
+    seq: index,
+    created_at: "2026-01-01T00:00:00.000Z",
+    payload: type === "message.created" ? { message_id: "msg-1", author_id: "human-1" } : undefined,
+  };
+}
+
 describe("ClickClack gateway", () => {
   beforeEach(() => {
     vi.clearAllMocks();
@@ -90,7 +104,7 @@ describe("ClickClack gateway", () => {
       avatar_url: "",
       created_at: "2026-01-01T00:00:00.000Z",
     });
-    mocks.client.events.mockResolvedValue([]);
+    mocks.client.eventPage.mockResolvedValue({ events: [] });
     mocks.resolveClickClackInboundAccess.mockResolvedValue({
       shouldDispatch: true,
       commandAuthorized: true,
@@ -118,6 +132,65 @@ describe("ClickClack gateway", () => {
     ]);
   });
 
+  it("opens realtime from the server startup tail without dispatching the returned page", async () => {
+    mocks.client.eventPage.mockResolvedValueOnce({
+      events: [createBacklogEvent(1)],
+      tailCursor: "cursor-501",
+    });
+    const socket = new FakeSocket();
+    mocks.client.websocket.mockReturnValue(socket);
+    const abort = new AbortController();
+    const ctx = createGatewayContext(abort.signal);
+    const run = startClickClackGatewayAccount(ctx);
+
+    await vi.waitFor(() => expect(mocks.client.websocket).toHaveBeenCalledTimes(1));
+
+    expect(mocks.client.eventPage).toHaveBeenCalledWith("workspace-1", { includeTail: true });
+    expect(mocks.client.websocket).toHaveBeenCalledWith("workspace-1", "cursor-501");
+    expect(mocks.handleClickClackInbound).not.toHaveBeenCalled();
+
+    abort.abort();
+    await run;
+  });
+
+  it("drains and processes every reconnect page before reopening realtime", async () => {
+    const firstSocket = new FakeSocket();
+    const secondSocket = new FakeSocket();
+    const firstReconnectPage = Array.from({ length: 500 }, (_, index) =>
+      createBacklogEvent(index + 1),
+    );
+    mocks.client.eventPage
+      .mockResolvedValueOnce({ events: [], tailCursor: "" })
+      .mockResolvedValueOnce({ events: firstReconnectPage })
+      .mockResolvedValueOnce({ events: [createBacklogEvent(501, "message.created")] });
+    mocks.client.websocket.mockReturnValueOnce(firstSocket).mockReturnValueOnce(secondSocket);
+    const abort = new AbortController();
+    const ctx = createGatewayContext(abort.signal);
+    const run = startClickClackGatewayAccount(ctx);
+
+    await vi.waitFor(() => expect(mocks.client.websocket).toHaveBeenCalledTimes(1));
+    firstSocket.emit("close");
+    await vi.waitFor(() => expect(mocks.client.websocket).toHaveBeenCalledTimes(2));
+
+    expect(mocks.client.eventPage).toHaveBeenNthCalledWith(2, "workspace-1", {
+      afterCursor: "",
+      limit: 500,
+    });
+    expect(mocks.client.eventPage).toHaveBeenNthCalledWith(3, "workspace-1", {
+      afterCursor: "cursor-500",
+      limit: 500,
+    });
+    expect(mocks.client.eventPage).toHaveBeenNthCalledWith(4, "workspace-1", {
+      afterCursor: "cursor-501",
+      limit: 500,
+    });
+    expect(mocks.handleClickClackInbound).toHaveBeenCalledTimes(1);
+    expect(mocks.client.websocket).toHaveBeenLastCalledWith("workspace-1", "cursor-501");
+
+    abort.abort();
+    await run;
+  });
+
   it("skips malformed websocket frames without stopping the monitor", async () => {
     const socket = new FakeSocket();
     mocks.client.websocket.mockReturnValue(socket);
@@ -324,7 +397,7 @@ describe("ClickClack gateway", () => {
   });
 
   it("clears running status when backlog polling fails", async () => {
-    mocks.client.events.mockRejectedValue(new Error("clickclack unavailable"));
+    mocks.client.eventPage.mockRejectedValue(new Error("clickclack unavailable"));
     const abort = new AbortController();
     const ctx = createGatewayContext(abort.signal);
 

extensions/clickclack/src/gateway.ts

@@ -17,6 +17,8 @@ import type {
   ResolvedClickClackAccount,
 } from "./types.js";
 
+const CLICKCLACK_EVENT_PAGE_LIMIT = 500;
+
 function payloadString(event: ClickClackEvent, key: string): string {
   const value = event.payload?.[key];
   return typeof value === "string" ? value : "";
@@ -132,6 +134,37 @@ async function processEvent(params: {
   });
 }
 
+async function drainEventBacklog(params: {
+  client: ReturnType<typeof createClickClackClient>;
+  workspaceId: string;
+  afterCursor: string;
+  abortSignal: AbortSignal;
+  onEvent: (event: ClickClackEvent) => Promise<void>;
+}): Promise<string> {
+  let afterCursor = params.afterCursor;
+  while (!params.abortSignal.aborted) {
+    const page = await params.client.eventPage(params.workspaceId, {
+      afterCursor,
+      limit: CLICKCLACK_EVENT_PAGE_LIMIT,
+    });
+    const events = page.events;
+    for (const event of events) {
+      if (params.abortSignal.aborted) {
+        return afterCursor;
+      }
+      if (!event.cursor || event.cursor === afterCursor) {
+        throw new Error("ClickClack event backlog returned a non-advancing cursor");
+      }
+      await params.onEvent(event);
+      afterCursor = event.cursor;
+    }
+    if (events.length === 0) {
+      return afterCursor;
+    }
+  }
+  return afterCursor;
+}
+
 export async function startClickClackGatewayAccount(
   ctx: ChannelGatewayContext<ResolvedClickClackAccount>,
 ) {
@@ -164,25 +197,39 @@ export async function startClickClackGatewayAccount(
   let initialized = false;
   try {
     while (!ctx.abortSignal.aborted) {
-      const backlog = await client.events(workspaceId, afterCursor);
       if (!initialized) {
-        // First pass establishes the cursor without replaying historical backlog
-        // into fresh gateway sessions.
-        for (const event of backlog) {
-          afterCursor = event.cursor || afterCursor;
+        const page = await client.eventPage(workspaceId, { includeTail: true });
+        // Newer servers capture this cursor before listing the page, so events
+        // created during startup remain eligible for websocket delivery.
+        if (page.tailCursor !== undefined) {
+          afterCursor = page.tailCursor;
+        } else {
+          // Older servers omit tail_cursor; preserve the shipped one-page
+          // startup behavior instead of extending the history-skip window.
+          for (const event of page.events) {
+            afterCursor = event.cursor || afterCursor;
+          }
         }
         initialized = true;
       } else {
-        for (const event of backlog) {
-          afterCursor = event.cursor || afterCursor;
-          await processEvent({
-            account,
-            config: ctx.cfg,
-            client,
-            event,
-            botUserId: account.botUserId,
-          });
-        }
+        afterCursor = await drainEventBacklog({
+          client,
+          workspaceId,
+          afterCursor,
+          abortSignal: ctx.abortSignal,
+          onEvent: async (event) => {
+            await processEvent({
+              account,
+              config: ctx.cfg,
+              client,
+              event,
+              botUserId: account.botUserId,
+            });
+          },
+        });
+      }
+      if (ctx.abortSignal.aborted) {
+        break;
       }
       const socket = client.websocket(workspaceId, afterCursor);
       await new Promise<void>((resolve, reject) => {

extensions/clickclack/src/http-client.test.ts

@@ -125,6 +125,35 @@ function streamedErrorResponse(body: string, limit: number) {
 }
 
 describe("ClickClack HTTP client", () => {
+  it("adds paged tail queries without changing the legacy events result", async () => {
+    const fetchMock = vi.fn(async () => Response.json({ events: [], tail_cursor: "cursor-900" }));
+    const client = createClickClackClient({
+      baseUrl: "https://clickclack.example",
+      token: "test-token",
+      fetch: fetchMock,
+    });
+
+    const page = await client.eventPage("workspace-1", {
+      afterCursor: "cursor-500",
+      limit: 500,
+      includeTail: true,
+    });
+    const legacyEvents = await client.events("workspace-1", "cursor-900");
+
+    expect(fetchMock).toHaveBeenNthCalledWith(
+      1,
+      "https://clickclack.example/api/realtime/events?workspace_id=workspace-1&after_cursor=cursor-500&limit=500&include_tail=true",
+      expect.any(Object),
+    );
+    expect(fetchMock).toHaveBeenNthCalledWith(
+      2,
+      "https://clickclack.example/api/realtime/events?workspace_id=workspace-1&after_cursor=cursor-900",
+      expect.any(Object),
+    );
+    expect(page).toEqual({ events: [], tailCursor: "cursor-900" });
+    expect(legacyEvents).toEqual([]);
+  });
+
   it("sends only safe bounded request correlation", async () => {
     const fetchMock = vi.fn(async (_input: string | URL | Request, _init?: RequestInit) =>
       Response.json({ user: { id: "usr_1" } }),

extensions/clickclack/src/http-client.ts

@@ -100,6 +100,33 @@ export function createClickClackClient(options: ClientOptions) {
     });
   }
 
+  async function fetchEventPage(
+    workspaceId: string,
+    pageOptions: {
+      afterCursor?: string;
+      limit?: number;
+      includeTail?: boolean;
+    } = {},
+  ): Promise<{ events: ClickClackEvent[]; tailCursor?: string }> {
+    const query = new URLSearchParams({ workspace_id: workspaceId });
+    if (pageOptions.afterCursor) {
+      query.set("after_cursor", pageOptions.afterCursor);
+    }
+    if (pageOptions.limit !== undefined) {
+      query.set("limit", String(pageOptions.limit));
+    }
+    if (pageOptions.includeTail) {
+      query.set("include_tail", "true");
+    }
+    const data = await request<{ events: ClickClackEvent[]; tail_cursor?: unknown }>(
+      `/api/realtime/events?${query.toString()}`,
+    );
+    return {
+      events: data.events,
+      ...(typeof data.tail_cursor === "string" ? { tailCursor: data.tail_cursor } : {}),
+    };
+  }
+
   return {
     me: async (): Promise<ClickClackUser> => {
       const data = await request<{ user: ClickClackUser }>("/api/me");
@@ -235,16 +262,9 @@ export function createClickClackClient(options: ClientOptions) {
       );
       return data.message;
     },
-    events: async (workspaceId: string, afterCursor?: string): Promise<ClickClackEvent[]> => {
-      const query = new URLSearchParams({ workspace_id: workspaceId });
-      if (afterCursor) {
-        query.set("after_cursor", afterCursor);
-      }
-      const data = await request<{ events: ClickClackEvent[] }>(
-        `/api/realtime/events?${query.toString()}`,
-      );
-      return data.events;
-    },
+    events: async (workspaceId: string, afterCursor?: string): Promise<ClickClackEvent[]> =>
+      (await fetchEventPage(workspaceId, { afterCursor })).events,
+    eventPage: fetchEventPage,
     websocket: (workspaceId: string, afterCursor?: string): WebSocket => {
       const url = new URL(`${baseUrl}/api/realtime/ws`);
       url.protocol = url.protocol === "https:" ? "wss:" : "ws:";