[1/2] feat(eve)!: key Telegram group sessions chat-wide

.changeset/telegram-chat-wide-group-sessions.md

@@ -0,0 +1,5 @@
+---
+"eve": minor
+---
+
+Telegram group and supergroup chats now keep one continuous session per chat or forum topic, the same way private chats do. A fresh mention resumes the conversation instead of starting a new session anchored to that message, replies to bot messages and callback queries land on the same session, and outbound group sends no longer re-key the session to the posted message id. Pass `conversationId` on a proactive `receive` target to pin a specific thread.

docs/channels/telegram.mdx

@@ -41,6 +41,12 @@ The model-visible `<telegram_context>` includes `bot_username` and `is_mentioned
 
 Forum topics carry `message_thread_id` in the continuation token, so each topic stays on its own thread.
 
+### Group sessions
+
+Each group chat keeps one continuous session keyed to the chat, the same way private chats do. Mentions, replies to the bot, callback queries, and outbound sends all resume that session, so the bot remembers what it said earlier in the chat. Forum topics get one session per topic; the `message_thread_id` that non-forum supergroups stamp on replies is a reply chain, not a topic, and does not split the session.
+
+To pin a proactive session to a specific thread instead, pass `conversationId` on the `receive` target.
+
 To customize auth or filtering, override `onMessage`. Return `title` alongside `auth` to set the title when the dispatch starts a run. Group privacy mode itself lives in BotFather, not here.
 
 ### Delivery
@@ -55,7 +61,7 @@ Human-in-the-loop (HITL) turns option requests into inline-keyboard buttons and
 
 Start a session without an inbound message through `to(telegram, target).send(message, { auth })` from a schedule `run` handler, or `ctx.to(telegram, target).send(message, { auth })` from another channel. `target.chatId` is required. Add `messageThreadId` to land in a specific forum topic.
 
-Private proactive chats stay keyed to the chat, or to the chat plus `messageThreadId` when you target a topic. Group and supergroup proactive sends anchor to the bot message id returned by Telegram, so replies to different bot messages can resume different sessions in the same chat. If Telegram does not return a recognized chat type for an outbound send, eve keeps the session unanchored instead of guessing.
+Proactive chats stay keyed to the chat, or to the chat plus `messageThreadId` when you target a topic, for private and group chats alike. Pass `conversationId` to resume a specific thread instead.
 
 ### Attachments
 

packages/eve/src/public/channels/telegram/inbound.ts

@@ -47,6 +47,8 @@ export interface TelegramAttachment {
 export interface TelegramMessageReference {
   readonly chat: TelegramChat;
   readonly from?: TelegramUser;
+  /** `true` when the message was sent to a forum topic. */
+  readonly isTopicMessage?: boolean;
   readonly messageId: string;
   readonly messageThreadId?: number;
 }
@@ -61,6 +63,8 @@ export interface TelegramMessage {
   readonly caption: string;
   readonly chat: TelegramChat;
   readonly from?: TelegramUser;
+  /** `true` when the message was sent to a forum topic. */
+  readonly isTopicMessage?: boolean;
   readonly messageId: string;
   readonly messageThreadId?: number;
   readonly raw: Record<string, unknown>;
@@ -155,6 +159,7 @@ function parseTelegramMessage(value: unknown): TelegramMessage | null {
     caption: typeof value.caption === "string" ? value.caption : "",
     chat,
     from: parseTelegramUser(value.from),
+    isTopicMessage: value.is_topic_message === true ? true : undefined,
     messageId,
     messageThreadId:
       typeof value.message_thread_id === "number" ? value.message_thread_id : undefined,
@@ -185,6 +190,7 @@ function parseMessageReference(value: unknown): TelegramMessageReference | undef
   return {
     chat,
     from: parseTelegramUser(value.from),
+    isTopicMessage: value.is_topic_message === true ? true : undefined,
     messageId,
     messageThreadId:
       typeof value.message_thread_id === "number" ? value.message_thread_id : undefined,

packages/eve/src/public/channels/telegram/state.ts

@@ -1,18 +1,21 @@
 import { telegramContinuationToken } from "#public/channels/telegram/api.js";
-import type { TelegramCallbackQuery, TelegramMessage } from "#public/channels/telegram/inbound.js";
+import type {
+  TelegramCallbackQuery,
+  TelegramMessage,
+  TelegramMessageReference,
+} from "#public/channels/telegram/inbound.js";
 import type { TelegramChannelState } from "#public/channels/telegram/telegramChannel.js";
 
 export function stateFromTelegramMessage(
   message: TelegramMessage,
   botUsername: string | undefined,
 ): TelegramChannelState {
-  const privateChat = message.chat.type === "private";
   return {
     ...initialTelegramState(botUsername),
     chatId: message.chat.id,
     chatType: message.chat.type,
-    conversationId: privateChat ? null : conversationIdForMessage(message),
-    messageThreadId: message.messageThreadId ?? null,
+    conversationId: null,
+    messageThreadId: sessionThreadId(message),
     triggeringUserId: message.from?.id ?? null,
   };
 }
@@ -25,13 +28,12 @@ export function stateFromTelegramCallbackQuery(
   if (!message) {
     return { ...initialTelegramState(botUsername), triggeringUserId: query.from.id };
   }
-  const privateChat = message.chat.type === "private";
   return {
     ...initialTelegramState(botUsername),
     chatId: message.chat.id,
     chatType: message.chat.type,
-    conversationId: privateChat ? null : message.messageId,
-    messageThreadId: message.messageThreadId ?? null,
+    conversationId: null,
+    messageThreadId: sessionThreadId(message),
     triggeringUserId: query.from.id,
   };
 }
@@ -59,8 +61,8 @@ export function initialTelegramState(botUsername: string | undefined): TelegramC
   };
 }
 
-function conversationIdForMessage(message: TelegramMessage): string {
-  return message.replyToMessage?.from?.isBot === true
-    ? message.replyToMessage.messageId
-    : message.messageId;
+function sessionThreadId(message: TelegramMessageReference): number | null {
+  const threadId = message.messageThreadId ?? null;
+  if (message.chat.type === "private") return threadId;
+  return message.isTopicMessage === true ? threadId : null;
 }

packages/eve/src/public/channels/telegram/telegramChannel.test.ts

@@ -234,12 +234,95 @@ describe("telegramChannel() inbound route", () => {
       },
     });
     expect(mentioned.send).toHaveBeenCalledTimes(1);
-    expect(mentioned.send.mock.calls[0]![0]).toBe("-1001::11");
+    expect(mentioned.send.mock.calls[0]![0]).toBe("-1001::");
     expect((mentioned.send.mock.calls[0]![1] as { context: string[] }).context[0]).toContain(
       "is_mentioned: true",
     );
   });
 
+  it("keys group mentions, bot replies, and callback queries chat-wide", async () => {
+    const channel = telegramChannel({
+      api: { fetch: fakeTelegramFetch() },
+      botUsername: "testbot",
+      credentials: { botToken: "bot-token", webhookSecretToken: SECRET },
+    });
+
+    const mentioned = await firePost(channel, {
+      message: {
+        message_id: 11,
+        from: { id: 42, is_bot: false },
+        chat: { id: -1001, type: "supergroup" },
+        text: "hello @testbot",
+      },
+    });
+    expect(mentioned.send).toHaveBeenCalledTimes(1);
+    expect(mentioned.send.mock.calls[0]![0]).toBe("-1001::");
+    expect(mentioned.send.mock.calls[0]![1]).toMatchObject({ state: { conversationId: null } });
+
+    const replied = await firePost(channel, {
+      message: {
+        message_id: 13,
+        from: { id: 42, is_bot: false },
+        chat: { id: -1001, type: "supergroup" },
+        reply_to_message: {
+          message_id: 12,
+          from: { id: 99, is_bot: true, username: "testbot" },
+          chat: { id: -1001, type: "supergroup" },
+        },
+        text: "follow-up",
+      },
+    });
+    expect(replied.send).toHaveBeenCalledTimes(1);
+    expect(replied.send.mock.calls[0]![0]).toBe("-1001::");
+    expect(replied.send.mock.calls[0]![1]).toMatchObject({
+      inputResponses: [{ requestId: "telegram_reply:12", text: "follow-up" }],
+      message: "follow-up",
+      state: { chatId: "-1001", conversationId: null },
+    });
+
+    const inTopic = await firePost(channel, {
+      message: {
+        message_id: 14,
+        message_thread_id: 7,
+        is_topic_message: true,
+        from: { id: 42, is_bot: false },
+        chat: { id: -1001, type: "supergroup" },
+        text: "hello @testbot",
+      },
+    });
+    expect(inTopic.send.mock.calls[0]![0]).toBe("-1001:7:");
+
+    const inReplyChain = await firePost(channel, {
+      message: {
+        message_id: 15,
+        message_thread_id: 11,
+        from: { id: 42, is_bot: false },
+        chat: { id: -1001, type: "supergroup" },
+        reply_to_message: {
+          message_id: 11,
+          from: { id: 42, is_bot: false },
+          chat: { id: -1001, type: "supergroup" },
+        },
+        text: "hello @testbot",
+      },
+    });
+    expect(inReplyChain.send.mock.calls[0]![0]).toBe("-1001::");
+    expect(inReplyChain.send.mock.calls[0]![1]).toMatchObject({ state: { messageThreadId: null } });
+
+    const callback = await firePost(channel, {
+      callback_query: {
+        id: "cb1",
+        from: { id: 42, is_bot: false },
+        data: "eve:0",
+        message: {
+          message_id: 55,
+          chat: { id: -1001, type: "supergroup" },
+        },
+      },
+    });
+    expect(callback.send).toHaveBeenCalledWith("-1001::", expect.anything());
+  });
+
   it("delivers Telegram callback queries as compact HITL input responses", async () => {
     const channel = telegramChannel({
       api: { fetch: fakeTelegramFetch() },
@@ -259,7 +342,7 @@ describe("telegramChannel() inbound route", () => {
     });
 
     expect(send).toHaveBeenCalledWith(
-      "-1001::55",
+      "-1001::",
       expect.objectContaining({
         auth: null,
         inputResponses: [{ optionId: "selected", requestId: "telegram_callback:eve:0" }],
@@ -740,7 +823,7 @@ describe("telegramChannel() default event handlers", () => {
     expect(ctx.state.conversationId).toBeNull();
   });
 
-  it("hydrates unknown group message posts and re-keys to the posted message id", async () => {
+  it("keeps group message posts chat-wide", async () => {
     const fetchMock = vi.fn().mockResolvedValue(
       new Response(
         JSON.stringify({
@@ -769,9 +852,9 @@ describe("telegramChannel() default event handlers", () => {
       ctx,
     );
 
-    expect(writes).toContainEqual(["eve.continuationToken", "telegram:-1001::77"]);
+    expect(writes).not.toContainEqual(["eve.continuationToken", "telegram:-1001::77"]);
     expect(ctx.state.chatType).toBe("supergroup");
-    expect(ctx.state.conversationId).toBe("77");
+    expect(ctx.state.conversationId).toBeNull();
   });
 
   it("preserves explicit conversation ids after Telegram identifies a private chat", async () => {
@@ -808,7 +891,7 @@ describe("telegramChannel() default event handlers", () => {
     expect(ctx.state.conversationId).toBe("caller-selected");
   });
 
-  it("group message posts re-key the session to the posted message id", async () => {
+  it("preserves explicit conversation ids on group message posts", async () => {
     const fetchMock = vi
       .fn()
       .mockResolvedValue(
@@ -834,8 +917,8 @@ describe("telegramChannel() default event handlers", () => {
       ctx,
     );
 
-    expect(writes).toContainEqual(["eve.continuationToken", "telegram:-1001::77"]);
-    expect(ctx.state.conversationId).toBe("77");
+    expect(writes).not.toContainEqual(["eve.continuationToken", "telegram:-1001::77"]);
+    expect(ctx.state.conversationId).toBe("10");
   });
 });
 
@@ -919,7 +1002,7 @@ describe("telegramChannel().receive", () => {
     );
   });
 
-  it("anchors group initialMessage sessions under Telegram's message id", async () => {
+  it("keeps group initialMessage sessions chat-wide", async () => {
     for (const chatType of ["group", "supergroup"] as const) {
       const fetchMock = vi.fn().mockResolvedValue(
         new Response(
@@ -947,13 +1030,13 @@ describe("telegramChannel().receive", () => {
       );
 
       expect(send).toHaveBeenCalledWith(
-        "-1001::88",
+        "-1001::",
         expect.objectContaining({
           message: "run",
           state: expect.objectContaining({
             chatId: "-1001",
             chatType,
-            conversationId: "88",
+            conversationId: null,
           }),
         }),
       );

packages/eve/src/public/channels/telegram/telegramChannel.ts

@@ -1,6 +1,11 @@
 import type { TelegramInstrumentationMetadata } from "#public/channels/telegram/index.js";
 import { defaultDeliverResult, type ChannelAdapterContext } from "#channel/adapter.js";
-import type { ChannelFrom, ChannelResolveSession } from "#channel/channel-operations.js";
+import {
+  INTERNAL_CHANNEL_DELIVER,
+  type ChannelFrom,
+  type ChannelResolveSession,
+  type InternalChannelSource,
+} from "#channel/channel-operations.js";
 import type { SessionHandle } from "#channel/session.js";
 import type { DeliverPayload, SessionAuthContext, TurnPolicy } from "#channel/types.js";
 import type { SessionContext } from "#public/definitions/callback-context.js";
@@ -16,7 +21,6 @@ import {
   sendTelegramChatAction,
   sendTelegramMessage,
   splitTelegramMessageText,
-  telegramContinuationToken,
   type TelegramApiOptions,
   type TelegramApiResponse,
   type TelegramCredentials,
@@ -96,7 +100,7 @@ export interface TelegramChannelState extends TelegramHitlState {
   chatId: string | null;
   /** Telegram chat type, when known from an inbound update. */
   chatType: TelegramChatType | null;
-  /** Group/supergroup conversation anchor message id. */
+  /** Group/supergroup conversation id, when a proactive target pins one. */
   conversationId: string | null;
   /** Forum topic id, when known. */
   messageThreadId: number | null;
@@ -113,7 +117,7 @@ export interface TelegramChannelCredentials extends TelegramCredentials {
   readonly webhookVerifier?: TelegramWebhookVerifier;
 }
 
-/** Target for `receive(telegram, { target })` proactive sessions. `chatId` is required. `conversationId` resumes an existing thread; `initialMessage` posts a seed message and starts a new thread from it. The two are mutually exclusive: supplying both throws. */
+/** Target for `receive(telegram, { target })` proactive sessions. `chatId` is required. `conversationId` pins the session to a caller-selected id instead of the chat-wide session; inbound messages never route to it. `initialMessage` posts a seed message before the session starts; in groups the session stays chat-wide. The two are mutually exclusive: supplying both throws. */
 export interface TelegramReceiveTarget {
   readonly chatId: number | string;
   readonly conversationId?: number | string;
@@ -368,21 +372,9 @@ function buildTelegramHandle(input: {
   const credentials = input.config.credentials;
 
   function anchor(posted: TelegramMessageResult): void {
-    const chatType = state.chatType ?? posted.chatType ?? null;
     if (state.chatType === null && posted.chatType !== undefined) {
       state.chatType = posted.chatType;
     }
-    if (!posted.id || !shouldAnchorTelegramConversation(chatType)) return;
-    state.conversationId = posted.id;
-    if (state.chatId) {
-      input.session?.continuation?.rekey(
-        telegramContinuationToken({
-          chatId: state.chatId,
-          conversationId: posted.id,
-          messageThreadId: state.messageThreadId ?? undefined,
-        }),
-      );
-    }
   }
 
   async function sendOne(body: TelegramMessageBody): Promise<TelegramMessageResult> {
@@ -497,10 +489,6 @@ function buildTelegramHandle(input: {
   };
 }
 
-function shouldAnchorTelegramConversation(chatType: TelegramChatType | null): boolean {
-  return chatType === "group" || chatType === "supergroup";
-}
-
 async function postTelegramMessage(
   message: string | TelegramMessageBody,
   sendOne: (body: TelegramMessageBody) => Promise<TelegramMessageResult>,
@@ -600,10 +588,14 @@ async function dispatchMessage(input: {
         title: result.title,
       });
     } else {
-      await source.respond(replyInputResponses, {
-        auth: result.auth,
-        context: [contextBlock, ...channelContext],
-      });
+      await (source as InternalChannelSource<TelegramChannelState>)[INTERNAL_CHANNEL_DELIVER](
+        {
+          context: [contextBlock, ...channelContext],
+          inputResponses: replyInputResponses,
+          message: turnMessage,
+        },
+        { auth: result.auth, state, title: result.title },
+      );
     }
   } catch (error) {
     log.error("message delivery failed", { error });