From f3f6a3ebafb6f0dbcfa4119c82637835b03a24c2 Mon Sep 17 00:00:00 2001 From: RRRosmontis Date: Sat, 5 Sep 2026 17:51:21 +0800 Subject: [PATCH] feat: admit photos and image documents into the bound session MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Telegram 图片消息此前一律回复「暂只支持文本消息」。本改动： - photo 取最大尺寸（TG 压成 jpeg），document 仅接受 image/png|jpeg|webp|gif - caption 作为伴随文本块进入同一条消息（不解析为命令） - 相册（同一 media_group_id）去抖 1.2s 合并为一条多图消息，避免 N 连回合 - 派发走宿主 sessions.prompt RPC（Web 端同通道）：模型图像能力校验 + 附件服务持久化 + mode queue 排队，与纯文本路径的 agent.followup 语义一致 - 错误映射：MODEL_DOES_NOT_SUPPORT_IMAGES → 提示换多模态模型；其余回退 图片发送失败 + [code] 详情；下载失败即时提示 - 客户端新增 getFile/downloadFile（GET file 端点，复用代理感知 fetch） tests +7（58/58）：photo+caption、png document、pdf 拒绝、相册合并、 模型不支持、下载失败、未绑定。无 emoji 文案。 --- README.md | 2 + lib/apiproxy.d.ts | 8 ++ lib/bridge.d.ts | 24 ++++ lib/bridge.js | 148 ++++++++++++++++++++++++- lib/client.d.ts | 32 ++++++ lib/client.js | 17 +++ lib/commands.d.ts | 3 + lib/commands.js | 9 ++ src/apiproxy.ts | 5 + src/bridge.ts | 164 ++++++++++++++++++++++++++- src/client.ts | 51 +++++++++ src/commands.ts | 8 ++ tests/bridge.test.ts | 256 +++++++++++++++++++++++++++++++++++++++++++ 13 files changed, 723 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 58b7c62..66b012e 100644 --- a/README.md +++ b/README.md @@ -34,6 +34,8 @@ Telegram **手机遥控器** for DeepSeek Harness：附着本机正在跑的 Web 2. 手机 Bot：`/sessions` → **工作区** → **会话** → 附着 3. 之后手机 ↔ Web 走**同一条**轨迹；可用 `/model` 切换模型（下一回合生效） +支持的消息：文本、图片（照片或以文件形式发送的 png/jpeg/webp/gif，可带文字说明；多张相册照片合并为一条消息进入会话）。语音/视频/其他文件暂不支持（会收到提示）。图片经宿主附件服务持久化，与 Web 端上传走同一通道；当前模型不支持图像输入时会收到明确提示。 + ### 效果截图 手机选择会话并发问： diff --git a/lib/apiproxy.d.ts b/lib/apiproxy.d.ts index 898e208..a0a0cd3 100644 --- a/lib/apiproxy.d.ts +++ b/lib/apiproxy.d.ts @@ -22,6 +22,14 @@ export type ApiProxySessions = { rpcId: string; payload: unknown; }) => Promise; + /** + * Prompt dispatch (Web 端同通道)：文本 + 图片字节(base64)经宿主附件服务持久化后 + * 进入会话；含图片时会校验当前模型是否支持图像输入。 + */ + prompt?: (req: { + rpcId: string; + payload: unknown; + }) => Promise; }; export type ApiProxyLike = { workspace?: { diff --git a/lib/bridge.d.ts b/lib/bridge.d.ts index 7c5d0a8..9350f70 100644 --- a/lib/bridge.d.ts +++ b/lib/bridge.d.ts @@ -53,6 +53,8 @@ export declare class TelegramBridge { private readonly compacting; /** sessionId → compaction abort controller (cancelled on bridge stop) */ private readonly compactAborts; + /** Album accumulation: chatId:media_group_id → parts buffered into one prompt. */ + private readonly mediaGroups; constructor(ctx: Context, options: TelegramBridgeOptions); /** Effective rendering mode for a chat: per-chat override, else the global mode. */ private modeFor; @@ -83,6 +85,28 @@ export declare class TelegramBridge { /** 等待压缩落定并汇报结果（不阻塞轮询循环）。 */ private runCompaction; private followupBound; + /** + * Extract wire prompt parts from a media message. + * Returns undefined when the message carries no media at all; returns [] for + * an unsupported media type (the user was already notified); otherwise the + * caption text part (when present) followed by the image part. + */ + private collectMediaParts; + /** Download one TG file and encode it as a canonical-base64 wire image part. */ + private downloadImagePart; + /** + * Album photos arrive as separate updates sharing a media_group_id — debounce + * them into ONE prompt so an album does not spawn N sequential turns. + */ + private accumulateMediaGroup; + /** + * Dispatch an image-bearing message to the bound session through the host + * prompt RPC — the exact channel the Web composer uses. The host checks + * whether the current model accepts image input, admits the bytes into the + * durable attachment store, then queues the message on the agent (mode + * 'queue' — same semantics as the text path's agent.followup). + */ + private sendImageFollowup; private pollLoop; private interruptibleDelay; private interruptibleSleep; diff --git a/lib/bridge.js b/lib/bridge.js index eefebfc..8feece3 100644 --- a/lib/bridge.js +++ b/lib/bridge.js @@ -1,3 +1,5 @@ +import { Buffer } from 'node:buffer'; +import { randomUUID } from 'node:crypto'; import { mkdirSync, readFileSync, writeFileSync } from 'node:fs'; import { homedir } from 'node:os'; import { dirname, join } from 'node:path'; @@ -14,6 +16,10 @@ import { formatLastTurn, loadLastTurn } from './history.js'; import { describeAgent, displayLabel, workspaceName } from './label.js'; import { formatModel, loadSessionModels, selectSessionModel, } from './models.js'; import { formatStatusText, } from './status.js'; +/** Host-admitted raster media types (@deepseek-ai/dsh-attachment). */ +const IMAGE_MEDIA_TYPES = ['image/png', 'image/jpeg', 'image/webp', 'image/gif']; +/** Album debounce window: TG emits each photo of an album as its own update. */ +const MEDIA_GROUP_WINDOW_MS = 1200; /** * 把 command-compact 返回的英文结果翻译成 TG 中文回复。 * 文案以宿主 @deepseek-ai/dsh-command-compact 的固定输出为准；无法识别时返回 @@ -115,6 +121,8 @@ export class TelegramBridge { compacting = new Set(); /** sessionId → compaction abort controller (cancelled on bridge stop) */ compactAborts = new Map(); + /** Album accumulation: chatId:media_group_id → parts buffered into one prompt. */ + mediaGroups = new Map(); constructor(ctx, options) { this.ctx = ctx; this.token = options.token; @@ -195,6 +203,9 @@ export class TelegramBridge { controller.abort(); this.compactAborts.clear(); this.compacting.clear(); + for (const group of this.mediaGroups.values()) + clearTimeout(group.timer); + this.mediaGroups.clear(); if (this.hookTimer) { clearTimeout(this.hookTimer); this.hookTimer = undefined; @@ -220,9 +231,23 @@ export class TelegramBridge { await this.client.sendMessage(chatId, MSG.DENIED); return; } - // Non-text messages: tell the (authorized) user instead of dropping silently. + // Media messages: photos / image documents (optionally with a caption) are + // admitted into the bound session; other media get an explicit notice. + // A caption is treated as accompanying text — never parsed as a command. + const mediaParts = await this.collectMediaParts(chatId, message); + if (mediaParts !== undefined) { + if (mediaParts.length > 0) { + if (message.media_group_id) + this.accumulateMediaGroup(chatId, message.media_group_id, mediaParts); + else + await this.sendImageFollowup(chatId, mediaParts); + } + return; + } + // Non-text messages without recognized media: tell the (authorized) user + // instead of dropping silently. if (!message.text) { - await this.enqueueNotice(chatId, '暂只支持文本消息（图片/语音/文件等媒体暂不处理）'); + await this.enqueueNotice(chatId, MSG.MEDIA_UNSUPPORTED); return; } const parsed = parseCommand(message.text); @@ -883,6 +908,125 @@ export class TelegramBridge { await this.enqueueNotice(chatId, '已加入队列：当前任务完成后将处理这条消息'); } } + /** + * Extract wire prompt parts from a media message. + * Returns undefined when the message carries no media at all; returns [] for + * an unsupported media type (the user was already notified); otherwise the + * caption text part (when present) followed by the image part. + */ + async collectMediaParts(chatId, message) { + const caption = message.caption?.trim(); + const textPart = caption ? { type: 'text', text: caption } : undefined; + const withText = (part) => textPart ? [textPart, part] : [part]; + if (message.photo && message.photo.length > 0) { + // TG photo sizes are ordered smallest → largest; the last is the best quality. + const best = message.photo[message.photo.length - 1]; + const part = await this.downloadImagePart(chatId, best.file_id, 'image/jpeg'); + return part ? withText(part) : []; + } + const doc = message.document; + if (doc) { + if (!doc.mime_type || !IMAGE_MEDIA_TYPES.includes(doc.mime_type)) { + await this.enqueueNotice(chatId, MSG.MEDIA_UNSUPPORTED); + return []; + } + const part = await this.downloadImagePart(chatId, doc.file_id, doc.mime_type, doc.file_name); + return part ? withText(part) : []; + } + return undefined; + } + /** Download one TG file and encode it as a canonical-base64 wire image part. */ + async downloadImagePart(chatId, fileId, mediaType, name) { + try { + const file = await this.client.getFile(fileId); + if (!file.file_path) + throw new Error('TG 未返回文件路径'); + const bytes = await this.client.downloadFile(file.file_path); + // Buffer base64 is canonical (standard alphabet, no whitespace) — the host + // attachment admission rejects non-canonical forms outright. + const data = Buffer.from(bytes).toString('base64'); + return { type: 'image', mediaType, data, ...(name === undefined ? {} : { name }) }; + } + catch (err) { + await this.enqueueNotice(chatId, MSG.IMAGE_FAILED(this.redact(err))); + return undefined; + } + } + /** + * Album photos arrive as separate updates sharing a media_group_id — debounce + * them into ONE prompt so an album does not spawn N sequential turns. + */ + accumulateMediaGroup(chatId, groupId, parts) { + const key = `${String(chatId)}:${groupId}`; + const existing = this.mediaGroups.get(key); + if (existing) { + clearTimeout(existing.timer); + existing.parts.push(...parts); + } + const entry = existing ?? { chatId, parts: [...parts] }; + entry.timer = setTimeout(() => { + this.mediaGroups.delete(key); + this.sendImageFollowup(entry.chatId, entry.parts).catch((err) => { + this.ctx.logger.error(this.redact(err)); + }); + }, MEDIA_GROUP_WINDOW_MS); + this.mediaGroups.set(key, entry); + } + /** + * Dispatch an image-bearing message to the bound session through the host + * prompt RPC — the exact channel the Web composer uses. The host checks + * whether the current model accepts image input, admits the bytes into the + * durable attachment store, then queues the message on the agent (mode + * 'queue' — same semantics as the text path's agent.followup). + */ + async sendImageFollowup(chatId, parts) { + const binding = this.bindings.get(String(chatId)); + if (!binding) { + await this.client.sendMessage(chatId, MSG.NEED_BIND); + return; + } + const agent = await this.ensureLiveAgent(binding.sessionId); + if (!agent) { + this.bindings.delete(String(chatId)); + await this.client.sendMessage(chatId, MSG.GONE); + return; + } + const proxy = resolveApiProxy(this.ctx); + if (!proxy?.sessions?.prompt) { + await this.enqueueNotice(chatId, '当前宿主不支持会话消息接口（sessions.prompt），无法发送图片。'); + return; + } + const busy = this.busySessions.has(binding.sessionId) + || Boolean(agent.phase?.kind + && agent.phase?.kind !== 'idle'); + let res; + try { + res = await proxy.sessions.prompt({ + rpcId: randomUUID(), + payload: { sessionId: binding.sessionId, mode: 'queue', content: parts }, + }); + } + catch (err) { + await this.enqueueNotice(chatId, MSG.IMAGE_FAILED(this.redact(err))); + return; + } + const result = res?.result; + if (!result?.ok) { + const code = result?.error?.code; + const message = String(result?.error?.message ?? '宿主未确认接收'); + if (code === 'attachment-error' && result?.error?.details?.reason === 'MODEL_DOES_NOT_SUPPORT_IMAGES') { + await this.enqueueNotice(chatId, MSG.IMAGE_MODEL_UNSUPPORTED); + } + else { + const detail = `${code ? `[${code}] ` : ''}${message}`.trim(); + await this.enqueueNotice(chatId, MSG.IMAGE_FAILED(detail)); + } + return; + } + if (busy) { + await this.enqueueNotice(chatId, '已加入队列：当前任务完成后将处理这条消息'); + } + } async pollLoop() { const signal = this.pollAbort?.signal; let errorCount = 0; diff --git a/lib/client.d.ts b/lib/client.d.ts index 54c4177..45b6aa2 100644 --- a/lib/client.d.ts +++ b/lib/client.d.ts @@ -10,12 +10,38 @@ export interface TelegramChat { title?: string; username?: string; } +export interface TelegramPhotoSize { + file_id: string; + file_unique_id: string; + width: number; + height: number; + file_size?: number; +} +export interface TelegramDocument { + file_id: string; + file_unique_id: string; + file_name?: string; + mime_type?: string; + file_size?: number; +} export interface TelegramMessage { message_id: number; date: number; chat: TelegramChat; from?: TelegramUser; text?: string; + /** Media caption (photos/documents); TG puts it here, not in `text`. */ + caption?: string; + photo?: TelegramPhotoSize[]; + document?: TelegramDocument; + /** Album grouping id: several photo updates share one media_group_id. */ + media_group_id?: string; +} +export interface TelegramFile { + file_id: string; + file_unique_id: string; + file_size?: number; + file_path?: string; } export interface TelegramCallbackQuery { id: string; @@ -56,6 +82,10 @@ export interface TelegramClientLike { sendChatAction(chatId: number, action: string): Promise; answerCallbackQuery(callbackQueryId: string, text?: string): Promise; setMyCommands(commands: TelegramBotCommand[]): Promise; + /** Resolve a file_id to a file_path (downloadable URL suffix). */ + getFile(fileId: string): Promise; + /** Download file bytes from the Bot API file endpoint (GET, proxy-aware). */ + downloadFile(filePath: string): Promise; } export declare class TelegramClient implements TelegramClientLike { private readonly token; @@ -72,4 +102,6 @@ export declare class TelegramClient implements TelegramClientLike { sendChatAction(chatId: number, action: string): Promise; answerCallbackQuery(callbackQueryId: string, text?: string): Promise; setMyCommands(commands: TelegramBotCommand[]): Promise; + getFile(fileId: string): Promise; + downloadFile(filePath: string): Promise; } diff --git a/lib/client.js b/lib/client.js index c0e5ea9..b1c0ebe 100644 --- a/lib/client.js +++ b/lib/client.js @@ -104,4 +104,21 @@ export class TelegramClient { async setMyCommands(commands) { return this.call('setMyCommands', { commands }); } + async getFile(fileId) { + return this.call('getFile', { file_id: fileId }); + } + async downloadFile(filePath) { + const url = `${this.baseUrl}/file/bot${this.token}/${filePath}`; + try { + const response = await this.fetchImpl(url); + if (!response.ok) { + throw new Error(`Telegram file download failed: HTTP ${String(response.status)}`); + } + return new Uint8Array(await response.arrayBuffer()); + } + catch (err) { + const message = err instanceof Error ? err.message : String(err); + throw new Error(this.redact(message)); + } + } } diff --git a/lib/commands.d.ts b/lib/commands.d.ts index ef8688a..08cbd11 100644 --- a/lib/commands.d.ts +++ b/lib/commands.d.ts @@ -28,6 +28,9 @@ export declare const MSG: { readonly COMPACT_PERSIST_FAILED: "压缩完成但保存失败。请检查本机存储与权限后重试。"; readonly COMPACT_FAILED: (detail?: string) => string; readonly GONE: "绑定的会话已不可用。请重新 /sessions。"; + readonly MEDIA_UNSUPPORTED: "暂不支持该消息类型（支持文本与 png/jpeg/webp/gif 图片；语音/视频/其他文件请在 Web 端处理）。"; + readonly IMAGE_MODEL_UNSUPPORTED: "当前模型不支持图片输入，请在 Web 端或 /model 切换到多模态模型后重试。"; + readonly IMAGE_FAILED: (detail?: string) => string; readonly LAST_FAILED: "无法读取上次对话。请确认已绑定，且本机 dsh web / apiProxy 可用。"; readonly MODEL_UNAVAILABLE: (detail?: string) => string; readonly MODEL_UNROUTABLE: (current: string) => string; diff --git a/lib/commands.js b/lib/commands.js index 8920d16..8d8056f 100644 --- a/lib/commands.js +++ b/lib/commands.js @@ -16,6 +16,7 @@ export const MSG = { '/help — 显示帮助', '', '绑定后直接发文字，消息会进入该本机会话；Web 与手机看到同一条轨迹。', + '直接发送图片（或带文字说明的图片）也会进入该会话，说明文字会一并作为消息文本。', '仅白名单用户可用。无会话时请先在 dsh web 开对话或保留历史会话。', ].join('\n'), NEED_BIND: '尚未绑定本机会话。请先发送 /sessions 选择一个。', @@ -59,6 +60,14 @@ export const MSG = { return `${tip}\n详情：${detail}`; }, GONE: '绑定的会话已不可用。请重新 /sessions。', + MEDIA_UNSUPPORTED: '暂不支持该消息类型（支持文本与 png/jpeg/webp/gif 图片；语音/视频/其他文件请在 Web 端处理）。', + IMAGE_MODEL_UNSUPPORTED: '当前模型不支持图片输入，请在 Web 端或 /model 切换到多模态模型后重试。', + IMAGE_FAILED(detail) { + const tip = '图片发送失败。'; + if (!detail) + return tip; + return `${tip}\n详情：${detail}`; + }, LAST_FAILED: '无法读取上次对话。请确认已绑定，且本机 dsh web / apiProxy 可用。', MODEL_UNAVAILABLE(detail) { const tip = '无法读取模型列表。请确认已绑定会话，且本机 dsh web 已加载 host-apiproxy。'; diff --git a/src/apiproxy.ts b/src/apiproxy.ts index 3affb53..3a7dcd3 100644 --- a/src/apiproxy.ts +++ b/src/apiproxy.ts @@ -8,6 +8,11 @@ export type ApiProxySessions = { selectModel?: (req: { rpcId: string; payload: unknown }) => Promise /** Create a session (used by /new; payload: workspaceId | cwd). */ create?: (req: { rpcId: string; payload: unknown }) => Promise + /** + * Prompt dispatch (Web 端同通道)：文本 + 图片字节(base64)经宿主附件服务持久化后 + * 进入会话；含图片时会校验当前模型是否支持图像输入。 + */ + prompt?: (req: { rpcId: string; payload: unknown }) => Promise } export type ApiProxyLike = { diff --git a/src/bridge.ts b/src/bridge.ts index 3336109..ee2a4f8 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -1,3 +1,5 @@ +import { Buffer } from 'node:buffer' +import { randomUUID } from 'node:crypto' import { mkdirSync, readFileSync, writeFileSync } from 'node:fs' import { homedir } from 'node:os' import { dirname, join } from 'node:path' @@ -22,6 +24,7 @@ import { TelegramClient, type InlineKeyboardMarkup, type TelegramClientLike, + type TelegramMessage, type TelegramUpdate, } from './client.js' import { MSG, LAST_CB, parseCommand } from './commands.js' @@ -147,6 +150,17 @@ interface CommandsRuntimeLike { /** command-compact 的 CommandResult.kind 值。 */ type CompactCommandKind = 'success' | 'error' +/** Wire prompt content part (mirrors host PromptContentPart): text or base64 image. */ +type WireContentPart = + | { type: 'text'; text: string } + | { type: 'image'; mediaType: string; data: string; name?: string } + +/** Host-admitted raster media types (@deepseek-ai/dsh-attachment). */ +const IMAGE_MEDIA_TYPES: readonly string[] = ['image/png', 'image/jpeg', 'image/webp', 'image/gif'] + +/** Album debounce window: TG emits each photo of an album as its own update. */ +const MEDIA_GROUP_WINDOW_MS = 1200 + /** * 把 command-compact 返回的英文结果翻译成 TG 中文回复。 * 文案以宿主 @deepseek-ai/dsh-command-compact 的固定输出为准；无法识别时返回 @@ -248,6 +262,8 @@ export class TelegramBridge { private readonly compacting = new Set() /** sessionId → compaction abort controller (cancelled on bridge stop) */ private readonly compactAborts = new Map() + /** Album accumulation: chatId:media_group_id → parts buffered into one prompt. */ + private readonly mediaGroups = new Map }>() constructor(ctx: Context, options: TelegramBridgeOptions) { this.ctx = ctx @@ -333,6 +349,8 @@ export class TelegramBridge { for (const controller of this.compactAborts.values()) controller.abort() this.compactAborts.clear() this.compacting.clear() + for (const group of this.mediaGroups.values()) clearTimeout(group.timer) + this.mediaGroups.clear() if (this.hookTimer) { clearTimeout(this.hookTimer) this.hookTimer = undefined @@ -361,9 +379,22 @@ export class TelegramBridge { return } - // Non-text messages: tell the (authorized) user instead of dropping silently. + // Media messages: photos / image documents (optionally with a caption) are + // admitted into the bound session; other media get an explicit notice. + // A caption is treated as accompanying text — never parsed as a command. + const mediaParts = await this.collectMediaParts(chatId, message) + if (mediaParts !== undefined) { + if (mediaParts.length > 0) { + if (message.media_group_id) this.accumulateMediaGroup(chatId, message.media_group_id, mediaParts) + else await this.sendImageFollowup(chatId, mediaParts) + } + return + } + + // Non-text messages without recognized media: tell the (authorized) user + // instead of dropping silently. if (!message.text) { - await this.enqueueNotice(chatId, '暂只支持文本消息（图片/语音/文件等媒体暂不处理）') + await this.enqueueNotice(chatId, MSG.MEDIA_UNSUPPORTED) return } @@ -1043,6 +1074,135 @@ export class TelegramBridge { } } + /** + * Extract wire prompt parts from a media message. + * Returns undefined when the message carries no media at all; returns [] for + * an unsupported media type (the user was already notified); otherwise the + * caption text part (when present) followed by the image part. + */ + private async collectMediaParts(chatId: number, message: TelegramMessage): Promise { + const caption = message.caption?.trim() + const textPart: WireContentPart | undefined = caption ? { type: 'text', text: caption } : undefined + const withText = (part: WireContentPart): WireContentPart[] => + textPart ? [textPart, part] : [part] + + if (message.photo && message.photo.length > 0) { + // TG photo sizes are ordered smallest → largest; the last is the best quality. + const best = message.photo[message.photo.length - 1]! + const part = await this.downloadImagePart(chatId, best.file_id, 'image/jpeg') + return part ? withText(part) : [] + } + + const doc = message.document + if (doc) { + if (!doc.mime_type || !IMAGE_MEDIA_TYPES.includes(doc.mime_type)) { + await this.enqueueNotice(chatId, MSG.MEDIA_UNSUPPORTED) + return [] + } + const part = await this.downloadImagePart(chatId, doc.file_id, doc.mime_type, doc.file_name) + return part ? withText(part) : [] + } + + return undefined + } + + /** Download one TG file and encode it as a canonical-base64 wire image part. */ + private async downloadImagePart( + chatId: number, + fileId: string, + mediaType: string, + name?: string, + ): Promise { + try { + const file = await this.client.getFile(fileId) + if (!file.file_path) throw new Error('TG 未返回文件路径') + const bytes = await this.client.downloadFile(file.file_path) + // Buffer base64 is canonical (standard alphabet, no whitespace) — the host + // attachment admission rejects non-canonical forms outright. + const data = Buffer.from(bytes).toString('base64') + return { type: 'image', mediaType, data, ...(name === undefined ? {} : { name }) } + } catch (err) { + await this.enqueueNotice(chatId, MSG.IMAGE_FAILED(this.redact(err))) + return undefined + } + } + + /** + * Album photos arrive as separate updates sharing a media_group_id — debounce + * them into ONE prompt so an album does not spawn N sequential turns. + */ + private accumulateMediaGroup(chatId: number, groupId: string, parts: WireContentPart[]): void { + const key = `${String(chatId)}:${groupId}` + const existing = this.mediaGroups.get(key) + if (existing) { + clearTimeout(existing.timer) + existing.parts.push(...parts) + } + const entry = existing ?? { chatId, parts: [...parts] } + entry.timer = setTimeout(() => { + this.mediaGroups.delete(key) + this.sendImageFollowup(entry.chatId, entry.parts).catch((err) => { + this.ctx.logger.error(this.redact(err)) + }) + }, MEDIA_GROUP_WINDOW_MS) + this.mediaGroups.set(key, entry) + } + + /** + * Dispatch an image-bearing message to the bound session through the host + * prompt RPC — the exact channel the Web composer uses. The host checks + * whether the current model accepts image input, admits the bytes into the + * durable attachment store, then queues the message on the agent (mode + * 'queue' — same semantics as the text path's agent.followup). + */ + private async sendImageFollowup(chatId: number, parts: WireContentPart[]): Promise { + const binding = this.bindings.get(String(chatId)) + if (!binding) { + await this.client.sendMessage(chatId, MSG.NEED_BIND) + return + } + const agent = await this.ensureLiveAgent(binding.sessionId) + if (!agent) { + this.bindings.delete(String(chatId)) + await this.client.sendMessage(chatId, MSG.GONE) + return + } + const proxy = resolveApiProxy(this.ctx) + if (!proxy?.sessions?.prompt) { + await this.enqueueNotice(chatId, '当前宿主不支持会话消息接口（sessions.prompt），无法发送图片。') + return + } + const busy = this.busySessions.has(binding.sessionId) + || Boolean((agent as { phase?: { kind?: string } }).phase?.kind + && (agent as { phase?: { kind?: string } }).phase?.kind !== 'idle') + type PromptResponse = { result?: { ok?: boolean; value?: { accepted?: boolean }; error?: { code?: string; message?: string; details?: { reason?: string } } } } + let res: PromptResponse | undefined + try { + res = await proxy.sessions.prompt({ + rpcId: randomUUID(), + payload: { sessionId: binding.sessionId, mode: 'queue', content: parts }, + }) as PromptResponse | undefined + } catch (err) { + await this.enqueueNotice(chatId, MSG.IMAGE_FAILED(this.redact(err))) + return + } + const result = res?.result + if (!result?.ok) { + const code = result?.error?.code + const message = String(result?.error?.message ?? '宿主未确认接收') + if (code === 'attachment-error' && result?.error?.details?.reason === 'MODEL_DOES_NOT_SUPPORT_IMAGES') { + await this.enqueueNotice(chatId, MSG.IMAGE_MODEL_UNSUPPORTED) + } else { + const detail = `${code ? `[${code}] ` : ''}${message}`.trim() + await this.enqueueNotice(chatId, MSG.IMAGE_FAILED(detail)) + } + return + } + if (busy) { + await this.enqueueNotice(chatId, '已加入队列：当前任务完成后将处理这条消息') + } + } + private async pollLoop(): Promise { const signal = this.pollAbort?.signal let errorCount = 0 diff --git a/src/client.ts b/src/client.ts index 757d113..754bf87 100644 --- a/src/client.ts +++ b/src/client.ts @@ -14,12 +14,41 @@ export interface TelegramChat { username?: string } +export interface TelegramPhotoSize { + file_id: string + file_unique_id: string + width: number + height: number + file_size?: number +} + +export interface TelegramDocument { + file_id: string + file_unique_id: string + file_name?: string + mime_type?: string + file_size?: number +} + export interface TelegramMessage { message_id: number date: number chat: TelegramChat from?: TelegramUser text?: string + /** Media caption (photos/documents); TG puts it here, not in `text`. */ + caption?: string + photo?: TelegramPhotoSize[] + document?: TelegramDocument + /** Album grouping id: several photo updates share one media_group_id. */ + media_group_id?: string +} + +export interface TelegramFile { + file_id: string + file_unique_id: string + file_size?: number + file_path?: string } export interface TelegramCallbackQuery { @@ -73,6 +102,10 @@ export interface TelegramClientLike { sendChatAction(chatId: number, action: string): Promise answerCallbackQuery(callbackQueryId: string, text?: string): Promise setMyCommands(commands: TelegramBotCommand[]): Promise + /** Resolve a file_id to a file_path (downloadable URL suffix). */ + getFile(fileId: string): Promise + /** Download file bytes from the Bot API file endpoint (GET, proxy-aware). */ + downloadFile(filePath: string): Promise } function resolveProxyUrl(): string | undefined { @@ -197,4 +230,22 @@ export class TelegramClient implements TelegramClientLike { async setMyCommands(commands: TelegramBotCommand[]): Promise { return this.call('setMyCommands', { commands }) } + + async getFile(fileId: string): Promise { + return this.call('getFile', { file_id: fileId }) + } + + async downloadFile(filePath: string): Promise { + const url = `${this.baseUrl}/file/bot${this.token}/${filePath}` + try { + const response = await this.fetchImpl(url) + if (!response.ok) { + throw new Error(`Telegram file download failed: HTTP ${String(response.status)}`) + } + return new Uint8Array(await response.arrayBuffer()) + } catch (err) { + const message = err instanceof Error ? err.message : String(err) + throw new Error(this.redact(message)) + } + } } diff --git a/src/commands.ts b/src/commands.ts index 8a7e6ee..4e18a2e 100644 --- a/src/commands.ts +++ b/src/commands.ts @@ -16,6 +16,7 @@ export const MSG = { '/help — 显示帮助', '', '绑定后直接发文字，消息会进入该本机会话；Web 与手机看到同一条轨迹。', + '直接发送图片（或带文字说明的图片）也会进入该会话，说明文字会一并作为消息文本。', '仅白名单用户可用。无会话时请先在 dsh web 开对话或保留历史会话。', ].join('\n'), NEED_BIND: '尚未绑定本机会话。请先发送 /sessions 选择一个。', @@ -58,6 +59,13 @@ export const MSG = { return `${tip}\n详情：${detail}` }, GONE: '绑定的会话已不可用。请重新 /sessions。', + MEDIA_UNSUPPORTED: '暂不支持该消息类型（支持文本与 png/jpeg/webp/gif 图片；语音/视频/其他文件请在 Web 端处理）。', + IMAGE_MODEL_UNSUPPORTED: '当前模型不支持图片输入，请在 Web 端或 /model 切换到多模态模型后重试。', + IMAGE_FAILED(detail?: string): string { + const tip = '图片发送失败。' + if (!detail) return tip + return `${tip}\n详情：${detail}` + }, LAST_FAILED: '无法读取上次对话。请确认已绑定，且本机 dsh web / apiProxy 可用。', MODEL_UNAVAILABLE(detail?: string): string { const tip = '无法读取模型列表。请确认已绑定会话，且本机 dsh web 已加载 host-apiproxy。' diff --git a/tests/bridge.test.ts b/tests/bridge.test.ts index 5b33af4..2d9606d 100644 --- a/tests/bridge.test.ts +++ b/tests/bridge.test.ts @@ -28,6 +28,8 @@ function fakeClient( sendChatAction: async () => true, answerCallbackQuery: async () => true, setMyCommands: async () => true, + getFile: async (fileId: string) => ({ file_id: fileId, file_unique_id: fileId, file_path: `path/${fileId}` }), + downloadFile: async () => new Uint8Array([1, 2, 3]), ...overrides, } } @@ -1053,3 +1055,257 @@ test('/compact without bind prompts NEED_BIND', async () => { await bridge.processUpdate(messageUpdate(10, 1, '/compact')) assert.equal(sent[0]?.text, MSG.NEED_BIND) }) + +// ── 图片消息 ── + +type PromptPayload = { sessionId: string; mode: string; content: Array<{ type: string; text?: string; mediaType?: string; data?: string; name?: string }> } + +function photoUpdate(chatId: number, userId: number, fileId: string, opts?: { + caption?: string + mediaGroupId?: string + updateId?: number +}): TelegramUpdate { + const updateId = opts?.updateId ?? 1 + return { + update_id: updateId, + message: { + message_id: updateId, + date: 0, + chat: { id: chatId, type: 'private' }, + from: { id: userId }, + caption: opts?.caption, + media_group_id: opts?.mediaGroupId, + photo: [ + { file_id: `${fileId}-small`, file_unique_id: `${fileId}-small`, width: 90, height: 90 }, + { file_id: fileId, file_unique_id: fileId, width: 1280, height: 960, file_size: 300000 }, + ], + }, + } +} + +function bindUpdate(chatId: number, userId: number, sessionId: string): TelegramUpdate { + return { + update_id: 1, + callback_query: { + id: 'bind', + from: { id: userId }, + message: { message_id: 1, date: 0, chat: { id: chatId, type: 'private' } }, + data: `${BIND_CB_PREFIX}${sessionId}`, + }, + } +} + +function imageBridgeCtx(sessionId: string, prompts: PromptPayload[], promptResult?: unknown) { + const agent = makeAgent(sessionId, []) + return { + agent, + ctx: { + logger: { info() {}, warn() {}, error() {} }, + agents: { + list: () => [agent], + roots: () => [agent], + get: (id: ReturnType) => (String(id) === sessionId ? agent : undefined), + }, + apiProxy: { + sessions: { + prompt: async (req: { payload: unknown }) => { + prompts.push(req.payload as PromptPayload) + return promptResult ?? rpcOk({ accepted: true }) + }, + }, + }, + on() { return () => {} }, + }, + } +} + +test('photo with caption is admitted via sessions.prompt (largest size, jpeg, base64)', async () => { + const sent: SentMessage[] = [] + const prompts: PromptPayload[] = [] + const sessionId = 'live-img' + const { ctx } = imageBridgeCtx(sessionId, prompts) + const jpegBytes = new Uint8Array([0xff, 0xd8, 1, 2, 3]) + const bridge = new TelegramBridge(ctx as any, { + token: 't', + allowedUserIds: [1], + allowAllUsers: false, + client: fakeClient(sent, { + getFile: async (fileId: string) => ({ file_id: fileId, file_unique_id: fileId, file_path: `dir/${fileId}` }), + downloadFile: async (filePath: string) => { + assert.ok(filePath.includes('big-photo'), 'downloads the largest photo size') + return jpegBytes + }, + }), + sleep: async () => {}, + }) + await bridge.processUpdate(bindUpdate(10, 1, sessionId)) + await bridge.processUpdate(photoUpdate(10, 1, 'big-photo', { caption: '看看这张图', updateId: 2 })) + assert.equal(prompts.length, 1, 'one prompt RPC') + const content = prompts[0]!.content + assert.deepEqual(content[0], { type: 'text', text: '看看这张图' }) + assert.equal(content[1]?.type, 'image') + assert.equal(content[1]?.mediaType, 'image/jpeg') + assert.equal(content[1]?.data, Buffer.from(jpegBytes).toString('base64')) + assert.ok(!sent.some((m) => m.text === MSG.MEDIA_UNSUPPORTED), 'no unsupported notice') +}) + +test('image document (png) is admitted with its file name', async () => { + const sent: SentMessage[] = [] + const prompts: PromptPayload[] = [] + const sessionId = 'live-img-doc' + const { ctx } = imageBridgeCtx(sessionId, prompts) + const bridge = new TelegramBridge(ctx as any, { + token: 't', + allowedUserIds: [1], + allowAllUsers: false, + client: fakeClient(sent), + sleep: async () => {}, + }) + await bridge.processUpdate(bindUpdate(10, 1, sessionId)) + await bridge.processUpdate({ + update_id: 2, + message: { + message_id: 2, + date: 0, + chat: { id: 10, type: 'private' }, + from: { id: 1 }, + document: { + file_id: 'doc-png', + file_unique_id: 'doc-png', + file_name: 'screenshot.png', + mime_type: 'image/png', + file_size: 5000, + }, + }, + }) + assert.equal(prompts.length, 1) + const image = prompts[0]!.content.find((p) => p.type === 'image') + assert.equal(image?.mediaType, 'image/png') + assert.equal(image?.name, 'screenshot.png') +}) + +test('non-image document gets MEDIA_UNSUPPORTED and no prompt', async () => { + const sent: SentMessage[] = [] + const prompts: PromptPayload[] = [] + const sessionId = 'live-img-pdf' + const { ctx } = imageBridgeCtx(sessionId, prompts) + const bridge = new TelegramBridge(ctx as any, { + token: 't', + allowedUserIds: [1], + allowAllUsers: false, + client: fakeClient(sent), + sleep: async () => {}, + }) + await bridge.processUpdate(bindUpdate(10, 1, sessionId)) + sent.length = 0 + await bridge.processUpdate({ + update_id: 2, + message: { + message_id: 2, + date: 0, + chat: { id: 10, type: 'private' }, + from: { id: 1 }, + document: { file_id: 'doc-pdf', file_unique_id: 'doc-pdf', file_name: 'report.pdf', mime_type: 'application/pdf' }, + }, + }) + assert.equal(prompts.length, 0, 'no prompt dispatched') + assert.ok(sent.some((m) => m.text === MSG.MEDIA_UNSUPPORTED), 'unsupported notice sent') +}) + +test('album photos collapse into a single prompt with all images', async () => { + const sent: SentMessage[] = [] + const prompts: PromptPayload[] = [] + const sessionId = 'live-img-album' + const { ctx } = imageBridgeCtx(sessionId, prompts) + const bridge = new TelegramBridge(ctx as any, { + token: 't', + allowedUserIds: [1], + allowAllUsers: false, + client: fakeClient(sent), + sleep: async () => {}, + }) + await bridge.processUpdate(bindUpdate(10, 1, sessionId)) + await bridge.processUpdate(photoUpdate(10, 1, 'album-1', { mediaGroupId: 'g1', updateId: 2 })) + await bridge.processUpdate(photoUpdate(10, 1, 'album-2', { mediaGroupId: 'g1', updateId: 3 })) + for (let i = 0; i < 500 && prompts.length === 0; i++) { + await new Promise((r) => setTimeout(r, 10)) + } + assert.equal(prompts.length, 1, 'album debounced into one prompt') + assert.equal(prompts[0]!.content.filter((p) => p.type === 'image').length, 2) +}) + +test('model without image support replies IMAGE_MODEL_UNSUPPORTED', async () => { + const sent: SentMessage[] = [] + const prompts: PromptPayload[] = [] + const sessionId = 'live-img-nv' + const { ctx } = imageBridgeCtx(sessionId, prompts, { + rpcId: 'x', + result: { + ok: false, + error: { + code: 'attachment-error', + message: 'Model "deepseek-chat" does not support image input.', + details: { reason: 'MODEL_DOES_NOT_SUPPORT_IMAGES' }, + }, + }, + }) + const bridge = new TelegramBridge(ctx as any, { + token: 't', + allowedUserIds: [1], + allowAllUsers: false, + client: fakeClient(sent), + sleep: async () => {}, + }) + await bridge.processUpdate(bindUpdate(10, 1, sessionId)) + sent.length = 0 + await bridge.processUpdate(photoUpdate(10, 1, 'nv-photo', { updateId: 2 })) + for (let i = 0; i < 100 && sent.length === 0; i++) { + await new Promise((r) => setTimeout(r, 5)) + } + assert.ok(sent.some((m) => m.text === MSG.IMAGE_MODEL_UNSUPPORTED), 'model unsupported notice') +}) + +test('photo download failure replies IMAGE_FAILED', async () => { + const sent: SentMessage[] = [] + const prompts: PromptPayload[] = [] + const sessionId = 'live-img-dl' + const { ctx } = imageBridgeCtx(sessionId, prompts) + const bridge = new TelegramBridge(ctx as any, { + token: 't', + allowedUserIds: [1], + allowAllUsers: false, + client: fakeClient(sent, { + getFile: async () => { throw new Error('telegram timeout') }, + }), + sleep: async () => {}, + }) + await bridge.processUpdate(bindUpdate(10, 1, sessionId)) + sent.length = 0 + await bridge.processUpdate(photoUpdate(10, 1, 'dl-photo', { updateId: 2 })) + for (let i = 0; i < 100 && sent.length === 0; i++) { + await new Promise((r) => setTimeout(r, 5)) + } + assert.equal(prompts.length, 0, 'no prompt dispatched') + assert.ok(sent.some((m) => m.text.startsWith('图片发送失败。')), 'failure notice') +}) + +test('photo without bind prompts NEED_BIND and no prompt', async () => { + const sent: SentMessage[] = [] + const prompts: PromptPayload[] = [] + const ctx = { + logger: { info() {}, warn() {}, error() {} }, + agents: { list: () => [], roots: () => [], get: () => undefined }, + apiProxy: { sessions: { prompt: async (req: { payload: unknown }) => { prompts.push(req.payload as PromptPayload); return rpcOk({ accepted: true }) } } }, + on() { return () => {} }, + } + const bridge = new TelegramBridge(ctx as any, { + token: 't', + allowedUserIds: [1], + allowAllUsers: false, + client: fakeClient(sent), + sleep: async () => {}, + }) + await bridge.processUpdate(photoUpdate(10, 1, 'unbound-photo', { updateId: 1 })) + assert.equal(prompts.length, 0) + assert.ok(sent.some((m) => m.text === MSG.NEED_BIND), 'NEED_BIND sent') +})