diff --git a/extensions/matrix/src/matrix/actions/messages.test.ts b/extensions/matrix/src/matrix/actions/messages.test.ts index c9417fa74687..3d1a07ca4ab7 100644 --- a/extensions/matrix/src/matrix/actions/messages.test.ts +++ b/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>; + 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()], diff --git a/extensions/matrix/src/matrix/actions/pins.test.ts b/extensions/matrix/src/matrix/actions/pins.test.ts index 32dbb9a14af5..73b05da3f3a3 100644 --- a/extensions/matrix/src/matrix/actions/pins.test.ts +++ b/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); + }); }); diff --git a/extensions/matrix/src/matrix/monitor/handler.body-for-agent.test.ts b/extensions/matrix/src/matrix/monitor/handler.body-for-agent.test.ts index 51f7069633f5..9864cadb96ba 100644 --- a/extensions/matrix/src/matrix/monitor/handler.body-for-agent.test.ts +++ b/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, + 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: { diff --git a/extensions/matrix/src/matrix/poll-summary.test.ts b/extensions/matrix/src/matrix/poll-summary.test.ts new file mode 100644 index 000000000000..94d789aa9033 --- /dev/null +++ b/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)"); + }); +}); diff --git a/extensions/matrix/src/matrix/poll-summary.ts b/extensions/matrix/src/matrix/poll-summary.ts index 55528c0a7562..5fdb5678eaba 100644 --- a/extensions/matrix/src/matrix/poll-summary.ts +++ b/extensions/matrix/src/matrix/poll-summary.ts @@ -35,6 +35,7 @@ async function readAllPollRelations( pollEventId: string, ): Promise { const relationEvents: MatrixRawEvent[] = []; + const seenCursors = new Set(); 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; }