From ee3eff2c3e8d00ed03332475cd4142a4dc959cad Mon Sep 17 00:00:00 2001 From: Taras Lukavyi Date: Fri, 4 Sep 2026 09:06:20 +0200 Subject: [PATCH 1/3] feat(eve)!: key Telegram group sessions chat-wide Telegram group and supergroup inbound messages keyed the session to the bot message being replied to, or to the mention itself. Every fresh mention therefore started a new session with one message of context, while private chats already keep one continuous session per chat. Groups now key to the chat plus forum topic, the same way private chats do. Mentions, replies to bot messages, callback queries, and outbound sends all resume that session. The message_thread_id that non-forum supergroups stamp on replies is a reply chain, not a topic, and does not split the session. Proactive targets can still pin a thread through conversationId. Related to #874 Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_015P6RBuagWTMzBmJLCiYKW9 Signed-off-by: Taras Lukavyi --- .../telegram-chat-wide-group-sessions.md | 5 + docs/channels/telegram.mdx | 8 +- .../src/public/channels/telegram/inbound.ts | 6 + .../eve/src/public/channels/telegram/state.ts | 24 ++-- .../channels/telegram/telegramChannel.test.ts | 103 ++++++++++++++++-- .../channels/telegram/telegramChannel.ts | 19 +--- 6 files changed, 124 insertions(+), 41 deletions(-) create mode 100644 .changeset/telegram-chat-wide-group-sessions.md diff --git a/.changeset/telegram-chat-wide-group-sessions.md b/.changeset/telegram-chat-wide-group-sessions.md new file mode 100644 index 0000000000..941abeb7d6 --- /dev/null +++ b/.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. diff --git a/docs/channels/telegram.mdx b/docs/channels/telegram.mdx index f67e5ac06a..0f29383d90 100644 --- a/docs/channels/telegram.mdx +++ b/docs/channels/telegram.mdx @@ -41,6 +41,12 @@ The model-visible `` 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 diff --git a/packages/eve/src/public/channels/telegram/inbound.ts b/packages/eve/src/public/channels/telegram/inbound.ts index 3c4ed06e8b..6499d9cdae 100644 --- a/packages/eve/src/public/channels/telegram/inbound.ts +++ b/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; @@ -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, diff --git a/packages/eve/src/public/channels/telegram/state.ts b/packages/eve/src/public/channels/telegram/state.ts index dfe29db50a..0d2c064bd8 100644 --- a/packages/eve/src/public/channels/telegram/state.ts +++ b/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; } diff --git a/packages/eve/src/public/channels/telegram/telegramChannel.test.ts b/packages/eve/src/public/channels/telegram/telegramChannel.test.ts index 06606f8721..257dd17f15 100644 --- a/packages/eve/src/public/channels/telegram/telegramChannel.test.ts +++ b/packages/eve/src/public/channels/telegram/telegramChannel.test.ts @@ -234,12 +234,93 @@ 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" }], + }); + + 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 +340,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 +821,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 +850,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 +889,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 +915,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 +1000,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 +1028,13 @@ describe("telegramChannel().receive", () => { ); expect(send).toHaveBeenCalledWith( - "-1001::88", + "-1001::", expect.objectContaining({ message: "run", state: expect.objectContaining({ chatId: "-1001", chatType, - conversationId: "88", + conversationId: null, }), }), ); diff --git a/packages/eve/src/public/channels/telegram/telegramChannel.ts b/packages/eve/src/public/channels/telegram/telegramChannel.ts index dafffc838e..37cd1cfe7d 100644 --- a/packages/eve/src/public/channels/telegram/telegramChannel.ts +++ b/packages/eve/src/public/channels/telegram/telegramChannel.ts @@ -16,7 +16,6 @@ import { sendTelegramChatAction, sendTelegramMessage, splitTelegramMessageText, - telegramContinuationToken, type TelegramApiOptions, type TelegramApiResponse, type TelegramCredentials, @@ -96,7 +95,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; @@ -368,21 +367,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 { @@ -497,10 +484,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, From 9114ab69d77fccd792425d0311444f8a62efd961 Mon Sep 17 00:00:00 2001 From: Taras Lukavyi Date: Fri, 4 Sep 2026 09:59:15 +0200 Subject: [PATCH 2/3] fix(eve): deliver Telegram bot replies as turn messages when no freeform prompt is pending A reply to a bot message went through respond() with only the synthetic telegram_reply: response. When that id was not a registered freeform prompt, the adapter resolved nothing, returned undefined, and the session re-parked without a turn, so the reply was dropped. The initial channel sent the text as message on the same delivery; #1597 replaced that with respond() and lost it. With chat-wide group sessions a reply to the bot is the natural follow-up, so carry the text as message again: the adapter still resolves a pending freeform answer first and otherwise runs the text as a normal turn. Pass state and title so a reply can start a session with a usable chat id when none owns the address. Also correct the TelegramReceiveTarget docstring: conversationId pins a caller-selected id that inbound messages never route to, and initialMessage no longer starts a separate thread in groups. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_015P6RBuagWTMzBmJLCiYKW9 Signed-off-by: Taras Lukavyi --- .../channels/telegram/telegramChannel.test.ts | 2 ++ .../channels/telegram/telegramChannel.ts | 21 +++++++++++++------ 2 files changed, 17 insertions(+), 6 deletions(-) diff --git a/packages/eve/src/public/channels/telegram/telegramChannel.test.ts b/packages/eve/src/public/channels/telegram/telegramChannel.test.ts index 257dd17f15..16e0e5f1c3 100644 --- a/packages/eve/src/public/channels/telegram/telegramChannel.test.ts +++ b/packages/eve/src/public/channels/telegram/telegramChannel.test.ts @@ -276,6 +276,8 @@ describe("telegramChannel() inbound route", () => { 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, { diff --git a/packages/eve/src/public/channels/telegram/telegramChannel.ts b/packages/eve/src/public/channels/telegram/telegramChannel.ts index 37cd1cfe7d..4365231050 100644 --- a/packages/eve/src/public/channels/telegram/telegramChannel.ts +++ b/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"; @@ -112,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; @@ -583,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)[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 }); From 98a3ecb5bfd777ea5f67fe9376ce84c486d7df5c Mon Sep 17 00:00:00 2001 From: Taras Lukavyi Date: Fri, 4 Sep 2026 09:48:58 +0200 Subject: [PATCH 3/3] feat(eve): observe channel deliveries as session history A group bot only ever sees the messages that address it, so even with one session per chat it cannot follow the conversation that happens between mentions. Slack has threadContext for this; Telegram had nothing. Add observe: true to channel deliveries. An observed delivery appends its message and context to the session as history without running a turn: eve buffers it on the parked or active session, never steers, never starts a session, and folds the buffered messages in arrival order into the next delivery that does run a turn. The observed backlog is capped at 256 payloads, oldest dropped first. Task deliveries settle on their own and leave observed history waiting for a channel turn. The Telegram channel exposes it as observe on the onMessage result and exports isTelegramBotMentioned so hosts can gate on it. Related to #874 Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_015P6RBuagWTMzBmJLCiYKW9 Signed-off-by: Taras Lukavyi --- .changeset/observe-deliveries.md | 5 ++ docs/channels/overview.mdx | 6 ++ docs/channels/telegram.mdx | 21 +++++ .../compatibility/channel/v17.ts | 3 + .../reports/channel/v18.json | 21 +++++ packages/eve/src/channel/channel-address.ts | 6 +- packages/eve/src/channel/routes.ts | 8 ++ packages/eve/src/channel/types.ts | 2 + .../src/compiler/extension-compatibility.ts | 4 +- .../src/execution/deliver-payloads.test.ts | 58 +++++++++++++- .../eve/src/execution/deliver-payloads.ts | 45 ++++++++++- .../execution/parked-delivery-wait.test.ts | 80 +++++++++++++++++++ .../eve/src/execution/parked-delivery-wait.ts | 35 ++++++-- .../execution/turn-control-receiver.test.ts | 19 +++++ .../src/execution/turn-control-receiver.ts | 8 +- .../eve/src/public/channels/telegram/index.ts | 2 +- .../channels/telegram/telegramChannel.test.ts | 42 ++++++++++ .../channels/telegram/telegramChannel.ts | 14 +++- 18 files changed, 362 insertions(+), 17 deletions(-) create mode 100644 .changeset/observe-deliveries.md create mode 100644 packages/eve/extension-contracts/compatibility/channel/v17.ts create mode 100644 packages/eve/extension-contracts/reports/channel/v18.json diff --git a/.changeset/observe-deliveries.md b/.changeset/observe-deliveries.md new file mode 100644 index 0000000000..d51a88ea88 --- /dev/null +++ b/.changeset/observe-deliveries.md @@ -0,0 +1,5 @@ +--- +"eve": minor +--- + +Add `observe: true` to channel deliveries. An observed delivery appends its message and context to the session as history without running a turn; eve buffers it on the parked or active session and folds it into the next delivery that does run a turn. The Telegram channel exposes it as `observe` on the `onMessage` result, so a group bot can follow the conversation between mentions and answer the next mention with that context. `isTelegramBotMentioned` is now exported for custom `onMessage` gating. diff --git a/docs/channels/overview.mdx b/docs/channels/overview.mdx index 5280d9c55e..0a6aec8a09 100644 --- a/docs/channels/overview.mdx +++ b/docs/channels/overview.mdx @@ -32,6 +32,12 @@ export default defineChannel({ Channel admission still runs first. Ignored mentions, rejected signatures, duplicates, and any other dropped platform events never affect the active turn. +## Observed messages + +Pass `observe: true` on a delivery to append a message to the session as history without running a turn. eve buffers observed deliveries on the session, whether it is parked or mid-turn, and folds them in arrival order into the next delivery that does run a turn, so the model sees them as part of that turn's input. An observed delivery never steers an active turn and never starts a session: when no session owns the address, it throws `RuntimeNoActiveSessionError` and the message is dropped. The buffer keeps the most recent 256 observed payloads. + +Use it for group surfaces where the bot should follow the conversation without answering every message. The Telegram channel exposes it as `observe` on the `onMessage` result; custom channels reach it through `deliver({ message, observe: true }, options)`. + Each channel has its own provider terms, data flow, auth model, and user-consent expectations. Before sending non-public, sensitive, regulated, or production data through a channel, confirm that the channel provider and your configured scopes, signature checks, route auth, and delivery behavior are appropriate for your use case. ## Where channels live diff --git a/docs/channels/telegram.mdx b/docs/channels/telegram.mdx index 0f29383d90..5380fe38a1 100644 --- a/docs/channels/telegram.mdx +++ b/docs/channels/telegram.mdx @@ -49,6 +49,27 @@ To pin a proactive session to a specific thread instead, pass `conversationId` o 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. +Return `observe: true` to keep a group message as session history without answering it. The next mention then runs with the conversation that happened in between. Observed messages need a session to attach to, so the chat's first mention still starts one and earlier messages are dropped. Telegram delivers unaddressed group messages to the bot only when privacy mode is off in BotFather (`/setprivacy`) or the bot is a group admin. + +```ts +import { + defaultTelegramAuth, + isTelegramBotMentioned, + telegramChannel, +} from "eve/channels/telegram"; + +telegramChannel({ + botUsername: "my_bot", + onMessage: (ctx, message) => { + if (message.chat.type === "private") return { auth: defaultTelegramAuth(message) }; + const addressed = + isTelegramBotMentioned(message, ctx.telegram.botUsername) || + message.replyToMessage?.from?.isBot === true; + return { auth: defaultTelegramAuth(message), observe: !addressed }; + }, +}); +``` + ### Delivery The default `message.completed` handler sends plain text via `sendMessage`. It passes no `parse_mode`, so any Markdown shows up literally. Replies longer than Telegram's 4096-char limit are split across messages. Custom handlers use `channel.telegram`. diff --git a/packages/eve/extension-contracts/compatibility/channel/v17.ts b/packages/eve/extension-contracts/compatibility/channel/v17.ts new file mode 100644 index 0000000000..67e58f0bbe --- /dev/null +++ b/packages/eve/extension-contracts/compatibility/channel/v17.ts @@ -0,0 +1,3 @@ +import { disableRoute } from "#public/channels/index.js"; + +export default disableRoute(); diff --git a/packages/eve/extension-contracts/reports/channel/v18.json b/packages/eve/extension-contracts/reports/channel/v18.json new file mode 100644 index 0000000000..d00d8bf711 --- /dev/null +++ b/packages/eve/extension-contracts/reports/channel/v18.json @@ -0,0 +1,21 @@ +{ + "kind": "eve-extension-capability-contract", + "capability": "channel", + "epoch": 18, + "sha256": "fd469133dfdc1b14cae01772e01b10ddd4e402eafe3de3891535fe7e77a37342", + "exports": [ + "DELETE", + "GET", + "HEAD", + "OPTIONS", + "PATCH", + "POST", + "PUT", + "WS", + "createWebSocketUpgradeServer", + "defineChannel", + "disableRoute", + "isChannel", + "isDisabledRouteSentinel" + ] +} diff --git a/packages/eve/src/channel/channel-address.ts b/packages/eve/src/channel/channel-address.ts index 4e9df5d576..e949409f11 100644 --- a/packages/eve/src/channel/channel-address.ts +++ b/packages/eve/src/channel/channel-address.ts @@ -22,6 +22,7 @@ import type { TurnPolicy, } from "#channel/types.js"; import { DEFAULT_TURN_POLICY } from "#channel/types.js"; +import { RuntimeNoActiveSessionError } from "#execution/runtime-errors.js"; import { isReservedSessionCommandToken } from "#execution/session-command-token.js"; import type { RunMode } from "#shared/run-mode.js"; @@ -100,7 +101,7 @@ export function createChannelAddress(input: { }, requestId: metadata.requestId, turnPolicy: - payload.message === undefined + payload.message === undefined || payload.observe === true ? undefined : (options.turnPolicy ?? input.turnPolicy ?? DEFAULT_TURN_POLICY), }; @@ -121,6 +122,9 @@ export function createChannelAddress(input: { const existing = await dispatch(); if (existing !== undefined) return existing; + if (payload.observe === true) { + throw new RuntimeNoActiveSessionError(namespacedToken); + } if (payload.inputResponses && payload.inputResponses.length > 0) { throw new Error( "Cannot deliver inputResponses — the target session was not found via continuation token.", diff --git a/packages/eve/src/channel/routes.ts b/packages/eve/src/channel/routes.ts index 15122cdec6..37d0b81aeb 100644 --- a/packages/eve/src/channel/routes.ts +++ b/packages/eve/src/channel/routes.ts @@ -49,6 +49,14 @@ export interface SendPayload { * schema in conversation and task mode; mode only decides failure behavior. */ readonly outputSchema?: JsonObject; + /** + * Appends `message` and `context` to the session as history without running + * a turn. eve buffers observed deliveries on the session and folds them, in + * arrival order, into the next delivery that does run a turn. Requires a + * session that already owns the address: an observed delivery never starts + * one and never steers an active turn. + */ + readonly observe?: boolean; } /** Attaches an I/O-free handle to one exact durable session ID. */ diff --git a/packages/eve/src/channel/types.ts b/packages/eve/src/channel/types.ts index 9f2b601807..8794ea0193 100644 --- a/packages/eve/src/channel/types.ts +++ b/packages/eve/src/channel/types.ts @@ -179,6 +179,8 @@ export interface DeliverPayload { readonly message?: string | UserContent; readonly context?: readonly string[]; readonly outputSchema?: JsonObject; + /** History-only delivery; see `SendPayload.observe`. */ + readonly observe?: boolean; /** Framework-only task envelopes consumed before adapter/model delivery. */ readonly task?: { /** Task HITL input-request batches for the parent's pre-model router. */ diff --git a/packages/eve/src/compiler/extension-compatibility.ts b/packages/eve/src/compiler/extension-compatibility.ts index 50593c4d08..3367363bf8 100644 --- a/packages/eve/src/compiler/extension-compatibility.ts +++ b/packages/eve/src/compiler/extension-compatibility.ts @@ -38,8 +38,8 @@ const EXTENSION_CAPABILITY_CONTRACTS = { }, }, channel: { - current: 17, - supported: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 13, 14, 15, 16, 17], + current: 18, + supported: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 13, 14, 15, 16, 17, 18], dropped: { 12: "Message and reasoning append events now expose deltas instead of cumulative snapshots.", }, diff --git a/packages/eve/src/execution/deliver-payloads.test.ts b/packages/eve/src/execution/deliver-payloads.test.ts index a1f8b1e824..5a56252d67 100644 --- a/packages/eve/src/execution/deliver-payloads.test.ts +++ b/packages/eve/src/execution/deliver-payloads.test.ts @@ -1,5 +1,11 @@ import { describe, expect, it } from "vitest"; -import { coalesceDeliverPayloads } from "#execution/deliver-payloads.js"; +import type { DeliverHookPayload } from "#channel/types.js"; +import { + bufferObservedDelivery, + coalesceDeliverPayloads, + hasAddressedDelivery, + isObserveOnlyDelivery, +} from "#execution/deliver-payloads.js"; const FIRST_MESSAGE = "Please summarize the synthetic release checklist before the rollout."; const SECOND_MESSAGE = "Proceed after the synthetic health check passes."; @@ -62,6 +68,22 @@ describe("coalesceDeliverPayloads", () => { }); }); + it("keeps observe only when every payload was observed", () => { + expect( + coalesceDeliverPayloads([ + { message: "U1: had a rough week", observe: true }, + { message: "U2: same here", observe: true }, + ]), + ).toEqual({ message: "U1: had a rough week\n\nU2: same here", observe: true }); + + expect( + coalesceDeliverPayloads([ + { message: "U1: had a rough week", observe: true }, + { message: "@bot what do you think?" }, + ]), + ).toEqual({ message: "U1: had a rough week\n\n@bot what do you think?" }); + }); + it("preserves task agent requests and authorization events across queued payloads", () => { const agentRequests = [ { @@ -112,3 +134,37 @@ describe("coalesceDeliverPayloads", () => { ).toEqual({ task: { agentRequests, authorizationEvents } }); }); }); + +describe("bufferObservedDelivery", () => { + function observed(message: string): DeliverHookPayload { + return { kind: "deliver", payloads: [{ message, observe: true }] }; + } + + it("buffers observe-only deliveries and leaves addressed ones to the caller", () => { + const buffer: DeliverHookPayload[] = []; + const addressed: DeliverHookPayload = { kind: "deliver", payloads: [{ message: "@bot hi" }] }; + + expect(bufferObservedDelivery(buffer, observed("aside"))).toBe(true); + expect(bufferObservedDelivery(buffer, addressed)).toBe(false); + + expect(buffer).toEqual([observed("aside")]); + expect(isObserveOnlyDelivery(observed("aside"))).toBe(true); + expect(isObserveOnlyDelivery(addressed)).toBe(false); + expect(hasAddressedDelivery(buffer)).toBe(false); + expect(hasAddressedDelivery([...buffer, addressed])).toBe(true); + }); + + it("drops the oldest observed deliveries past the buffer limit", () => { + const buffer: DeliverHookPayload[] = [ + { kind: "deliver", payloads: [{ inputResponses: [{ optionId: "yes", requestId: "r1" }] }] }, + ]; + for (let index = 0; index < 257; index += 1) { + bufferObservedDelivery(buffer, observed(`message ${index}`)); + } + + expect(buffer).toHaveLength(257); + expect(buffer[0]?.payloads[0]?.inputResponses).toBeDefined(); + expect(buffer[1]).toEqual(observed("message 1")); + expect(buffer.at(-1)).toEqual(observed("message 256")); + }); +}); diff --git a/packages/eve/src/execution/deliver-payloads.ts b/packages/eve/src/execution/deliver-payloads.ts index d674aa80da..4a85f5dd99 100644 --- a/packages/eve/src/execution/deliver-payloads.ts +++ b/packages/eve/src/execution/deliver-payloads.ts @@ -1,4 +1,4 @@ -import type { DeliverPayload } from "#channel/types.js"; +import type { DeliverHookPayload, DeliverPayload } from "#channel/types.js"; import { coalesceTurnInputs } from "#harness/messages.js"; import type { StepInput } from "#harness/types.js"; @@ -6,10 +6,14 @@ const COALESCED_DELIVER_FIELDS = [ "context", "inputResponses", "message", + "observe", "outputSchema", "task", ] as const; +/** Observed payloads a parked or active session keeps before dropping the oldest. */ +const OBSERVED_PAYLOAD_BUFFER_LIMIT = 256; + type TaskEnvelope = NonNullable; /** Coalesces channel payloads while preserving turn input and adapter-specific fields. */ @@ -47,6 +51,45 @@ export function coalesceDeliverPayloads(payloads: readonly DeliverPayload[]): De if (authorizationEvents.length > 0) task.authorizationEvents = authorizationEvents; if (views.length > 0) task.views = views; if (Object.keys(task).length > 0) merged.task = task; + if (payloads.every((payload) => payload.observe === true)) merged.observe = true; return Object.assign(merged, turnInput); } + +/** True when every payload in the delivery is history-only. */ +export function isObserveOnlyDelivery(delivery: DeliverHookPayload): boolean { + return ( + delivery.payloads.length > 0 && delivery.payloads.every((payload) => payload.observe === true) + ); +} + +/** True when at least one buffered delivery should run a turn. */ +export function hasAddressedDelivery(deliveries: readonly DeliverHookPayload[]): boolean { + return deliveries.some((delivery) => !isObserveOnlyDelivery(delivery)); +} + +/** + * Buffers an observe-only delivery without waking the session. Returns `false` + * for deliveries that should run a turn so the caller handles them as before. + * The observed backlog is capped; the oldest observed deliveries drop first. + */ +export function bufferObservedDelivery( + buffer: DeliverHookPayload[], + delivery: DeliverHookPayload, +): boolean { + if (!isObserveOnlyDelivery(delivery)) return false; + buffer.push(delivery); + + let excess = + buffer.reduce( + (count, entry) => count + (isObserveOnlyDelivery(entry) ? entry.payloads.length : 0), + 0, + ) - OBSERVED_PAYLOAD_BUFFER_LIMIT; + while (excess > 0) { + const oldest = buffer.findIndex(isObserveOnlyDelivery); + if (oldest < 0) break; + const [dropped] = buffer.splice(oldest, 1); + excess -= dropped?.payloads.length ?? 0; + } + return true; +} diff --git a/packages/eve/src/execution/parked-delivery-wait.test.ts b/packages/eve/src/execution/parked-delivery-wait.test.ts index 4200a647e9..8437d0628d 100644 --- a/packages/eve/src/execution/parked-delivery-wait.test.ts +++ b/packages/eve/src/execution/parked-delivery-wait.test.ts @@ -96,6 +96,13 @@ function messageRead(message: string): ScriptedRead { }; } +function observedRead(message: string): ScriptedRead { + return { + result: { done: false, value: { kind: "send", payload: { message, observe: true } } }, + source: "session", + }; +} + // Routing never runs in these tests: scripted reads stop at authorization // instructions or exhaust before any deliver-kind turn payload. const sessionState = { sessionId: "ses-parked-wait" } as DurableSessionState; @@ -184,6 +191,79 @@ describe("nextTurnDelivery", () => { ]); }); + it("stays parked on observed deliveries until an addressed one arrives", async () => { + const inbox = createMockInbox([ + observedRead("U1: had a rough week"), + observedRead("U2: same here"), + messageRead("@bot what do you think?"), + ]); + const bufferedDeliveries: DeliverHookPayload[] = []; + vi.mocked(routeDeliverToChildren).mockImplementation(async ({ delivery }) => ({ + kind: "continue", + remainder: delivery, + serializedContext: {}, + sessionState, + })); + + const next = await nextTurnDelivery({ + ...waitInput(inbox), + awaitAuthorizationCallbacks: false, + bufferedDeliveries, + }); + + expect(next.kind).toBe("turn"); + if (next.kind !== "turn") throw new Error("unreachable"); + expect(next.delivery.payloads).toEqual([ + { message: "U1: had a rough week", observe: true }, + { message: "U2: same here", observe: true }, + { message: "@bot what do you think?" }, + ]); + expect(bufferedDeliveries).toEqual([]); + }); + + it("does not wake a parked session for buffered observed deliveries alone", async () => { + const inbox = createMockInbox([authorizationRead()]); + const bufferedDeliveries: DeliverHookPayload[] = [ + { kind: "deliver", payloads: [{ message: "aside", observe: true }] }, + ]; + + const next = await nextTurnDelivery({ ...waitInput(inbox), bufferedDeliveries }); + + expect(next.kind).toBe("authorization"); + expect(bufferedDeliveries).toHaveLength(1); + }); + + it("leaves observed history behind when a task delivery runs the turn", async () => { + const inbox = createMockInbox([]); + const observed: DeliverHookPayload = { + kind: "deliver", + payloads: [{ message: "aside", observe: true }], + }; + const task: DeliverHookPayload = { + kind: "deliver", + payloads: [{ task: { views: [] } }], + taskDeliveryId: "task-1:done", + }; + const bufferedDeliveries: DeliverHookPayload[] = [observed, task]; + vi.mocked(routeDeliverToChildren).mockImplementation(async ({ delivery }) => ({ + kind: "continue", + remainder: delivery, + serializedContext: {}, + sessionState, + })); + + const next = await nextTurnDelivery({ + ...waitInput(inbox), + awaitAuthorizationCallbacks: false, + bufferedDeliveries, + }); + + expect(next.kind).toBe("turn"); + if (next.kind !== "turn") throw new Error("unreachable"); + expect(next.delivery.taskDeliveryId).toBe("task-1:done"); + expect(bufferedDeliveries).toEqual([observed]); + }); + it("reports a closed authorization hook", async () => { const inbox = createMockInbox([ { result: { done: true, value: undefined }, source: "authorization" }, diff --git a/packages/eve/src/execution/parked-delivery-wait.ts b/packages/eve/src/execution/parked-delivery-wait.ts index 6389215fb3..7ed76b617d 100644 --- a/packages/eve/src/execution/parked-delivery-wait.ts +++ b/packages/eve/src/execution/parked-delivery-wait.ts @@ -9,6 +9,11 @@ import { SessionInboxWireError, type DecodedSessionInbox, } from "#execution/wire/session-inbox-wire.js"; +import { + bufferObservedDelivery, + hasAddressedDelivery, + isObserveOnlyDelivery, +} from "#execution/deliver-payloads.js"; import { coalesceDeliveries } from "#harness/messages.js"; type NextSessionAction = @@ -154,16 +159,14 @@ async function waitForNextSessionAction(input: { return { kind: pendingSessionControl }; } - while ( - input.bufferedDeliveries[0] !== undefined && - isCancelledTaskDelivery(input.bufferedDeliveries[0], input.cancelledTaskIds) - ) { - input.bufferedDeliveries.shift(); - } + const live = input.bufferedDeliveries.filter( + (delivery) => !isCancelledTaskDelivery(delivery, input.cancelledTaskIds), + ); + input.bufferedDeliveries.splice(0, input.bufferedDeliveries.length, ...live); if ( input.deferDeliveries !== true && !input.commandInbox.hasReadyAuthorization() && - input.bufferedDeliveries.length > 0 + hasAddressedDelivery(input.bufferedDeliveries) ) { return { delivery: takeBufferedTurnDelivery(input.bufferedDeliveries), @@ -242,10 +245,16 @@ async function waitForNextSessionAction(input: { input.seenTaskDeliveries.add(deliveryId); } + // Observed deliveries only ever ride along with the next addressed one. + if (bufferObservedDelivery(input.bufferedDeliveries, decoded)) continue; if (input.deferDeliveries === true) { input.bufferedDeliveries.push(decoded); continue; } + if (input.bufferedDeliveries.length > 0) { + input.bufferedDeliveries.push(decoded); + return { delivery: takeBufferedTurnDelivery(input.bufferedDeliveries), kind: "delivery" }; + } return { delivery: decoded, kind: "delivery" }; } } @@ -268,12 +277,22 @@ function isCancelledTaskDeliveryId( } function takeBufferedTurnDelivery(bufferedDeliveries: DeliverHookPayload[]): DeliverHookPayload { + const observed: DeliverHookPayload[] = []; + while (bufferedDeliveries[0] !== undefined && isObserveOnlyDelivery(bufferedDeliveries[0])) { + observed.push(bufferedDeliveries.shift()!); + } const first = bufferedDeliveries.shift(); if (first === undefined) { + bufferedDeliveries.unshift(...observed); throw new Error("Cannot take a turn delivery from an empty buffer."); } + // Task deliveries settle on their own; observed history waits for a channel turn. + if (first.taskDeliveryId !== undefined) { + bufferedDeliveries.unshift(...observed); + return coalesceDeliveries([first]); + } - const turnDeliveries = [first]; + const turnDeliveries = [...observed, first]; let caller = first.caller; while (bufferedDeliveries.length > 0) { const next = bufferedDeliveries[0]; diff --git a/packages/eve/src/execution/turn-control-receiver.test.ts b/packages/eve/src/execution/turn-control-receiver.test.ts index 1fa2788817..32e4d15766 100644 --- a/packages/eve/src/execution/turn-control-receiver.test.ts +++ b/packages/eve/src/execution/turn-control-receiver.test.ts @@ -61,6 +61,25 @@ describe("TurnControlReceiver", () => { expect(bufferedDeliveries).toEqual([]); }); + it("buffers observed deliveries without steering or forwarding them", async () => { + const observed: DeliverHookPayload = { + kind: "deliver", + payloads: [{ message: "U1: had a rough week", observe: true }], + turnPolicy: "steer", + }; + installControlHook([parkResult()], true); + const bufferedDeliveries: DeliverHookPayload[] = []; + + const action = await runReceiver(bufferedDeliveries, { + commandInbox: createCommandInbox([observed]), + }); + + expect(action.kind).toBe("park"); + expect(forwardTurnCancellationStep).not.toHaveBeenCalled(); + expect(forwardTurnDeliveryStep).not.toHaveBeenCalled(); + expect(bufferedDeliveries).toEqual([observed]); + }); + it("re-buffers the outstanding delivery when the turn cancels its request", async () => { const delivery: DeliverHookPayload = { kind: "deliver", diff --git a/packages/eve/src/execution/turn-control-receiver.ts b/packages/eve/src/execution/turn-control-receiver.ts index 653158781f..d8b1baa846 100644 --- a/packages/eve/src/execution/turn-control-receiver.ts +++ b/packages/eve/src/execution/turn-control-receiver.ts @@ -1,6 +1,7 @@ import { createHook, type Hook } from "#compiled/@workflow/core/index.js"; import type { DeliverHookPayload } from "#channel/types.js"; +import { bufferObservedDelivery, isObserveOnlyDelivery } from "#execution/deliver-payloads.js"; import { cancelAllIndexedSessionTasksStep } from "#execution/cancel-indexed-session-tasks-step.js"; import { forwardTurnCancellationStep } from "#execution/forward-turn-cancellation-step.js"; import type { TurnControlPayload } from "#execution/turn-control-protocol.js"; @@ -138,6 +139,7 @@ export class TurnControlReceiver { } private async bufferDelivery(delivery: DeliverHookPayload): Promise { + if (bufferObservedDelivery(this.bufferedDeliveries, delivery)) return; this.bufferedDeliveries.push(delivery); if (delivery.turnPolicy !== "steer" || !deliveryHasMessage(delivery)) return; @@ -267,7 +269,7 @@ export class TurnControlReceiver { } if (decoded.kind === "deliver") { if (!this.acceptTaskDelivery(decoded)) continue; - if (deliveryHasMessage(decoded)) { + if (isObserveOnlyDelivery(decoded) || deliveryHasMessage(decoded)) { await this.bufferDelivery(decoded); } else { delivery = decoded; @@ -298,7 +300,9 @@ export class TurnControlReceiver { } private takeInputResponseDelivery(): DeliverHookPayload | undefined { - const index = this.bufferedDeliveries.findIndex((delivery) => !deliveryHasMessage(delivery)); + const index = this.bufferedDeliveries.findIndex( + (delivery) => !isObserveOnlyDelivery(delivery) && !deliveryHasMessage(delivery), + ); if (index === -1) return undefined; return this.bufferedDeliveries.splice(index, 1)[0]; } diff --git a/packages/eve/src/public/channels/telegram/index.ts b/packages/eve/src/public/channels/telegram/index.ts index 843c34b3e8..392a9af8f8 100644 --- a/packages/eve/src/public/channels/telegram/index.ts +++ b/packages/eve/src/public/channels/telegram/index.ts @@ -87,7 +87,7 @@ export { createTelegramFileUrl, } from "#public/channels/telegram/attachments.js"; -export { defaultTelegramAuth } from "#public/channels/telegram/defaults.js"; +export { defaultTelegramAuth, isTelegramBotMentioned } from "#public/channels/telegram/defaults.js"; export { resolveTelegramWebhookSecretToken, diff --git a/packages/eve/src/public/channels/telegram/telegramChannel.test.ts b/packages/eve/src/public/channels/telegram/telegramChannel.test.ts index 16e0e5f1c3..b9be5fc5a3 100644 --- a/packages/eve/src/public/channels/telegram/telegramChannel.test.ts +++ b/packages/eve/src/public/channels/telegram/telegramChannel.test.ts @@ -323,6 +323,48 @@ describe("telegramChannel() inbound route", () => { expect(callback.send).toHaveBeenCalledWith("-1001::", expect.anything()); }); + it("delivers observed group messages as history without a turn", async () => { + const channel = telegramChannel({ + api: { fetch: fakeTelegramFetch() }, + botUsername: "testbot", + credentials: { botToken: "bot-token", webhookSecretToken: SECRET }, + onMessage: (_ctx, message) => ({ + auth: null, + observe: !isTelegramBotMentioned(message, "testbot"), + }), + }); + + const aside = await firePost(channel, { + message: { + message_id: 20, + from: { id: 42, is_bot: false }, + chat: { id: -1001, type: "supergroup" }, + text: "had a rough week", + }, + }); + expect(aside.send).toHaveBeenCalledTimes(1); + expect(aside.send.mock.calls[0]![0]).toBe("-1001::"); + expect(aside.send.mock.calls[0]![1]).toMatchObject({ + message: "had a rough week", + observe: true, + state: { chatId: "-1001", conversationId: null }, + }); + expect((aside.send.mock.calls[0]![1] as { context: string[] }).context[0]).toContain( + "is_mentioned: false", + ); + + const mention = await firePost(channel, { + message: { + message_id: 21, + from: { id: 42, is_bot: false }, + chat: { id: -1001, type: "supergroup" }, + text: "@testbot what do you think?", + }, + }); + expect(mention.send.mock.calls[0]![0]).toBe("-1001::"); + expect(mention.send.mock.calls[0]![1]).not.toHaveProperty("observe"); + }); + it("delivers Telegram callback queries as compact HITL input responses", async () => { const channel = telegramChannel({ api: { fetch: fakeTelegramFetch() }, diff --git a/packages/eve/src/public/channels/telegram/telegramChannel.ts b/packages/eve/src/public/channels/telegram/telegramChannel.ts index 4365231050..8a464aee24 100644 --- a/packages/eve/src/public/channels/telegram/telegramChannel.ts +++ b/packages/eve/src/public/channels/telegram/telegramChannel.ts @@ -6,6 +6,7 @@ import { type ChannelResolveSession, type InternalChannelSource, } from "#channel/channel-operations.js"; +import { isRuntimeNoActiveSessionError } from "#execution/runtime-errors.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"; @@ -129,6 +130,8 @@ export interface TelegramReceiveTarget { export type TelegramInboundResult = { readonly auth: SessionAuthContext | null; readonly context?: readonly string[]; + /** Appends the message to the chat's session as history without running a turn. Dropped when the chat has no session yet. */ + readonly observe?: boolean; /** Overrides the workflow run title without changing the message sent to the model. */ readonly title?: string; } | null; @@ -580,7 +583,12 @@ async function dispatchMessage(input: { try { const source = input.from(telegramContinuationTokenFromState(state)); - if (replyInputResponses === undefined) { + if (result.observe === true) { + await (source as InternalChannelSource)[INTERNAL_CHANNEL_DELIVER]( + { context: [contextBlock, ...channelContext], message: turnMessage, observe: true }, + { auth: result.auth, state }, + ); + } else if (replyInputResponses === undefined) { await source.send(turnMessage, { auth: result.auth, context: [contextBlock, ...channelContext], @@ -598,6 +606,10 @@ async function dispatchMessage(input: { ); } } catch (error) { + if (result.observe === true && isRuntimeNoActiveSessionError(error)) { + log.debug("observed message dropped: chat has no session yet", { chatId: state.chatId }); + return; + } log.error("message delivery failed", { error }); } }