diff --git a/extensions/telegram/src/action-runtime.test.ts b/extensions/telegram/src/action-runtime.test.ts index 1bb47b3069de..9e09a790b3ec 100644 --- a/extensions/telegram/src/action-runtime.test.ts +++ b/extensions/telegram/src/action-runtime.test.ts @@ -404,7 +404,7 @@ describe("handleTelegramAction", () => { expect(reactMessageTelegram).not.toHaveBeenCalled(); }); - it("soft-fails a topicless delegated reaction during a trusted topic turn", async () => { + it("binds a topicless delegated reaction to the trusted current topic", async () => { const result = await handleTelegramAction( { action: "react", @@ -420,6 +420,57 @@ describe("handleTelegramAction", () => { currentChannelProvider: "telegram", currentChannelId: "telegram:-1001:topic:77", currentMessageId: "456", + currentThreadTs: "77", + }, + }, + ); + + expect(resultDetails(result)).toMatchObject({ ok: true }); + expect(mockCall(reactMessageTelegram, 0, "topicless reaction")[0]).toBe("-1001"); + }); + + it("binds a General-topic reaction using trusted thread context", async () => { + const result = await handleTelegramAction( + { + action: "react", + chatId: "-1001", + messageId: 456, + emoji: "✅", + }, + reactionConfig("minimal"), + { + conversationReadOrigin: "delegated", + requesterAccountId: "default", + toolContext: { + currentChannelProvider: "telegram", + currentChannelId: "telegram:-1001", + currentMessageId: "456", + currentThreadTs: "1", + }, + }, + ); + + expect(resultDetails(result)).toMatchObject({ ok: true }); + expect(mockCall(reactMessageTelegram, 0, "General-topic reaction")[0]).toBe("-1001"); + }); + + it("soft-fails a topicless different-chat reaction during a trusted topic turn", async () => { + const result = await handleTelegramAction( + { + action: "react", + chatId: "-1002", + messageId: 456, + emoji: "✅", + }, + reactionConfig("minimal"), + { + conversationReadOrigin: "delegated", + requesterAccountId: "default", + toolContext: { + currentChannelProvider: "telegram", + currentChannelId: "telegram:-1001:topic:77", + currentMessageId: "456", + currentThreadTs: "77", }, }, ); diff --git a/extensions/telegram/src/draft-stream.test.ts b/extensions/telegram/src/draft-stream.test.ts index fd9e5679a660..9c7a87ba2ae6 100644 --- a/extensions/telegram/src/draft-stream.test.ts +++ b/extensions/telegram/src/draft-stream.test.ts @@ -582,8 +582,10 @@ describe("createTelegramDraftStream", () => { const api = createMockDraftApi(); api.sendMessage.mockReturnValueOnce(firstSend).mockResolvedValueOnce({ message_id: 42 }); const onSupersededPreview = vi.fn(); + const onProviderMessage = vi.fn(); const stream = createDraftStream(api, { onRetainedPage: onSupersededPreview, + onProviderMessage, replyToMessageId: 411, replyToMode, thread: { id: 42, scope: "dm" }, @@ -610,6 +612,8 @@ describe("createTelegramDraftStream", () => { // The raced first send is NOT retained as a durable chunk... expect(onSupersededPreview).not.toHaveBeenCalled(); + expect(onProviderMessage).toHaveBeenCalledTimes(1); + expect(onProviderMessage).toHaveBeenCalledWith(expect.objectContaining({ message_id: 42 })); expect(api.deleteMessage).not.toHaveBeenCalled(); // ...it is deleted deferred, so no orphaned stale bubble is left behind. await vi.advanceTimersByTimeAsync(4_000); diff --git a/extensions/telegram/src/draft-stream.ts b/extensions/telegram/src/draft-stream.ts index 895254e169f4..a09e573337d8 100644 --- a/extensions/telegram/src/draft-stream.ts +++ b/extensions/telegram/src/draft-stream.ts @@ -456,13 +456,15 @@ export function createTelegramDraftStream(params: { return true; } retainReplyTarget(sendGeneration, normalizedMessageId); - try { - await params.onProviderMessage?.(sent.message); - } catch (err) { - // Observation runs after Telegram accepted the send. Never turn a cache - // failure into a transport retry that could duplicate the message. - params.warn?.(`telegram stream preview observation failed: ${formatErrorMessage(err)}`); - } + const observeProviderMessage = async () => { + try { + await params.onProviderMessage?.(sent.message); + } catch (err) { + // Observation runs after Telegram accepted the send. Never turn a cache + // failure into a transport retry that could duplicate the message. + params.warn?.(`telegram stream preview observation failed: ${formatErrorMessage(err)}`); + } + }; if (sendGeneration !== generation) { const visibleSinceMs = Date.now(); if (repositionedSendGenerations.delete(sendGeneration)) { @@ -476,12 +478,14 @@ export function createTelegramDraftStream(params: { textSnapshot: sent.snapshot.text, visibleSinceMs, }); + await observeProviderMessage(); return true; } const visibleSinceMs = Date.now(); streamMessageId = normalizedMessageId; streamMessageSnapshot = sent.snapshot; streamVisibleSinceMs = visibleSinceMs; + await observeProviderMessage(); return true; }; const sendOrEditPlannedPage = async (page: PlannedTelegramDraftPage): Promise => { diff --git a/extensions/telegram/src/message-topic-binding.test.ts b/extensions/telegram/src/message-topic-binding.test.ts index e809a12a4eb8..0fd9cb94173f 100644 --- a/extensions/telegram/src/message-topic-binding.test.ts +++ b/extensions/telegram/src/message-topic-binding.test.ts @@ -102,7 +102,7 @@ describe("Telegram message topic binding", () => { ).resolves.toBe("-1001"); }); - it("rejects a delegated base-chat spelling during a trusted topic turn", async () => { + it("binds a delegated base-chat spelling to the trusted current topic", async () => { await expect( resolveTelegramMessageMutationChatId({ chatId: "-1001", @@ -111,6 +111,59 @@ describe("Telegram message topic binding", () => { accountId: "default", context: delegatedContext(), }), + ).resolves.toBe("-1001"); + }); + + it("binds General-topic mutations using trusted thread context", async () => { + await recordMessage({ messageId: 900, threadId: 1, providerObserved: true }); + await recordMessage({ messageId: 899, threadId: 2, providerObserved: true }); + const context = delegatedContext({ + toolContext: { + currentChannelProvider: "telegram", + currentChannelId: "telegram:-1001", + currentMessageId: "901", + currentThreadTs: "1", + }, + }); + + await expect( + resolveTelegramMessageMutationChatId({ + chatId: "-1001", + messageId: 901, + cfg, + accountId: "default", + context, + }), + ).resolves.toBe("-1001"); + await expect( + resolveTelegramMessageMutationChatId({ + chatId: "-1001", + messageId: 900, + cfg, + accountId: "default", + context, + }), + ).resolves.toBe("-1001"); + await expect( + resolveTelegramMessageMutationChatId({ + chatId: "-1001", + messageId: 899, + cfg, + accountId: "default", + context, + }), + ).rejects.toThrow("provider-observed binding"); + }); + + it("rejects a topicless different-chat target during a trusted topic turn", async () => { + await expect( + resolveTelegramMessageMutationChatId({ + chatId: "-1002", + messageId: 901, + cfg, + accountId: "default", + context: delegatedContext(), + }), ).rejects.toThrow("provider-observed binding"); }); @@ -218,6 +271,15 @@ describe("Telegram message topic binding", () => { }, }), }, + { + name: "conflicting trusted thread context", + context: delegatedContext({ + toolContext: { + ...delegatedContext().toolContext, + currentThreadTs: "88", + }, + }), + }, { name: "conflicting target forms", context: delegatedContext({ diff --git a/extensions/telegram/src/message-topic-binding.ts b/extensions/telegram/src/message-topic-binding.ts index 25e514d121f6..06fb89f6a567 100644 --- a/extensions/telegram/src/message-topic-binding.ts +++ b/extensions/telegram/src/message-topic-binding.ts @@ -32,34 +32,37 @@ function rejectUnboundTopicMutation(): never { throw new Error(TOPIC_BINDING_ERROR); } -function matchesCurrentTopic( +type CurrentTelegramConversation = { + hasThreadContext: boolean; + matchesChat: boolean; + threadId?: number; +}; + +function resolveCurrentTelegramConversation( toolContext: ChannelThreadingToolContext | undefined, chatId: string, - threadId: number, -): boolean { +): CurrentTelegramConversation { if (toolContext?.currentChannelProvider?.trim().toLowerCase() !== "telegram") { - return false; + return { hasThreadContext: false, matchesChat: false }; } const targets = [toolContext.currentChannelId, toolContext.currentMessagingTarget].filter( (value): value is string => typeof value === "string" && Boolean(value.trim()), ); - return ( + const parsedTargets = targets.map((value) => parseTelegramTarget(value)); + const threadIds = [ + ...parsedTargets.map((target) => target.messageThreadId), + parseStrictPositiveInteger(toolContext.currentThreadTs), + ].filter((value): value is number => value !== undefined); + const threadId = threadIds[0]; + const matchesChat = targets.length > 0 && - targets.every((value) => { - const current = parseTelegramTarget(value); - return current.chatId === chatId && current.messageThreadId === threadId; - }) - ); -} - -function hasCurrentTelegramTopic(toolContext: ChannelThreadingToolContext | undefined): boolean { - if (toolContext?.currentChannelProvider?.trim().toLowerCase() !== "telegram") { - return false; - } - return [toolContext.currentChannelId, toolContext.currentMessagingTarget].some( - (value) => - typeof value === "string" && parseTelegramTarget(value).messageThreadId !== undefined, - ); + parsedTargets.every((target) => target.chatId === chatId) && + (threadId === undefined || threadIds.every((value) => value === threadId)); + return { + hasThreadContext: threadIds.length > 0, + matchesChat, + ...(threadId !== undefined ? { threadId } : {}), + }; } export async function resolveTelegramMessageMutationChatId(params: { @@ -70,19 +73,20 @@ export async function resolveTelegramMessageMutationChatId(params: { context?: TelegramMessageMutationContext; }): Promise { const target = parseTelegramTarget(String(params.chatId)); - if (target.messageThreadId === undefined) { - // A topicless spelling must not bypass the provider check if this operation - // escaped shared normalization while still carrying trusted topic context. - if ( - params.context?.conversationReadOrigin !== "direct-operator" && - hasCurrentTelegramTopic(params.context?.toolContext) - ) { - return rejectUnboundTopicMutation(); - } + if (params.context?.conversationReadOrigin === "direct-operator") { + return target.messageThreadId === undefined ? params.chatId : target.chatId; + } + + const currentConversation = resolveCurrentTelegramConversation( + params.context?.toolContext, + target.chatId, + ); + const threadId = target.messageThreadId ?? currentConversation.threadId; + if (threadId === undefined && !currentConversation.hasThreadContext) { return params.chatId; } - if (params.context?.conversationReadOrigin === "direct-operator") { - return target.chatId; + if (threadId === undefined) { + return rejectUnboundTopicMutation(); } const selectedAccountId = normalizeOptionalAccountId( @@ -93,7 +97,8 @@ export async function resolveTelegramMessageMutationChatId(params: { !selectedAccountId || !requesterAccountId || normalizeAccountId(selectedAccountId) !== normalizeAccountId(requesterAccountId) || - !matchesCurrentTopic(params.context?.toolContext, target.chatId, target.messageThreadId) + !currentConversation.matchesChat || + currentConversation.threadId !== threadId ) { return rejectUnboundTopicMutation(); } @@ -115,7 +120,7 @@ export async function resolveTelegramMessageMutationChatId(params: { chatId: target.chatId, messageId: String(params.messageId), }); - if (!hasProviderObservedTelegramThreadBinding(cached, target.messageThreadId)) { + if (!hasProviderObservedTelegramThreadBinding(cached, threadId)) { return rejectUnboundTopicMutation(); } return target.chatId;