[2/2] feat(eve): observe channel deliveries as session history

.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.

.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/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

docs/channels/telegram.mdx

@@ -41,8 +41,35 @@ 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.
 
+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`.
@@ -55,7 +82,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/extension-contracts/compatibility/channel/v17.ts

@@ -0,0 +1,3 @@
+import { disableRoute } from "#public/channels/index.js";
+
+export default disableRoute();

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"
+  ]
+}

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<TState = undefined>(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<TState = undefined>(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.",

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. */

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. */

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.",
     },

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

packages/eve/src/execution/deliver-payloads.ts

@@ -1,15 +1,19 @@
-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";
 
 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<DeliverPayload["task"]>;
 
 /** 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;
+}

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