From b1a043a995def500da73bb376e5a0c42411116a7 Mon Sep 17 00:00:00 2001 From: 7418 <7418@openclaw.ai> Date: Tue, 3 Mar 2026 23:20:47 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20Telegram=20=E6=B5=81=E5=BC=8F=E9=A2=84?= =?UTF-8?q?=E8=A7=88=EF=BC=88sendMessageDraft=EF=BC=89+=20=E5=87=BA?= =?UTF-8?q?=E7=AB=99=E6=B8=B2=E6=9F=93=E8=A7=A3=E8=80=A6=20+=20=E6=8A=95?= =?UTF-8?q?=E9=80=92=E5=AE=B9=E9=94=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- docs/handover/bridge-system.md | 31 +++- package-lock.json | 4 +- package.json | 2 +- src/app/api/bridge/settings/route.ts | 5 + src/lib/bridge/adapters/telegram-adapter.ts | 46 +++++- src/lib/bridge/adapters/telegram-utils.ts | 18 +++ src/lib/bridge/bridge-manager.ts | 156 ++++++++++++++++++-- src/lib/bridge/channel-adapter.ts | 20 +++ src/lib/bridge/conversation-engine.ts | 16 +- src/lib/bridge/delivery-layer.ts | 19 ++- src/lib/bridge/types.ts | 19 +++ 11 files changed, 313 insertions(+), 23 deletions(-) diff --git a/docs/handover/bridge-system.md b/docs/handover/bridge-system.md index 6bec08c..2b145e2 100644 --- a/docs/handover/bridge-system.md +++ b/docs/handover/bridge-system.md @@ -11,10 +11,10 @@ src/lib/bridge/ ├── types.ts # 共享类型（ChannelBinding, BridgeStatus, InboundMessage 等） ├── channel-adapter.ts # 抽象基类 + adapter 注册表（registerAdapterFactory/createAdapter） ├── channel-router.ts # (channel, user, thread) → session 映射，自动创建/绑定会话 -├── conversation-engine.ts # 服务端消费 streamClaude() SSE 流，保存消息到 DB +├── conversation-engine.ts # 服务端消费 streamClaude() SSE 流，保存消息到 DB，onPartialText 流式回调 ├── permission-broker.ts # 权限请求转发到 IM 内联按钮，处理回调审批 ├── delivery-layer.ts # 出站消息分片、限流、重试退避、HTML 降级 -├── bridge-manager.ts # 生命周期编排，adapter 事件循环，/stop abort，命令路由 +├── bridge-manager.ts # 生命周期编排，adapter 事件循环，流式预览状态机，deliverResponse 渲染分发 ├── markdown/ │ ├── ir.ts # Markdown → IR 中间表示解析器（基于 markdown-it） │ ├── render.ts # IR → 格式化输出的通用标记渲染器 @@ -23,7 +23,7 @@ src/lib/bridge/ │ ├── index.ts # Adapter 目录文件（side-effect import 自注册所有 adapter） │ ├── telegram-adapter.ts # Telegram 长轮询 + offset 安全水位 + 图片/相册处理 + 自注册 │ ├── telegram-media.ts # Telegram 图片下载、尺寸选择、base64 转换 -│ └── telegram-utils.ts # callTelegramApi / escapeHtml / splitMessage +│ └── telegram-utils.ts # callTelegramApi / sendMessageDraft / escapeHtml / splitMessage └── security/ ├── rate-limiter.ts # 按 chat 滑动窗口限流（20 条/分钟） └── validators.ts # 路径/SessionID/危险输入校验 @@ -39,16 +39,18 @@ Telegram 消息 → TelegramAdapter.pollLoop() → BridgeManager.runAdapterLoop() → handleMessage() → 命令? → handleCommand() 处理 /new /bind /cwd /mode /stop 等 → 普通消息/图片? → ChannelRouter.resolve() 获取 ChannelBinding - → ConversationEngine.processMessage(binding, text, ..., files?) + → ConversationEngine.processMessage(binding, text, ..., files?, onPartialText?) → 有图片时：写入 .codepilot-uploads/ + text 格式存 DB（桌面 UI 可渲染） → streamClaude({ prompt, files }) → Claude vision API → consumeStream() 服务端消费 → permission_request → 立即回调 → PermissionBroker 转发到 IM - → text/tool_use/tool_result → 累积内容块 + → text → 累积 currentText + previewText → onPartialText(previewText) 回调 + → tool_use/tool_result → 累积内容块（currentText 清零，previewText 不清零） → result → 捕获 tokenUsage + sdkSessionId → addMessage() 保存到 DB - → markdownToTelegramChunks() → Markdown→IR→HTML render-first 分片 - → DeliveryLayer.deliverRendered() → 限流 + HTML/plain 双通道发送到 Telegram + → deliverResponse() 按 channelType 分发渲染: + → Telegram: markdownToTelegramChunks() → deliverRendered() → 限流 + HTML/plain 双通道 + → 其他 IM: deliver() → 纯文本分块发送 → finally: adapter.acknowledgeUpdate(updateId) → 推进 committedOffset 并持久化 ``` @@ -105,7 +107,15 @@ Claude 的回复是 Markdown 格式，Telegram 仅支持有限 HTML 标签（b/i - **渲染层**（`markdown/render.ts`）：通用标记渲染器 `renderMarkdownWithMarkers(ir, options)`，接受样式→标签映射表 + escapeText + buildLink 回调，输出格式化文本。使用 boundary tracking + LIFO stack 处理嵌套。 - **Telegram 层**（`markdown/telegram.ts`）：组合 IR+渲染器，映射样式到 Telegram HTML 标签。`wrapFileReferencesInHtml()` 防止 `README.md`、`main.go` 等文件名被 Telegram linkify 误识别为 URL（用 `` 包裹）。`markdownToTelegramChunks(text, limit)` 实现 render-first 分片：先按 IR text 长度分块，再渲染每块为 HTML，若 HTML 超出 4096 限制则按比例重新分割。 -`bridge-manager.ts` 对 Claude 回复调用 `markdownToTelegramChunks()`，通过 `deliverRendered()` 发送预渲染 chunks（每个 chunk 包含 html + text 双通道，HTML 解析失败自动降级纯文本）。命令响应和错误消息仍使用 `escapeHtml()` + `deliver()`。 +`bridge-manager.ts` 通过 `deliverResponse()` 按 `adapter.channelType` 分发渲染：Telegram 走 `markdownToTelegramChunks()` + `deliverRendered()`（HTML/plain 双通道），其他 IM 走 `deliver()` 纯文本。`deliverRendered()` 在分块部分失败时继续投递剩余 chunk 并追发截断提示，最终返回 `ok: false` 标识不完整投递。命令响应和错误消息仍使用 `escapeHtml()` + `deliver()`。 + +**12. Telegram 流式预览（sendMessageDraft）** +利用 Telegram Bot API 9.5 的 `sendMessageDraft` 方法，在 Claude 生成过程中以草稿形式实时展示文本预览。架构上抽象为通道级可选能力（`BaseChannelAdapter` 的 `getPreviewCapabilities`/`sendPreview`/`endPreview` 三个可选方法），未实现这些方法的 adapter 自动跳过。 + +- **引擎层**：`consumeStream()` 维护独立的 `previewText` 变量（只累积、不因 `tool_use` 清零），通过 `onPartialText` 回调同步传递完整预览文本。 +- **编排层**：`bridge-manager.handleMessage()` 检查 adapter 能力 → 分配 `draftId` → 构建节流闭包（间隔 700ms + 最小增量 20 字符 + trailing-edge timer）→ `flushPreview()` fire-and-forget 发送 → finally 清理 timer + `endPreview()`。 +- **降级**：`sendPreview` 返回 `'sent'|'skip'|'degrade'` 三态。400/404（API 不支持）→ 永久降级该 chatId；429/网络错误 → 仅跳过本次。`previewDegraded` Set 在 adapter `stop()` 时清空。 +- **线程安全**：`processWithSessionLock` 保证同 session 串行 → 同时刻只有一个 `previewState`。多个 in-flight `sendMessageDraft` 安全：Telegram 对同 `draft_id` last-write-wins。 ## 设置项（settings 表） @@ -120,6 +130,11 @@ Claude 的回复是 Markdown 格式，Telegram 仅支持有限 HTML 标签（b/i | telegram_bridge_allowed_users | 白名单用户 ID（逗号分隔） | | bridge_telegram_image_enabled | Telegram 图片接收开关（默认 true，设为 false 关闭） | | bridge_telegram_max_image_size | 图片大小上限（字节，默认 20MB） | +| bridge_telegram_stream_enabled | 流式预览总开关（默认启用，设为 `false` 关闭） | +| bridge_telegram_stream_interval_ms | 预览节流间隔（默认 700ms） | +| bridge_telegram_stream_min_delta_chars | 最小增量字符数（默认 20） | +| bridge_telegram_stream_max_chars | 草稿截断阈值（默认 3900） | +| bridge_telegram_stream_private_only | 仅私聊启用预览（默认 true，群聊自动跳过） | ## API 路由 diff --git a/package-lock.json b/package-lock.json index 26fbd97..3ad99a4 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "codepilot", - "version": "0.23.1", + "version": "0.24.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "codepilot", - "version": "0.23.1", + "version": "0.24.0", "dependencies": { "@ai-sdk/anthropic": "^3.0.47", "@ai-sdk/google": "^3.0.31", diff --git a/package.json b/package.json index a61ad84..5b6d4e3 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "codepilot", - "version": "0.23.1", + "version": "0.24.0", "private": true, "author": { "name": "op7418", diff --git a/src/app/api/bridge/settings/route.ts b/src/app/api/bridge/settings/route.ts index ef564ac..30e84f1 100644 --- a/src/app/api/bridge/settings/route.ts +++ b/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() { diff --git a/src/lib/bridge/adapters/telegram-adapter.ts b/src/lib/bridge/adapters/telegram-adapter.ts index f114012..173220b 100644 --- a/src/lib/bridge/adapters/telegram-adapter.ts +++ b/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>(); private mediaGroupBuffers = new Map(); + /** Chat IDs where sendMessageDraft has permanently failed (method not found / 400 / 404). */ + private previewDegraded = new Set(); /** 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 { diff --git a/src/lib/bridge/adapters/telegram-utils.ts b/src/lib/bridge/adapters/telegram-utils.ts index 76d804f..8f180ab 100644 --- a/src/lib/bridge/adapters/telegram-utils.ts +++ b/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 { + 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. */ diff --git a/src/lib/bridge/bridge-manager.ts b/src/lib/bridge/bridge-manager.ts index 9af5c14..b935b84 100644 --- a/src/lib/bridge/bridge-manager.ts +++ b/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 { + 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); diff --git a/src/lib/bridge/channel-adapter.ts b/src/lib/bridge/channel-adapter.ts index 8fbbcda..12cc752 100644 --- a/src/lib/bridge/channel-adapter.ts +++ b/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 ──────────────────────────────────────────── diff --git a/src/lib/bridge/conversation-engine.ts b/src/lib/bridge/conversation-engine.ts index 1cfd07c..c213fd5 100644 --- a/src/lib/bridge/conversation-engine.ts +++ b/src/lib/bridge/conversation-engine.ts @@ -43,6 +43,12 @@ export interface PermissionRequestInfo { */ export type OnPermissionRequest = (perm: PermissionRequestInfo) => Promise; +/** + * 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 { 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, sessionId: string, onPermissionRequest?: OnPermissionRequest, + onPartialText?: OnPartialText, ): Promise { 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': { diff --git a/src/lib/bridge/delivery-layer.ts b/src/lib/bridge/delivery-layer.ts index 47d4da2..1963e8a 100644 --- a/src/lib/bridge/delivery-layer.ts +++ b/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 }; } diff --git a/src/lib/bridge/types.ts b/src/lib/bridge/types.ts index b321e94..a009e24 100644 --- a/src/lib/bridge/types.ts +++ b/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 | null; + pendingText: string; // latest accumulated text (may not yet be sent due to throttle) +} + // ── Config ───────────────────────────────────────────────────── /** Platform-specific message length limits */