fix(matrix): stop poll pagination on repeated cursors (#128214)

extensions/matrix/src/matrix/actions/messages.test.ts

@@ -110,14 +110,24 @@ function createMessagesClient(params: {
     }
     return null;
   });
-  const getRelations = vi.fn(async (_roomId: string, _eventId: string, relType: string) => ({
-    events:
-      relType === "m.thread"
-        ? (params.threadRelations ?? params.pollRelations ?? [])
-        : (params.pollRelations ?? []),
-    nextBatch: null,
-    prevBatch: null,
-  }));
+  const getRelations = vi.fn(
+    async (
+      _roomId: string,
+      _eventId: string,
+      relType: string,
+    ): Promise<{
+      events: Array<Record<string, unknown>>;
+      nextBatch: string | null;
+      prevBatch: string | null;
+    }> => ({
+      events:
+        relType === "m.thread"
+          ? (params.threadRelations ?? params.pollRelations ?? [])
+          : (params.pollRelations ?? []),
+      nextBatch: null,
+      prevBatch: null,
+    }),
+  );
 
   return {
     client: {
@@ -356,6 +366,33 @@ describe("matrix message actions", () => {
     });
   });
 
+  it.each([
+    { name: "room history", threadId: undefined },
+    { name: "poll-rooted thread history", threadId: "$poll" },
+  ])("fails visibly on cyclic poll pagination in $name", async ({ threadId }) => {
+    const pollRoot = createPollStartEvent();
+    const { client, getRelations } = createMessagesClient({
+      chunk: threadId ? [] : [pollRoot],
+      pollRoot,
+    });
+    let pollPageCalls = 0;
+    getRelations.mockImplementation(async (_roomId, _eventId, relationType) => {
+      if (relationType !== "m.reference") {
+        return { events: [], nextBatch: null, prevBatch: null };
+      }
+      pollPageCalls += 1;
+      if (pollPageCalls > 2) {
+        throw new Error("test stopped unbounded Matrix poll pagination");
+      }
+      return { events: [], nextBatch: "stuck", prevBatch: null };
+    });
+
+    await expect(
+      readMatrixMessages("room:!room:example.org", { client, threadId }),
+    ).rejects.toThrow("Matrix poll pagination returned a repeated cursor");
+    expect(pollPageCalls).toBe(2);
+  });
+
   it("dedupes multiple poll events for the same poll within one read page", async () => {
     const { client, getEvent } = createMessagesClient({
       chunk: [createPollResponseEvent(), createPollStartEvent()],

extensions/matrix/src/matrix/actions/pins.test.ts

@@ -77,4 +77,47 @@ describe("matrix pins actions", () => {
       },
     ]);
   });
+
+  it("keeps other pinned messages visible when a poll repeats its pagination cursor", async () => {
+    let pollPageCalls = 0;
+    const getRelations = vi.fn(async () => {
+      pollPageCalls += 1;
+      if (pollPageCalls > 2) {
+        throw new Error("test stopped unbounded Matrix poll pagination");
+      }
+      return { events: [], nextBatch: "stuck", prevBatch: null };
+    });
+    const client = {
+      getRoomStateEvent: async () => ({ pinned: ["$poll", "$message"] }),
+      getEvent: async (_roomId: string, eventId: string) =>
+        eventId === "$poll"
+          ? {
+              event_id: "$poll",
+              sender: "@alice:example.org",
+              type: "m.poll.start",
+              origin_server_ts: 1,
+              content: {
+                "m.poll.start": {
+                  question: { "m.text": "Lunch?" },
+                  answers: [{ id: "pizza", "m.text": "Pizza" }],
+                },
+              },
+            }
+          : {
+              event_id: "$message",
+              sender: "@alice:example.org",
+              type: "m.room.message",
+              origin_server_ts: 2,
+              content: { msgtype: "m.text", body: "Still visible" },
+            },
+      getRelations,
+      stop: vi.fn(),
+    } as unknown as MatrixClient;
+
+    const result = await listMatrixPins("!room:example.org", { client });
+
+    expect(result.pinned).toEqual(["$poll", "$message"]);
+    expect(result.events.map((event) => event.eventId)).toEqual(["$message"]);
+    expect(getRelations).toHaveBeenCalledTimes(2);
+  });
 });

extensions/matrix/src/matrix/monitor/handler.body-for-agent.test.ts

@@ -211,6 +211,54 @@ describe("createMatrixRoomMessageHandler inbound body formatting", () => {
     expect(latestSessionKey(recordInboundSession)).toBe("agent:ops:main");
   });
 
+  it("settles inbound poll events when relation pagination repeats a cursor", async () => {
+    let pollPageCalls = 0;
+    const getRelations = vi.fn(async () => {
+      pollPageCalls += 1;
+      if (pollPageCalls > 2) {
+        throw new Error("test stopped unbounded Matrix poll pagination");
+      }
+      return { events: [], nextBatch: "stuck", prevBatch: null };
+    });
+    const logVerboseMessage = vi.fn();
+    const { handler, finalizeInboundContext } = createMatrixHandlerTestHarness({
+      client: {
+        getEvent: async () => ({
+          event_id: "$poll",
+          sender: "@bot:example.org",
+          type: "m.poll.start",
+          origin_server_ts: 1,
+          content: {
+            "m.poll.start": {
+              question: { "m.text": "Lunch?" },
+              answers: [{ id: "pizza", "m.text": "Pizza" }],
+            },
+          },
+        }),
+        getRelations,
+      } as unknown as Partial<MatrixClient>,
+      isDirectMessage: true,
+      logVerboseMessage,
+    });
+
+    await handler("!room:example.org", {
+      type: "m.poll.response",
+      sender: "@user:example.org",
+      event_id: "$vote",
+      origin_server_ts: 2,
+      content: {
+        "m.poll.response": { answers: ["pizza"] },
+        "m.relates_to": { rel_type: "m.reference", event_id: "$poll" },
+      },
+    } as MatrixRawEvent);
+
+    expect(getRelations).toHaveBeenCalledTimes(2);
+    expect(finalizeInboundContext).not.toHaveBeenCalled();
+    expect(logVerboseMessage).toHaveBeenCalledWith(
+      expect.stringContaining("Matrix poll pagination returned a repeated cursor"),
+    );
+  });
+
   it("records reply context for quoted poll start events inside always-threaded replies", async () => {
     const { handler, finalizeInboundContext } = createMatrixHandlerTestHarness({
       client: {

extensions/matrix/src/matrix/poll-summary.test.ts

@@ -0,0 +1,81 @@
+import { describe, expect, it, vi } from "vitest";
+import { fetchMatrixPollSnapshot } from "./poll-summary.js";
+import type { MatrixClient, MatrixRawEvent } from "./sdk.js";
+
+function createPollStartEvent(): MatrixRawEvent {
+  return {
+    event_id: "$poll",
+    sender: "@alice:example.org",
+    type: "m.poll.start",
+    origin_server_ts: 1,
+    content: {
+      "m.poll.start": {
+        question: { "m.text": "Lunch?" },
+        answers: [{ id: "pizza", "m.text": "Pizza" }],
+      },
+    },
+  } as MatrixRawEvent;
+}
+
+describe("Matrix poll snapshot pagination", () => {
+  it.each([
+    { name: "a repeated cursor", cursors: ["stuck", "stuck"] },
+    { name: "a cursor cycle", cursors: ["first", "second", "first"] },
+  ])("fails visibly on $name", async ({ cursors }) => {
+    let calls = 0;
+    const getRelations = vi.fn(async () => {
+      const nextBatch = cursors[calls++];
+      if (nextBatch === undefined) {
+        throw new Error("test stopped unbounded Matrix poll pagination");
+      }
+      return { events: [], nextBatch, prevBatch: null };
+    });
+
+    await expect(
+      fetchMatrixPollSnapshot(
+        { getRelations } as unknown as MatrixClient,
+        "!room:example.org",
+        createPollStartEvent(),
+      ),
+    ).rejects.toThrow("Matrix poll pagination returned a repeated cursor");
+    expect(getRelations).toHaveBeenCalledTimes(cursors.length);
+  });
+
+  it("follows valid empty pages before collecting later poll votes", async () => {
+    const getRelations = vi
+      .fn()
+      .mockResolvedValueOnce({ events: [], nextBatch: "next-page", prevBatch: null })
+      .mockResolvedValueOnce({
+        events: [
+          {
+            event_id: "$vote",
+            sender: "@bob:example.org",
+            type: "m.poll.response",
+            origin_server_ts: 2,
+            content: {
+              "m.poll.response": { answers: ["pizza"] },
+              "m.relates_to": { rel_type: "m.reference", event_id: "$poll" },
+            },
+          },
+        ],
+        nextBatch: null,
+        prevBatch: null,
+      });
+
+    const snapshot = await fetchMatrixPollSnapshot(
+      { getRelations } as unknown as MatrixClient,
+      "!room:example.org",
+      createPollStartEvent(),
+    );
+
+    expect(getRelations).toHaveBeenNthCalledWith(
+      2,
+      "!room:example.org",
+      "$poll",
+      "m.reference",
+      undefined,
+      { from: "next-page" },
+    );
+    expect(snapshot?.text).toContain("1. Pizza (1 vote)");
+  });
+});

extensions/matrix/src/matrix/poll-summary.ts

@@ -35,6 +35,7 @@ async function readAllPollRelations(
   pollEventId: string,
 ): Promise<MatrixRawEvent[]> {
   const relationEvents: MatrixRawEvent[] = [];
+  const seenCursors = new Set<string>();
   let nextBatch: string | undefined;
   do {
     const page = await client.getRelations(roomId, pollEventId, "m.reference", undefined, {
@@ -42,6 +43,13 @@ async function readAllPollRelations(
     });
     relationEvents.push(...page.events);
     nextBatch = page.nextBatch ?? undefined;
+    // Encrypted pages may be empty; only a repeated cursor proves pagination cannot progress.
+    if (nextBatch && seenCursors.has(nextBatch)) {
+      throw new Error("Matrix poll pagination returned a repeated cursor");
+    }
+    if (nextBatch) {
+      seenCursors.add(nextBatch);
+    }
   } while (nextBatch);
   return relationEvents;
 }