diff --git a/.changeset/teams-stable-continuation.md b/.changeset/teams-stable-continuation.md new file mode 100644 index 000000000..24b394b57 --- /dev/null +++ b/.changeset/teams-stable-continuation.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Normalize Teams channel-thread conversation ids when deriving continuation tokens so parked messages and later card invokes resume the same session. diff --git a/packages/eve/src/public/channels/teams/api.test.ts b/packages/eve/src/public/channels/teams/api.test.ts index 45870c2a7..fd8b46fd6 100644 --- a/packages/eve/src/public/channels/teams/api.test.ts +++ b/packages/eve/src/public/channels/teams/api.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it, vi } from "vitest"; import { callTeamsConnectorApi, + parseTeamsConversationThreadId, replyToTeamsActivity, resolveTeamsAccessToken, sendTeamsActivity, @@ -22,6 +23,30 @@ describe("Teams Connector API wrapper", () => { ).toBe("T%201:19%3Aabc%40thread.skype:A%3A1"); }); + it("normalizes channel-thread conversation suffixes in continuation tokens", () => { + expect(parseTeamsConversationThreadId("19:abc@thread.skype;messageid=A%3A1")).toEqual({ + conversationId: "19:abc@thread.skype", + threadRootActivityId: "A:1", + }); + expect(parseTeamsConversationThreadId("19:abc@thread.skype;messageid=")).toEqual({ + conversationId: "19:abc@thread.skype", + threadRootActivityId: undefined, + }); + expect( + teamsContinuationToken({ + conversationId: "19:abc@thread.skype;messageid=A%3A1", + tenantId: "T 1", + }), + ).toBe("T%201:19%3Aabc%40thread.skype:A%3A1"); + expect( + teamsContinuationToken({ + conversationId: "19:abc@thread.skype;messageid=stale", + replyToActivityId: "A:1", + tenantId: "T 1", + }), + ).toBe("T%201:19%3Aabc%40thread.skype:A%3A1"); + }); + it("requests and caches Bot Connector access tokens", async () => { const apiFetch = vi.fn(async (_url: string | URL | Request, _init?: RequestInit) => Response.json({ access_token: "token-1", expires_in: 3600 }), diff --git a/packages/eve/src/public/channels/teams/api.ts b/packages/eve/src/public/channels/teams/api.ts index 5affd1c8b..c2b0e41f8 100644 --- a/packages/eve/src/public/channels/teams/api.ts +++ b/packages/eve/src/public/channels/teams/api.ts @@ -129,13 +129,48 @@ const accessTokenCache = new Map< { readonly accessToken: string; readonly expiresAt: number } >(); +/** Normalized Teams conversation id plus the thread root encoded in `;messageid=...`, when present. */ +export interface TeamsConversationThreadParts { + readonly conversationId: string; + readonly threadRootActivityId?: string; +} + +/** Splits Bot Framework channel-thread conversation ids like `19:...;messageid=...`. */ +export function parseTeamsConversationThreadId( + conversationId: string, +): TeamsConversationThreadParts { + const [baseConversationId, ...parameters] = conversationId.split(";"); + let threadRootActivityId: string | undefined; + + for (const parameter of parameters) { + const separatorIndex = parameter.indexOf("="); + const key = separatorIndex === -1 ? parameter : parameter.slice(0, separatorIndex); + if (key.toLowerCase() !== "messageid") continue; + + const rawValue = separatorIndex === -1 ? "" : parameter.slice(separatorIndex + 1); + const decodedValue = decodeTeamsConversationPart(rawValue); + if (decodedValue) threadRootActivityId = decodedValue; + break; + } + + return { + conversationId: baseConversationId || conversationId, + threadRootActivityId, + }; +} + /** Builds the channel-local continuation token for one Teams conversation/thread. */ export function teamsContinuationToken(input: { readonly conversationId: string; readonly replyToActivityId?: string | null; readonly tenantId?: string | null; }): string { - return [input.tenantId ?? "_", input.conversationId, input.replyToActivityId ?? ""] + const conversation = parseTeamsConversationThreadId(input.conversationId); + return [ + input.tenantId ?? "_", + conversation.conversationId, + input.replyToActivityId ?? conversation.threadRootActivityId ?? "", + ] .map((component) => encodeURIComponent(component)) .join(":"); } @@ -366,6 +401,14 @@ function normalizeAccessTokenResult(result: TeamsAccessTokenResult): { }; } +function decodeTeamsConversationPart(value: string): string { + try { + return decodeURIComponent(value); + } catch { + return value; + } +} + function toPostedActivity(body: unknown): TeamsPostedActivity { const raw = isObject(body) ? body : {}; return { diff --git a/packages/eve/src/public/channels/teams/inbound.test.ts b/packages/eve/src/public/channels/teams/inbound.test.ts index e7f7d6319..4ee9bb2a1 100644 --- a/packages/eve/src/public/channels/teams/inbound.test.ts +++ b/packages/eve/src/public/channels/teams/inbound.test.ts @@ -35,6 +35,20 @@ describe("Teams inbound parsing", () => { ); }); + it("prefers channel-thread messageid suffixes as the Teams thread root", () => { + const activity = parseTeamsActivity( + messageActivity({ + conversationId: "CONV;messageid=ROOT_FROM_CONVERSATION", + conversationType: "channel", + replyToId: "VOLATILE_REPLY", + }), + ); + + expect(activity?.type === "message" ? teamsThreadRootActivityId(activity) : "bad").toBe( + "ROOT_FROM_CONVERSATION", + ); + }); + it("keeps ambient RSC messages parseable but unmentioned", () => { const raw = messageActivity({ conversationType: "groupChat", text: "ambient" }); raw.entities = []; @@ -64,7 +78,9 @@ describe("Teams inbound parsing", () => { }); function messageActivity(input: { + readonly conversationId?: string; readonly conversationType: string; + readonly replyToId?: string; readonly text?: string; }): Record { return { @@ -73,7 +89,7 @@ function messageActivity(input: { team: { id: "TEAM" }, tenant: { id: "TENANT" }, }, - conversation: { conversationType: input.conversationType, id: "CONV" }, + conversation: { conversationType: input.conversationType, id: input.conversationId ?? "CONV" }, entities: [ { mentioned: { id: "BOT", name: "eve Bot" }, @@ -84,6 +100,7 @@ function messageActivity(input: { from: { id: "USER", name: "Ada" }, id: "ACTIVITY_1", recipient: { id: "BOT", name: "eve Bot" }, + replyToId: input.replyToId, serviceUrl: "https://smba.example.test/teams", text: input.text ?? "hello", textFormat: "xml", diff --git a/packages/eve/src/public/channels/teams/inbound.ts b/packages/eve/src/public/channels/teams/inbound.ts index 2c77006e9..828e4dad0 100644 --- a/packages/eve/src/public/channels/teams/inbound.ts +++ b/packages/eve/src/public/channels/teams/inbound.ts @@ -12,6 +12,7 @@ import type { TeamsChannelAccount, TeamsMention, } from "#public/channels/teams/api.js"; +import { parseTeamsConversationThreadId } from "#public/channels/teams/api.js"; import { isNonEmptyString, isObject } from "#shared/guards.js"; import { parseJsonObject } from "#shared/json.js"; @@ -126,7 +127,11 @@ export function teamsThreadRootActivityId( activity: TeamsMessageActivity | TeamsInvokeActivity, ): string | null { if (activity.scope === "personal") return null; - return activity.replyToId ?? activity.id; + return ( + parseTeamsConversationThreadId(activity.conversation.id).threadRootActivityId ?? + activity.replyToId ?? + activity.id + ); } /** diff --git a/packages/eve/src/public/channels/teams/teamsChannel.test.ts b/packages/eve/src/public/channels/teams/teamsChannel.test.ts index 69c1f5ee9..295903d6a 100644 --- a/packages/eve/src/public/channels/teams/teamsChannel.test.ts +++ b/packages/eve/src/public/channels/teams/teamsChannel.test.ts @@ -133,6 +133,63 @@ describe("teamsChannel", () => { ); }); + it("uses the same continuation token for suffixed channel messages and bare invokes", async () => { + const channel = teamsChannel({ + credentials: { webhookVerifier: () => true }, + onMessage() { + return { + auth: { + attributes: {}, + authenticator: "test", + principalId: "USER", + principalType: "user", + }, + }; + }, + }); + + const message = await firePost( + channel, + messageActivity({ + conversationId: "CONV;messageid=ROOT_ACTIVITY", + conversationType: "channel", + id: "REPLY_ACTIVITY", + replyToId: "VOLATILE_REPLY", + }), + ); + const invoke = await firePost(channel, { + ...baseActivity({ + conversationId: "CONV", + conversationType: "channel", + replyToId: "ROOT_ACTIVITY", + }), + name: "adaptiveCard/action", + type: "invoke", + value: { + action: { + data: { + eve_input: { requestId: "REQ", optionId: "approve" }, + }, + }, + }, + }); + + expect(message.send.mock.calls[0]![1]).toMatchObject({ + continuationToken: "TENANT:CONV:ROOT_ACTIVITY", + state: { + conversationId: "CONV;messageid=ROOT_ACTIVITY", + replyToActivityId: "ROOT_ACTIVITY", + }, + }); + expect(invoke.send.mock.calls[0]![1]).toMatchObject({ + continuationToken: "TENANT:CONV:ROOT_ACTIVITY", + state: { + conversationId: "CONV", + replyToActivityId: "ROOT_ACTIVITY", + }, + }); + }); + it("receive starts proactive sessions and anchors initial channel messages", async () => { const requests: Array<{ body: unknown; url: string }> = []; const channel = teamsChannel({ @@ -176,7 +233,12 @@ describe("teamsChannel", () => { }); }); -function messageActivity(input: { readonly conversationType: string }): Record { +function messageActivity(input: { + readonly conversationId?: string; + readonly conversationType: string; + readonly id?: string; + readonly replyToId?: string; +}): Record { return { ...baseActivity(input), entities: [ @@ -192,17 +254,23 @@ function messageActivity(input: { readonly conversationType: string }): Record { +function baseActivity(input: { + readonly conversationId?: string; + readonly conversationType: string; + readonly id?: string; + readonly replyToId?: string; +}): Record { return { channelData: { channel: { id: "CHANNEL" }, team: { id: "TEAM" }, tenant: { id: "TENANT" }, }, - conversation: { conversationType: input.conversationType, id: "CONV" }, + conversation: { conversationType: input.conversationType, id: input.conversationId ?? "CONV" }, from: { id: "USER", name: "Ada" }, - id: "ACTIVITY_1", + id: input.id ?? "ACTIVITY_1", recipient: { id: "BOT", name: "eve Bot" }, + replyToId: input.replyToId, serviceUrl: "https://smba.example.test/teams", }; }