fix(telegram): bind general topic mutations

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",
         },
       },
     );

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);

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<boolean> => {

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({

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<string | number> {
   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;