diff --git a/extensions/qa-lab/src/scenario-catalog-matrix-session-identity-proof.test.ts b/extensions/qa-lab/src/scenario-catalog-matrix-session-identity-proof.test.ts new file mode 100644 index 000000000000..91086ae6d4da --- /dev/null +++ b/extensions/qa-lab/src/scenario-catalog-matrix-session-identity-proof.test.ts @@ -0,0 +1,268 @@ +import { describe, expect, it, vi } from "vitest"; +import { createQaBusState } from "./bus-state.js"; +import { readQaScenarioById } from "./scenario-catalog.js"; +import { runLoadedScenarioFlow } from "./scenario-flow-runner.test-support.js"; + +const matrixDriverId = "@driver:matrix.test"; +const primaryRoomId = "!primary:matrix.test"; +const secondaryRoomId = "!secondary:matrix.test"; +const sharedSessionKey = `agent:qa:matrix:direct:${matrixDriverId}`; + +type MatrixScenarioId = "dm-per-room-session" | "dm-shared-session"; + +function createMatrixSessionEntry( + sessionId: string, + roomId: string, + updatedAt: number, + nativeDirectUserId = matrixDriverId, +) { + return { + sessionId, + updatedAt, + chatType: "direct", + delivery: { + kind: "external", + route: { + channel: "matrix", + accountId: "sut", + target: { + to: `room:${roomId}`, + chatType: "direct", + }, + }, + context: { + channel: "matrix", + to: `room:${roomId}`, + accountId: "sut", + }, + origin: { + provider: "matrix", + surface: "matrix", + accountId: "sut", + chatType: "direct", + from: `matrix:${nativeDirectUserId}`, + to: `room:${roomId}`, + nativeChannelId: roomId, + nativeDirectUserId, + }, + }, + }; +} + +async function runMatrixSessionScenario(params: { + scenarioId: MatrixScenarioId; + sharedSessionIdentity: boolean; + sharedSessionKey?: boolean; + sharedTranscriptId?: boolean; + returnedSenderId?: string; + secondaryNativeSenderId?: string; +}) { + const scenario = readQaScenarioById(params.scenarioId); + const config = scenario.execution.config as { + primaryConversationId: string; + secondaryConversationId: string; + primaryMarker: string; + secondaryMarker: string; + }; + const state = createQaBusState(); + const usesSharedSessionKey = params.sharedSessionKey ?? params.sharedSessionIdentity; + const usesSharedTranscriptId = params.sharedTranscriptId ?? params.sharedSessionIdentity; + const primarySessionKey = usesSharedSessionKey + ? sharedSessionKey + : `agent:qa:matrix:channel:${primaryRoomId}`; + const secondarySessionKey = usesSharedSessionKey + ? sharedSessionKey + : `agent:qa:matrix:channel:${secondaryRoomId}`; + const primarySessionId = "matrix-session-primary"; + const secondarySessionId = usesSharedTranscriptId ? primarySessionId : "matrix-session-secondary"; + + const readRawQaSessionStore = vi.fn(async () => { + const inboundMessages = state + .getSnapshot() + .messages.filter((message) => message.direction === "inbound"); + if (inboundMessages.length === 0) { + return {}; + } + if (inboundMessages.length === 1) { + return { + [primarySessionKey]: createMatrixSessionEntry(primarySessionId, primaryRoomId, 1), + }; + } + if (usesSharedSessionKey) { + return { + [sharedSessionKey]: createMatrixSessionEntry( + secondarySessionId, + secondaryRoomId, + 2, + params.secondaryNativeSenderId, + ), + }; + } + return { + [primarySessionKey]: createMatrixSessionEntry(primarySessionId, primaryRoomId, 1), + [secondarySessionKey]: createMatrixSessionEntry( + secondarySessionId, + secondaryRoomId, + 2, + params.secondaryNativeSenderId, + ), + }; + }); + + let outboundWaitCount = 0; + const transport = { + id: "matrix", + accountId: "sut", + state, + reset: async () => { + state.reset(); + }, + sendInbound: async (input: Parameters[0]) => + state.addInboundMessage({ + ...input, + accountId: "sut", + senderId: params.returnedSenderId ?? matrixDriverId, + }), + waitForOutbound: async (input: { + conversation?: { id: string; kind: string }; + sinceIndex?: number; + textIncludes?: string; + timeoutMs?: number; + }) => { + outboundWaitCount += 1; + if (outboundWaitCount === 1) { + state.addOutboundMessage({ + accountId: "sut", + to: `dm:${config.primaryConversationId}`, + text: config.primaryMarker, + }); + } else if (outboundWaitCount === 2) { + if (params.scenarioId === "dm-shared-session") { + state.addOutboundMessage({ + accountId: "sut", + to: `dm:${config.secondaryConversationId}`, + text: "This Matrix DM is sharing a session with another room. Set channels.matrix.dm.sessionScope to per-room to isolate it.", + }); + } + state.addOutboundMessage({ + accountId: "sut", + to: `dm:${config.secondaryConversationId}`, + text: config.secondaryMarker, + }); + } + const match = state + .getSnapshot() + .messages.filter((message) => message.direction === "outbound") + .slice(input.sinceIndex ?? 0) + .find( + (message) => + (!input.conversation || message.conversation.id === input.conversation.id) && + (!input.conversation || message.conversation.kind === input.conversation.kind) && + (!input.textIncludes || message.text.includes(input.textIncludes)), + ); + if (!match) { + throw new Error(`timed out after ${input.timeoutMs}ms waiting for Matrix outbound marker`); + } + return match; + }, + }; + + return await runLoadedScenarioFlow(params.scenarioId, { + state, + api: { + env: { + providerMode: "mock-openai", + gateway: { tempRoot: "/qa-matrix-session-identity" }, + }, + transport, + readRawQaSessionStore, + }, + }); +} + +describe("Matrix DM scenario session identity evidence", () => { + it.each([ + { lane: "live", returnedSenderId: matrixDriverId }, + { lane: "crabline", returnedSenderId: "driver" }, + ])( + "accepts distinct room-owned sessions on the $lane Matrix lane", + async ({ returnedSenderId }) => { + await expect( + runMatrixSessionScenario({ + scenarioId: "dm-per-room-session", + sharedSessionIdentity: false, + returnedSenderId, + }), + ).resolves.toMatchObject({ status: "pass" }); + }, + ); + + it("rejects a shared session in per-room mode even when both replies and notice policy pass", async () => { + await expect( + runMatrixSessionScenario({ + scenarioId: "dm-per-room-session", + sharedSessionIdentity: true, + }), + ).rejects.toThrow(/session|room|isolat|shared/i); + }); + + it("rejects a reused per-room session key even when its transcript id rotates", async () => { + await expect( + runMatrixSessionScenario({ + scenarioId: "dm-per-room-session", + sharedSessionIdentity: false, + sharedSessionKey: true, + sharedTranscriptId: false, + }), + ).rejects.toThrow(/session|room|isolat|shared/i); + }); + + it.each([ + { lane: "live", returnedSenderId: matrixDriverId }, + { lane: "crabline", returnedSenderId: "driver" }, + ])( + "accepts one shared user-owned session on the $lane Matrix lane", + async ({ returnedSenderId }) => { + await expect( + runMatrixSessionScenario({ + scenarioId: "dm-shared-session", + sharedSessionIdentity: true, + returnedSenderId, + }), + ).resolves.toMatchObject({ status: "pass" }); + }, + ); + + it("rejects separate sessions in shared mode even when the expected notice is emitted", async () => { + await expect( + runMatrixSessionScenario({ + scenarioId: "dm-shared-session", + sharedSessionIdentity: false, + }), + ).rejects.toThrow(/session|room|isolat|shared/i); + }); + + it("rejects distinct shared-mode session keys even when they alias one transcript id", async () => { + await expect( + runMatrixSessionScenario({ + scenarioId: "dm-shared-session", + sharedSessionIdentity: true, + sharedSessionKey: false, + sharedTranscriptId: true, + }), + ).rejects.toThrow(/session|room|isolat|shared/i); + }); + + it.each([ + { scenarioId: "dm-per-room-session" as const, sharedSessionIdentity: false }, + { scenarioId: "dm-shared-session" as const, sharedSessionIdentity: true }, + ])("rejects a different persisted Matrix sender in $scenarioId", async (scenario) => { + await expect( + runMatrixSessionScenario({ + ...scenario, + returnedSenderId: "driver", + secondaryNativeSenderId: "@intruder:matrix.test", + }), + ).rejects.toThrow(/session|room|isolat|shared/i); + }); +}); diff --git a/qa/scenarios/channels/dm-per-room-session.yaml b/qa/scenarios/channels/dm-per-room-session.yaml index a506a5ed81c9..1186ec5e4c8a 100644 --- a/qa/scenarios/channels/dm-per-room-session.yaml +++ b/qa/scenarios/channels/dm-per-room-session.yaml @@ -57,6 +57,27 @@ flow: ref: config.primaryMarker timeoutMs: expr: liveTurnTimeoutMs(env, 45000) + - call: readRawQaSessionStore + saveAs: primarySessionStore + args: + - ref: env + - set: primarySession + value: + expr: |- + Object.entries(primarySessionStore) + .filter(([, entry]) => + entry?.delivery?.kind === 'external' && + entry.delivery.context?.channel === 'matrix' && + entry.delivery.context?.accountId === transport.accountId && + entry.delivery.origin?.chatType === 'direct' && + typeof entry.delivery.origin?.nativeDirectUserId === 'string' && + entry.delivery.origin.nativeDirectUserId.length > 0 && + typeof entry.delivery.origin?.nativeChannelId === 'string' && + typeof entry.sessionId === 'string') + .toSorted(([, left], [, right]) => Number(right.updatedAt ?? 0) - Number(left.updatedAt ?? 0))[0] + - assert: + expr: Boolean(primarySession) + message: first Matrix DM turn did not create a persisted account-scoped direct session - set: secondaryStartIndex value: expr: state.getSnapshot().messages.filter((candidate) => candidate.direction === 'outbound').length @@ -81,6 +102,31 @@ flow: timeoutMs: expr: liveTurnTimeoutMs(env, 45000) saveAs: reply + - call: readRawQaSessionStore + saveAs: secondarySessionStore + args: + - ref: env + - set: secondarySession + value: + expr: |- + Object.entries(secondarySessionStore) + .filter(([, entry]) => + entry?.delivery?.kind === 'external' && + entry.delivery.context?.channel === 'matrix' && + entry.delivery.context?.accountId === transport.accountId && + entry.delivery.origin?.chatType === 'direct' && + entry.delivery.origin?.nativeDirectUserId === primarySession[1].delivery.origin.nativeDirectUserId && + entry.delivery.origin?.nativeChannelId !== primarySession[1].delivery.origin.nativeChannelId && + typeof entry.delivery.origin?.nativeChannelId === 'string' && + typeof entry.sessionId === 'string') + .toSorted(([, left], [, right]) => Number(right.updatedAt ?? 0) - Number(left.updatedAt ?? 0))[0] + - assert: + expr: Boolean(secondarySession) + message: second Matrix DM turn did not create a persisted session for a distinct room + - assert: + expr: "secondarySession[0] !== primarySession[0] && secondarySession[1].sessionId !== primarySession[1].sessionId" + message: + expr: "`Matrix per-room DM rooms must own distinct session keys and transcripts: ${JSON.stringify({ primaryKey: primarySession[0], secondaryKey: secondarySession[0], primarySessionId: primarySession[1].sessionId, secondarySessionId: secondarySession[1].sessionId })}`" - call: sleep args: - 1500 diff --git a/qa/scenarios/channels/dm-shared-session.yaml b/qa/scenarios/channels/dm-shared-session.yaml index 189ad77653b0..e0706d5ee1ea 100644 --- a/qa/scenarios/channels/dm-shared-session.yaml +++ b/qa/scenarios/channels/dm-shared-session.yaml @@ -51,6 +51,27 @@ flow: ref: config.primaryMarker timeoutMs: expr: liveTurnTimeoutMs(env, 45000) + - call: readRawQaSessionStore + saveAs: primarySessionStore + args: + - ref: env + - set: primarySession + value: + expr: |- + Object.entries(primarySessionStore) + .filter(([, entry]) => + entry?.delivery?.kind === 'external' && + entry.delivery.context?.channel === 'matrix' && + entry.delivery.context?.accountId === transport.accountId && + entry.delivery.origin?.chatType === 'direct' && + typeof entry.delivery.origin?.nativeDirectUserId === 'string' && + entry.delivery.origin.nativeDirectUserId.length > 0 && + typeof entry.delivery.origin?.nativeChannelId === 'string' && + typeof entry.sessionId === 'string') + .toSorted(([, left], [, right]) => Number(right.updatedAt ?? 0) - Number(left.updatedAt ?? 0))[0] + - assert: + expr: Boolean(primarySession) + message: first Matrix DM turn did not create a persisted account-scoped direct session - set: secondaryStartIndex value: expr: state.getSnapshot().messages.filter((candidate) => candidate.direction === 'outbound').length @@ -75,6 +96,31 @@ flow: timeoutMs: expr: liveTurnTimeoutMs(env, 45000) saveAs: reply + - call: readRawQaSessionStore + saveAs: secondarySessionStore + args: + - ref: env + - set: secondarySession + value: + expr: |- + Object.entries(secondarySessionStore) + .filter(([, entry]) => + entry?.delivery?.kind === 'external' && + entry.delivery.context?.channel === 'matrix' && + entry.delivery.context?.accountId === transport.accountId && + entry.delivery.origin?.chatType === 'direct' && + entry.delivery.origin?.nativeDirectUserId === primarySession[1].delivery.origin.nativeDirectUserId && + entry.delivery.origin?.nativeChannelId !== primarySession[1].delivery.origin.nativeChannelId && + typeof entry.delivery.origin?.nativeChannelId === 'string' && + typeof entry.sessionId === 'string') + .toSorted(([, left], [, right]) => Number(right.updatedAt ?? 0) - Number(left.updatedAt ?? 0))[0] + - assert: + expr: Boolean(secondarySession) + message: second Matrix DM turn did not update the persisted session for its distinct room + - assert: + expr: "secondarySession[0] === primarySession[0] && secondarySession[1].sessionId === primarySession[1].sessionId" + message: + expr: "`Matrix per-user DM rooms unexpectedly use separate sessions: ${JSON.stringify({ primaryKey: primarySession[0], secondaryKey: secondarySession[0], primarySessionId: primarySession[1].sessionId, secondarySessionId: secondarySession[1].sessionId })}`" - waitForOutbound: conversation: id: