fix(telegram): buffered streaming mode for Pi token-level chunks

Pi's text_delta events can drop newlines and spaces from the model's
output. The assembled transcript from agent_end.messages preserves the
correct formatting. This change suppresses text_delta streaming and
emits the full assembled text instead.

For Telegram specifically, adds a 'buffered' streaming mode that
coalesces chunks via a 3-second debounce timer. This prevents each
token from becoming a separate Telegram message.

- Pi event bridge: suppress text_delta, emit assembled text at agent_end
- Telegram adapter: buffered mode with debounce, short-buffer skip
- Whitespace-only message guard for Telegram
- Shutdown flush to avoid losing in-flight buffered text
- Tests for buffered mode and updated streaming tail tests

Note: Pi's assembled transcript escapes nested markdown (e.g.
**bold and *italic*** becomes **\*\*bold and \*italic\*\*\*),
which is a Pi SDK bug that cannot be fixed in Archon.

Author: fezzzza

bun.lock

@@ -23,7 +23,7 @@
     "@slack/bolt": "^4.6.0",
     "discord.js": "^14.16.0",
     "grammy": "^1.36.0",
-    "telegramify-markdown": "^1.3.0"
+    "telegramify-markdown": "^1.3.3"
   },
   "peerDependencies": {
     "typescript": "^5.0.0"

packages/adapters/package.json

@@ -166,6 +166,47 @@ describe('TelegramAdapter', () => {
     });
   });
 
+  describe('buffered mode', () => {
+    let adapter: TelegramAdapter;
+    let mockSendMessage: Mock<() => Promise<void>>;
+
+    beforeEach(() => {
+      adapter = new TelegramAdapter('fake-token-for-testing', 'buffered');
+      mockSendMessage = mock(() => Promise.resolve());
+      (adapter.getBot().api as unknown as { sendMessage: Mock<() => Promise<void>> }).sendMessage =
+        mockSendMessage;
+    });
+
+    test('should report stream mode to orchestrator (not batch)', () => {
+      expect(adapter.getStreamingMode()).toBe('stream');
+    });
+
+    test('should coalesce rapid chunks into a single message', async () => {
+      // Rapid-fire chunks within the debounce window (total > BUFFER_MIN_FLUSH_LENGTH)
+      await adapter.sendMessage('12345', 'Hello from the buffered');
+      await adapter.sendMessage('12345', ' Telegram adapter test');
+      await adapter.sendMessage('12345', ' with enough content to flush!');
+
+      // Nothing sent yet — still buffering
+      expect(mockSendMessage).not.toHaveBeenCalled();
+
+      // Wait for debounce timer to fire (3000ms)
+      await new Promise(resolve => setTimeout(resolve, 3100));
+
+      // Should have sent one coalesced message
+      expect(mockSendMessage).toHaveBeenCalledTimes(1);
+      const call = mockSendMessage.mock.calls[0];
+      expect(call[0]).toBe(12345);
+      expect(call[1]).toContain('Hello from the buffered');
+    });
+
+    test('should skip whitespace-only chunks without flushing', async () => {
+      await adapter.sendMessage('12345', '\n');
+      await new Promise(resolve => setTimeout(resolve, 3100));
+      expect(mockSendMessage).not.toHaveBeenCalled();
+    });
+  });
+
   describe('getConversationId', () => {
     test('should return chat.id as string for private chat', () => {
       const adapter = new TelegramAdapter('fake-token-for-testing');

packages/adapters/src/chat/telegram/adapter.test.ts

@@ -19,16 +19,32 @@ function getLog(): ReturnType<typeof createLogger> {
 
 const MAX_LENGTH = 4096;
 
+/** Streaming mode for the Telegram adapter. */
+export type TelegramStreamingMode = 'stream' | 'batch' | 'buffered';
+
+/** Buffered mode: debounce interval (ms). */
+const BUFFER_FLUSH_MS = 3000;
+/** Buffered mode: skip flushing buffers shorter than this (likely a single token before a thinking pause). */
+const BUFFER_MIN_FLUSH_LENGTH = 50;
+
+/** State for a single chat's buffer. */
+interface BufferState {
+  text: string;
+  timer: ReturnType<typeof setTimeout> | null;
+}
+
 export class TelegramAdapter implements IPlatformAdapter {
   private bot: Bot;
-  private streamingMode: 'stream' | 'batch';
+  private mode: TelegramStreamingMode;
   private allowedUserIds: number[];
   private messageHandler: ((ctx: TelegramMessageContext) => Promise<void>) | null = null;
+  // Buffered mode: per-chat accumulation state
+  private buffers = new Map<string, BufferState>();
 
-  constructor(token: string, mode: 'stream' | 'batch' = 'stream') {
+  constructor(token: string, mode: TelegramStreamingMode = 'stream') {
     // grammY does not impose a handler timeout by default (unlike Telegraf's 90s limit)
     this.bot = new Bot(token);
-    this.streamingMode = mode;
+    this.mode = mode;
 
     // Parse Telegram user whitelist (optional - empty = open access)
     // Support both TELEGRAM_ALLOWED_USER_IDS and TELEGRAM_ALLOWED_USERS
@@ -54,14 +70,93 @@ export class TelegramAdapter implements IPlatformAdapter {
    *   (paragraphs rarely have formatting that spans across them)
    */
   async sendMessage(chatId: string, message: string, _metadata?: MessageMetadata): Promise<void> {
+    // Telegram rejects whitespace-only messages (400: text must be non-empty).
+    // Reasoning models (e.g. GLM-4.5-Air via Pi) can emit newline-only chunks
+    // during streaming — skip silently.
+    if (!message.trim()) return;
+
+    // Buffered mode: accumulate chunks and flush on debounce timer or size threshold.
+    // Recommended for providers that emit token-level chunks (e.g. Pi/z.ai with
+    // GLM-4.5-Air) where each token would otherwise become a separate Telegram message.
+    if (this.mode === 'buffered') {
+      this.bufferChunk(chatId, message);
+      return;
+    }
+
+    await this.deliverMessage(chatId, message);
+  }
+
+  /**
+   * Accumulate a chunk into the per-chat buffer. Flushes when
+   * BUFFER_FLUSH_MS elapses with no new chunks (end of AI response).
+   * Long responses are split at paragraph boundaries by deliverMessage.
+   */
+  private bufferChunk(chatId: string, chunk: string): void {
+    let state = this.buffers.get(chatId);
+    if (!state) {
+      state = { text: '', timer: null };
+      this.buffers.set(chatId, state);
+    }
+
+    // Append chunk and reset debounce timer
+    state.text += chunk;
+    if (state.timer) clearTimeout(state.timer);
+
+    state.timer = setTimeout(() => {
+      const current = this.buffers.get(chatId);
+      if (current && current.text.trim().length > 0) {
+        this.flushBuffer(chatId, current);
+      }
+    }, BUFFER_FLUSH_MS);
+  }
+
+  /** Flush a buffered chat's accumulated text and clean up state. */
+  private flushBuffer(chatId: string, state: BufferState, force = false): void {
+    if (state.timer) {
+      clearTimeout(state.timer);
+      state.timer = null;
+    }
+    const text = state.text;
+    this.buffers.delete(chatId);
+
+    if (!text.trim()) return;
+
+    // Skip very short buffers (likely a single token before a thinking pause).
+    // Will be accumulated with subsequent chunks. Force flush on shutdown.
+    if (!force && text.trim().length < BUFFER_MIN_FLUSH_LENGTH) {
+      getLog().debug({ chatId, textLength: text.trim().length }, 'telegram.buffer_skip_short');
+      // Re-buffer: put the text back for the next accumulation cycle
+      const existing = this.buffers.get(chatId);
+      if (existing) {
+        existing.text = text + existing.text;
+      } else {
+        this.buffers.set(chatId, { text, timer: null });
+      }
+      return;
+    }
+
+    getLog().debug({ chatId, textLength: text.length }, 'telegram.buffer_flush');
+    // Fire-and-forget — errors are logged inside deliverMessage/sendFormattedChunk
+    void this.deliverMessage(chatId, text).catch((err: unknown) => {
+      getLog().error({ err, chatId }, 'telegram.buffered_flush_failed');
+    });
+  }
+
+  /** Flush all pending buffers — called during shutdown to avoid losing in-flight text. */
+  private flushAllBuffers(): void {
+    for (const [chatId, state] of this.buffers) {
+      this.flushBuffer(chatId, state, true);
+    }
+  }
+
+  /** Send a complete (non-buffered) message to Telegram. */
+  private async deliverMessage(chatId: string, message: string): Promise<void> {
     const id = parseInt(chatId);
     getLog().debug({ chatId, messageLength: message.length }, 'telegram.send_message');
 
     if (message.length <= MAX_LENGTH) {
-      // Short message: try MarkdownV2 formatting
       await this.sendFormattedChunk(id, message);
     } else {
-      // Long message: split by paragraphs, format each chunk
       getLog().debug({ messageLength: message.length }, 'telegram.message_splitting');
       const chunks = splitIntoParagraphChunks(message, MAX_LENGTH - 200);
 
@@ -71,9 +166,7 @@ export class TelegramAdapter implements IPlatformAdapter {
     }
   }
 
-  /**
-   * Send a single chunk with MarkdownV2 formatting, with fallback to plain text
-   */
+  /** Send a single chunk with MarkdownV2 formatting, with fallback to plain text. */
   private async sendFormattedChunk(id: number, chunk: string): Promise<void> {
     // If chunk is still too long after paragraph splitting, fall back to plain text
     if (chunk.length > MAX_LENGTH) {
@@ -122,10 +215,12 @@ export class TelegramAdapter implements IPlatformAdapter {
   }
 
   /**
-   * Get the configured streaming mode
+   * Get the configured streaming mode.
+   * Buffered mode reports 'stream' to the orchestrator — chunks are sent
+   * one at a time and coalesced inside the adapter.
    */
   getStreamingMode(): 'stream' | 'batch' {
-    return this.streamingMode;
+    return this.mode === 'batch' ? 'batch' : 'stream';
   }
 
   /**
@@ -239,9 +334,11 @@ export class TelegramAdapter implements IPlatformAdapter {
   }
 
   /**
-   * Stop the bot gracefully
+   * Stop the bot gracefully.
+   * Flushes any pending buffered messages before stopping so in-flight text is not lost.
    */
   stop(): void {
+    this.flushAllBuffers();
     this.bot.stop();
     getLog().info('telegram.bot_stopped');
   }

packages/adapters/src/chat/telegram/adapter.ts

@@ -1 +1 @@
-export { TelegramAdapter } from './adapter';
+export { TelegramAdapter, type TelegramStreamingMode } from './adapter';

packages/adapters/src/chat/telegram/index.ts

@@ -1,5 +1,5 @@
 // Chat adapters
-export { TelegramAdapter } from './chat/telegram';
+export { TelegramAdapter, type TelegramStreamingMode } from './chat/telegram';
 export { SlackAdapter } from './chat/slack';
 
 // Forge adapters

packages/adapters/src/index.ts

@@ -652,11 +652,10 @@ describe('streaming tail completion', () => {
     } as unknown as AgentSessionEvent;
   }
 
-  test('emits corrective assistant chunk when streaming truncated', async () => {
+  test('emits assembled text from agent_end when streaming differs', async () => {
     const streamed = 'The repo is cloned. Let me register it.\n\n/register-project';
     const full =
       'The repo is cloned. Let me register it.\n\n/register-project SaberEngine "/path/to/repo"';
-    const tail = full.slice(streamed.length);
 
     let listener: ((event: AgentSessionEvent) => void) | undefined;
     const mockSession = {
@@ -666,7 +665,6 @@ describe('streaming tail completion', () => {
         return () => {};
       },
       prompt: async () => {
-        listener?.({ type: 'turn_start' } as AgentSessionEvent);
         listener?.(makeTextDeltaEvent(streamed));
         listener?.(makeAgentEndEvent(full));
       },
@@ -679,14 +677,14 @@ describe('streaming tail completion', () => {
       chunks.push(chunk);
     }
 
+    // text_deltas are suppressed; assembled text emitted at agent_end
     const assistantChunks = chunks.filter(c => c.type === 'assistant');
-    expect(assistantChunks).toHaveLength(2);
-    expect(assistantChunks[0].content).toBe(streamed);
-    expect(assistantChunks[1].content).toBe(tail);
+    expect(assistantChunks).toHaveLength(1);
+    expect(assistantChunks[0].content).toBe(full);
     expect(chunks[chunks.length - 1].type).toBe('result');
   });
 
-  test('does not emit corrective chunk when streaming is complete', async () => {
+  test('emits assembled text once when streaming matches', async () => {
     const full = 'complete text no truncation';
 
     let listener: ((event: AgentSessionEvent) => void) | undefined;
@@ -697,7 +695,6 @@ describe('streaming tail completion', () => {
         return () => {};
       },
       prompt: async () => {
-        listener?.({ type: 'turn_start' } as AgentSessionEvent);
         listener?.(makeTextDeltaEvent(full));
         listener?.(makeAgentEndEvent(full));
       },
@@ -715,7 +712,7 @@ describe('streaming tail completion', () => {
     expect(assistantChunks[0].content).toBe(full);
   });
 
-  test('does not emit corrective chunk when assembled text does not start with streamed (mismatch)', async () => {
+  test('emits assembled text even when it differs from streamed text', async () => {
     let listener: ((event: AgentSessionEvent) => void) | undefined;
     const mockSession = {
       sessionId: 'session-1',
@@ -724,7 +721,6 @@ describe('streaming tail completion', () => {
         return () => {};
       },
       prompt: async () => {
-        listener?.({ type: 'turn_start' } as AgentSessionEvent);
         listener?.(makeTextDeltaEvent('different content'));
         listener?.(makeAgentEndEvent('assembled is completely different'));
       },
@@ -739,10 +735,11 @@ describe('streaming tail completion', () => {
 
     const assistantChunks = chunks.filter(c => c.type === 'assistant');
     expect(assistantChunks).toHaveLength(1);
-    expect(assistantChunks[0].content).toBe('different content');
+    // Assembled text replaces streamed text
+    expect(assistantChunks[0].content).toBe('assembled is completely different');
   });
 
-  test('resets per-turn text on turn_start so only final turn is checked', async () => {
+  test('emits final assembled text from multi-turn session', async () => {
     let listener: ((event: AgentSessionEvent) => void) | undefined;
     const mockSession = {
       sessionId: 'session-1',
@@ -751,11 +748,9 @@ describe('streaming tail completion', () => {
         return () => {};
       },
       prompt: async () => {
-        listener?.({ type: 'turn_start' } as AgentSessionEvent);
         listener?.(makeTextDeltaEvent('turn one text'));
-        listener?.({ type: 'turn_start' } as AgentSessionEvent); // second turn resets counter
         listener?.(makeTextDeltaEvent('turn two'));
-        listener?.(makeAgentEndEvent('turn two')); // last assistant msg matches turn 2
+        listener?.(makeAgentEndEvent('turn two')); // last assistant msg
       },
       abort: async () => {},
       dispose: () => {},
@@ -766,13 +761,13 @@ describe('streaming tail completion', () => {
       chunks.push(chunk);
     }
 
+    // text_deltas suppressed; only assembled text emitted
     const assistantChunks = chunks.filter(c => c.type === 'assistant');
-    expect(assistantChunks).toHaveLength(2);
-    expect(assistantChunks[0].content).toBe('turn one text');
-    expect(assistantChunks[1].content).toBe('turn two');
+    expect(assistantChunks).toHaveLength(1);
+    expect(assistantChunks[0].content).toBe('turn two');
   });
 
-  test('corrective chunk is added to assistantBuffer when wantsStructured', async () => {
+  test('assembled text is used for structured output parsing', async () => {
     const streamed = '{"partial":';
     const full = '{"partial":true}';
 
@@ -784,7 +779,6 @@ describe('streaming tail completion', () => {
         return () => {};
       },
       prompt: async () => {
-        listener?.({ type: 'turn_start' } as AgentSessionEvent);
         listener?.(makeTextDeltaEvent(streamed));
         listener?.(makeAgentEndEvent(full));
       },