feat: Telegram 流式预览（sendMessageDraft）+ 出站渲染解耦 + 投递容错

Bridge 新增"边生成边显示"能力，利用 Telegram Bot API 9.5 的
sendMessageDraft 在私聊中以草稿形式流式展示 Claude 生成进度。
架构上抽象为通道级可选能力，便于后续扩展到其他 IM。

types.ts
- 新增 PreviewCapabilities、StreamingPreviewState 接口

channel-adapter.ts
- BaseChannelAdapter 新增三个可选方法：
  getPreviewCapabilities / sendPreview / endPreview
- sendPreview 返回 'sent'|'skip'|'degrade' 三态，
  区分瞬时失败（跳过）和永久失败（降级）

telegram-utils.ts
- 新增 sendMessageDraft() API 封装（plain text，截断到 4096）

telegram-adapter.ts
- 实现 getPreviewCapabilities: 开关检查 + 私聊/群聊检测 + 降级状态
- 实现 sendPreview: 调 sendMessageDraft，400/404 永久降级，429 仅跳过
- 实现 endPreview: 空操作
- stop() 清空 previewDegraded Set

conversation-engine.ts
- 新增 OnPartialText 回调类型
- processMessage / consumeStream 透传 onPartialText 参数
- 使用独立 previewText 变量（只累积、不因 tool_use 清零），
  避免预览文本在工具调用后回退/抖动

bridge-manager.ts
- 流式预览状态机：generateDraftId / getStreamConfig / flushPreview
  节流闭包（间隔+最小增量+trailing-edge timer），finally 清理
- flushPreview 仅在 'degrade' 时降级，'skip'/网络错误不降级
- 新增 deliverResponse() 按 channelType 分发渲染：
  Telegram → markdownToTelegramChunks + deliverRendered
  其他 IM → deliver() 纯文本（内部自动分块）
  解除 handleMessage 对 Telegram 渲染的硬编码依赖

delivery-layer.ts
- deliverRendered: chunk 失败后继续投递剩余 chunk（不中断），
  追发截断提示通知用户，返回 ok:false 标识不完整投递
- deliver: 保持原有 fail-fast 行为（短消息无需跳过）

route.ts (bridge/settings)
- BRIDGE_SETTING_KEYS 白名单追加 5 个流式配置项

docs/handover/bridge-system.md
- 更新目录结构、数据流、设计决策（#12 流式预览）、设置项表

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

src/app/api/bridge/settings/route.ts

@@ -9,6 +9,11 @@ const BRIDGE_SETTING_KEYS = [
   'bridge_default_work_dir',
   'bridge_default_model',
   'bridge_default_provider_id',
+  'bridge_telegram_stream_enabled',
+  'bridge_telegram_stream_interval_ms',
+  'bridge_telegram_stream_min_delta_chars',
+  'bridge_telegram_stream_max_chars',
+  'bridge_telegram_stream_private_only',
 ] as const;
 
 export async function GET() {

src/lib/bridge/adapters/telegram-adapter.ts

@@ -10,11 +10,12 @@ import type {
   ChannelType,
   InboundMessage,
   OutboundMessage,
+  PreviewCapabilities,
   SendResult,
 } from '../types';
 import type { FileAttachment } from '@/types';
 import { BaseChannelAdapter, registerAdapterFactory } from '../channel-adapter';
-import { callTelegramApi } from './telegram-utils';
+import { callTelegramApi, sendMessageDraft } from './telegram-utils';
 import {
   isImageEnabled,
   downloadPhoto,
@@ -79,6 +80,8 @@ export class TelegramAdapter extends BaseChannelAdapter {
   private waiters: Array<(msg: InboundMessage | null) => void> = [];
   private typingIntervals = new Map<string, ReturnType<typeof setInterval>>();
   private mediaGroupBuffers = new Map<string, MediaGroupBufferEntry>();
+  /** Chat IDs where sendMessageDraft has permanently failed (method not found / 400 / 404). */
+  private previewDegraded = new Set<string>();
 
   /** Committed offset — the highest update_id that has been safely enqueued or skipped. */
   private committedOffset = 0;
@@ -145,6 +148,9 @@ export class TelegramAdapter extends BaseChannelAdapter {
     }
     this.mediaGroupBuffers.clear();
 
+    // Reset preview degradation state
+    this.previewDegraded.clear();
+
     console.log('[telegram-adapter] Stopped');
   }
 
@@ -280,6 +286,44 @@ export class TelegramAdapter extends BaseChannelAdapter {
     this.persistCommittedOffset();
   }
 
+  // ── Streaming preview ────────────────────────────────────────
+
+  getPreviewCapabilities(chatId: string): PreviewCapabilities | null {
+    // Global kill switch
+    if (getSetting('bridge_telegram_stream_enabled') === 'false') return null;
+
+    // Private-only check: positive chatId = private, negative = group/channel
+    const privateOnly = getSetting('bridge_telegram_stream_private_only') !== 'false';
+    if (privateOnly && parseInt(chatId, 10) < 0) return null;
+
+    // Already degraded for this chat
+    if (this.previewDegraded.has(chatId)) return null;
+
+    return { supported: true, privateOnly };
+  }
+
+  async sendPreview(chatId: string, text: string, draftId: number): Promise<'sent' | 'skip' | 'degrade'> {
+    const token = this.botToken;
+    if (!token) return 'skip';
+
+    const result = await sendMessageDraft(token, chatId, text, draftId);
+    if (result.ok) return 'sent';
+
+    // Classify failure
+    const status = result.httpStatus;
+    if (status === 400 || status === 404) {
+      // Method not found or bad request — permanent degradation
+      this.previewDegraded.add(chatId);
+      return 'degrade';
+    }
+    // 429 (rate limit) or transient — skip this update but don't degrade
+    return 'skip';
+  }
+
+  endPreview(_chatId: string, _draftId: number): void {
+    // No-op: the final sendMessage naturally replaces the draft
+  }
+
   // ── Lifecycle hooks (called generically by bridge-manager) ───
 
   onMessageStart(chatId: string): void {

src/lib/bridge/adapters/telegram-utils.ts

@@ -66,6 +66,24 @@ export async function callTelegramApi(
   }
 }
 
+/**
+ * Send a draft message preview via Telegram Bot API 9.5 sendMessageDraft.
+ * Plain text only (no parse_mode) — used for streaming preview.
+ */
+export async function sendMessageDraft(
+  botToken: string,
+  chatId: string,
+  text: string,
+  draftId: number,
+): Promise<TelegramSendResult> {
+  const truncated = text.length > 4096 ? text.slice(0, 4096) : text;
+  return callTelegramApi(botToken, 'sendMessageDraft', {
+    chat_id: chatId,
+    text: truncated,
+    draft_id: draftId,
+  });
+}
+
 /**
  * Escape special HTML characters for Telegram HTML mode.
  */

src/lib/bridge/bridge-manager.ts

@@ -7,7 +7,7 @@
  * Uses globalThis to survive Next.js HMR in development.
  */
 
-import type { BridgeStatus, InboundMessage, OutboundMessage } from './types';
+import type { BridgeStatus, InboundMessage, OutboundMessage, StreamingPreviewState } from './types';
 import { createAdapter, getRegisteredTypes } from './channel-adapter';
 import type { BaseChannelAdapter } from './channel-adapter';
 // Side-effect import: triggers self-registration of all adapter factories
@@ -30,6 +30,79 @@ import {
 
 const GLOBAL_KEY = '__bridge_manager__';
 
+// ── Streaming preview helpers ──────────────────────────────────
+
+/** Generate a non-zero random 31-bit integer for use as draft_id. */
+function generateDraftId(): number {
+  return (Math.floor(Math.random() * 0x7FFFFFFE) + 1); // 1 .. 2^31-1
+}
+
+interface StreamConfig {
+  intervalMs: number;
+  minDeltaChars: number;
+  maxChars: number;
+}
+
+function getStreamConfig(): StreamConfig {
+  const intervalMs = parseInt(getSetting('bridge_telegram_stream_interval_ms') || '', 10) || 700;
+  const minDeltaChars = parseInt(getSetting('bridge_telegram_stream_min_delta_chars') || '', 10) || 20;
+  const maxChars = parseInt(getSetting('bridge_telegram_stream_max_chars') || '', 10) || 3900;
+  return { intervalMs, minDeltaChars, maxChars };
+}
+
+/** Fire-and-forget: send a preview draft. Only degrades on permanent failure. */
+function flushPreview(
+  adapter: BaseChannelAdapter,
+  state: StreamingPreviewState,
+  config: StreamConfig,
+): void {
+  if (state.degraded || !adapter.sendPreview) return;
+
+  const text = state.pendingText.length > config.maxChars
+    ? state.pendingText.slice(0, config.maxChars) + '...'
+    : state.pendingText;
+
+  state.lastSentText = text;
+  state.lastSentAt = Date.now();
+
+  adapter.sendPreview(state.chatId, text, state.draftId).then(result => {
+    if (result === 'degrade') state.degraded = true;
+    // 'skip' — transient failure, next flush will retry naturally
+  }).catch(() => {
+    // Network error — transient, don't degrade
+  });
+}
+
+// ── Channel-aware rendering dispatch ──────────────────────────
+
+import type { ChannelAddress, SendResult } from './types';
+
+/**
+ * Render response text and deliver via the appropriate channel format.
+ * Telegram: Markdown → HTML chunks via deliverRendered.
+ * Other channels: plain text via deliver (no HTML).
+ */
+async function deliverResponse(
+  adapter: BaseChannelAdapter,
+  address: ChannelAddress,
+  responseText: string,
+  sessionId: string,
+): Promise<SendResult> {
+  if (adapter.channelType === 'telegram') {
+    const chunks = markdownToTelegramChunks(responseText, 4096);
+    if (chunks.length > 0) {
+      return deliverRendered(adapter, address, chunks, { sessionId });
+    }
+    return { ok: true };
+  }
+  // Generic fallback: deliver as plain text (deliver() handles chunking internally)
+  return deliver(adapter, {
+    address,
+    text: responseText,
+    parseMode: 'plain',
+  }, { sessionId });
+}
+
 interface AdapterMeta {
   lastMessageAt: string | null;
   lastError: string | null;
@@ -365,6 +438,67 @@ async function handleMessage(
   const state = getState();
   state.activeTasks.set(binding.codepilotSessionId, taskAbort);
 
+  // ── Streaming preview setup ──────────────────────────────────
+  let previewState: StreamingPreviewState | null = null;
+  const caps = adapter.getPreviewCapabilities?.(msg.address.chatId) ?? null;
+  if (caps?.supported) {
+    previewState = {
+      draftId: generateDraftId(),
+      chatId: msg.address.chatId,
+      lastSentText: '',
+      lastSentAt: 0,
+      degraded: false,
+      throttleTimer: null,
+      pendingText: '',
+    };
+  }
+
+  const streamCfg = previewState ? getStreamConfig() : null;
+
+  // Build the onPartialText callback (or undefined if preview not supported)
+  const onPartialText = (previewState && streamCfg) ? (fullText: string) => {
+    const ps = previewState!;
+    const cfg = streamCfg!;
+    if (ps.degraded) return;
+
+    // Truncate to maxChars + ellipsis
+    ps.pendingText = fullText.length > cfg.maxChars
+      ? fullText.slice(0, cfg.maxChars) + '...'
+      : fullText;
+
+    const delta = ps.pendingText.length - ps.lastSentText.length;
+    const elapsed = Date.now() - ps.lastSentAt;
+
+    if (delta < cfg.minDeltaChars && ps.lastSentAt > 0) {
+      // Not enough new content — schedule trailing-edge timer if not already set
+      if (!ps.throttleTimer) {
+        ps.throttleTimer = setTimeout(() => {
+          ps.throttleTimer = null;
+          if (!ps.degraded) flushPreview(adapter, ps, cfg);
+        }, cfg.intervalMs);
+      }
+      return;
+    }
+
+    if (elapsed < cfg.intervalMs && ps.lastSentAt > 0) {
+      // Too soon — schedule trailing-edge timer to ensure latest text is sent
+      if (!ps.throttleTimer) {
+        ps.throttleTimer = setTimeout(() => {
+          ps.throttleTimer = null;
+          if (!ps.degraded) flushPreview(adapter, ps, cfg);
+        }, cfg.intervalMs - elapsed);
+      }
+      return;
+    }
+
+    // Clear any pending trailing-edge timer and flush immediately
+    if (ps.throttleTimer) {
+      clearTimeout(ps.throttleTimer);
+      ps.throttleTimer = null;
+    }
+    flushPreview(adapter, ps, cfg);
+  } : undefined;
+
   try {
     // Pass permission callback so requests are forwarded to IM immediately
     // during streaming (the stream blocks until permission is resolved).
@@ -381,16 +515,11 @@ async function handleMessage(
         binding.codepilotSessionId,
         perm.suggestions,
       );
-    }, taskAbort.signal, hasAttachments ? msg.attachments : undefined);
+    }, taskAbort.signal, hasAttachments ? msg.attachments : undefined, onPartialText);
 
-    // Send response text — render Markdown to Telegram HTML
+    // Send response text — render via channel-appropriate format
     if (result.responseText) {
-      const chunks = markdownToTelegramChunks(result.responseText, 4096);
-      if (chunks.length > 0) {
-        await deliverRendered(adapter, msg.address, chunks, {
-          sessionId: binding.codepilotSessionId,
-        });
-      }
+      await deliverResponse(adapter, msg.address, result.responseText, binding.codepilotSessionId);
     } else if (result.hasError) {
       const errorResponse: OutboundMessage = {
         address: msg.address,
@@ -413,6 +542,15 @@ async function handleMessage(
       } catch { /* best effort */ }
     }
   } finally {
+    // Clean up preview state
+    if (previewState) {
+      if (previewState.throttleTimer) {
+        clearTimeout(previewState.throttleTimer);
+        previewState.throttleTimer = null;
+      }
+      adapter.endPreview?.(msg.address.chatId, previewState.draftId);
+    }
+
     state.activeTasks.delete(binding.codepilotSessionId);
     // Notify adapter that message processing ended
     adapter.onMessageEnd?.(msg.address.chatId);

src/lib/bridge/channel-adapter.ts

@@ -9,6 +9,7 @@ import type {
   ChannelType,
   InboundMessage,
   OutboundMessage,
+  PreviewCapabilities,
   SendResult,
 } from './types';
 
@@ -76,6 +77,25 @@ export abstract class BaseChannelAdapter {
    * Default is a no-op; override in adapters that need deferred offset tracking.
    */
   acknowledgeUpdate?(_updateId: number): void;
+
+  /**
+   * Return preview capabilities for a given chat.
+   * Returning null means streaming preview is not available for this chat.
+   */
+  getPreviewCapabilities?(_chatId: string): PreviewCapabilities | null;
+
+  /**
+   * Send (or update) a streaming preview draft.
+   * Returns 'sent' on success, 'skip' for transient failures (caller should
+   * retry later), or 'degrade' for permanent failures (caller should stop).
+   */
+  sendPreview?(_chatId: string, _text: string, _draftId: number): Promise<'sent' | 'skip' | 'degrade'>;
+
+  /**
+   * Signal the end of a preview cycle. The final message is sent via the
+   * normal delivery path, so this is typically a no-op.
+   */
+  endPreview?(_chatId: string, _draftId: number): void;
 }
 
 // ── Adapter Registry ────────────────────────────────────────────

src/lib/bridge/conversation-engine.ts

@@ -43,6 +43,12 @@ export interface PermissionRequestInfo {
  */
 export type OnPermissionRequest = (perm: PermissionRequestInfo) => Promise<void>;
 
+/**
+ * Callback invoked on each `text` SSE event with the full accumulated text so far.
+ * Must return synchronously — the bridge-manager handles throttling and fire-and-forget.
+ */
+export type OnPartialText = (fullText: string) => void;
+
 export interface ConversationResult {
   responseText: string;
   tokenUsage: TokenUsage | null;
@@ -64,6 +70,7 @@ export async function processMessage(
   onPermissionRequest?: OnPermissionRequest,
   abortSignal?: AbortSignal,
   files?: FileAttachment[],
+  onPartialText?: OnPartialText,
 ): Promise<ConversationResult> {
   const sessionId = binding.codepilotSessionId;
 
@@ -179,7 +186,7 @@ export async function processMessage(
     // Consume the stream server-side (replicate collectStreamResponse pattern).
     // Permission requests are forwarded immediately via the callback during streaming
     // because the stream blocks until permission is resolved — we can't wait until after.
-    return await consumeStream(stream, sessionId, onPermissionRequest);
+    return await consumeStream(stream, sessionId, onPermissionRequest, onPartialText);
   } finally {
     clearInterval(renewalInterval);
     releaseSessionLock(sessionId, lockId);
@@ -195,10 +202,13 @@ async function consumeStream(
   stream: ReadableStream<string>,
   sessionId: string,
   onPermissionRequest?: OnPermissionRequest,
+  onPartialText?: OnPartialText,
 ): Promise<ConversationResult> {
   const reader = stream.getReader();
   const contentBlocks: MessageContentBlock[] = [];
   let currentText = '';
+  /** Monotonically accumulated text for streaming preview — never resets on tool_use. */
+  let previewText = '';
   let tokenUsage: TokenUsage | null = null;
   let hasError = false;
   let errorMessage = '';
@@ -225,6 +235,10 @@ async function consumeStream(
         switch (event.type) {
           case 'text':
             currentText += event.data;
+            if (onPartialText) {
+              previewText += event.data;
+              try { onPartialText(previewText); } catch { /* non-critical */ }
+            }
             break;
 
           case 'tool_use': {

src/lib/bridge/delivery-layer.ts

@@ -287,6 +287,7 @@ export async function deliverRendered(
   }
 
   let lastMessageId: string | undefined;
+  let failedCount = 0;
 
   for (let i = 0; i < chunks.length; i++) {
     await rateLimiter.acquire(address.chatId);
@@ -303,7 +304,14 @@ export async function deliverRendered(
 
     // Try HTML first, fall back to plain text on parse error
     const result = await sendWithRetry(adapter, htmlMessage, chunk.text);
-    if (!result.ok) return result;
+    if (!result.ok) {
+      console.warn(
+        `[delivery-layer] Chunk ${i + 1}/${chunks.length} failed for chat ${address.chatId}: ${result.error}`,
+      );
+      failedCount++;
+      // Continue delivering remaining chunks instead of aborting
+      continue;
+    }
     lastMessageId = result.messageId;
 
     if (result.messageId && opts?.sessionId) {
@@ -319,6 +327,12 @@ export async function deliverRendered(
     }
   }
 
+  // Notify user about incomplete delivery
+  if (failedCount > 0 && lastMessageId) {
+    const notice = `[${failedCount}/${chunks.length} part(s) failed to send — response may be incomplete]`;
+    await adapter.send({ address, text: notice, parseMode: 'plain' }).catch(() => {});
+  }
+
   if (opts?.dedupKey) {
     try { insertDedup(opts.dedupKey); } catch { /* best effort */ }
   }
@@ -333,5 +347,8 @@ export async function deliverRendered(
     });
   } catch { /* best effort */ }
 
+  if (failedCount > 0) {
+    return { ok: false, error: `${failedCount}/${chunks.length} chunk(s) failed`, messageId: lastMessageId };
+  }
   return { ok: true, messageId: lastMessageId };
 }

src/lib/bridge/types.ts

@@ -145,6 +145,25 @@ export interface PermissionLink {
   createdAt: string;
 }
 
+// ── Streaming Preview ─────────────────────────────────────────
+
+/** Capabilities of a channel adapter's streaming preview support */
+export interface PreviewCapabilities {
+  supported: boolean;
+  privateOnly: boolean;
+}
+
+/** Mutable state for an in-flight streaming preview */
+export interface StreamingPreviewState {
+  draftId: number;           // non-zero 31-bit random integer, reused within one answer cycle
+  chatId: string;
+  lastSentText: string;      // last text actually sent as draft
+  lastSentAt: number;        // timestamp (ms) of last sent draft
+  degraded: boolean;         // set true after API failure → skip further previews
+  throttleTimer: ReturnType<typeof setTimeout> | null;
+  pendingText: string;       // latest accumulated text (may not yet be sent due to throttle)
+}
+
 // ── Config ─────────────────────────────────────────────────────
 
 /** Platform-specific message length limits */