mirror of
https://github.com/NanmiCoder/claude-code-haha.git
synced 2026-10-10 03:43:11 +08:00
程序员阿江-Relakkes
729 lines
27 KiB
TypeScript
729 lines
27 KiB
TypeScript
import { adapterMigrationLifecycle, registerAdapterShutdown } from '../common/migration-lifecycle.js'
|
|
/**
|
|
* Telegram Adapter for Claude Code Desktop
|
|
*
|
|
* 基于 grammY 的轻量 Telegram Bot，直连服务端 /ws/:sessionId。
|
|
* 启动：TELEGRAM_BOT_TOKEN=xxx bun run telegram/index.ts
|
|
*/
|
|
|
|
import { Bot, InlineKeyboard, type Context } from 'grammy'
|
|
import * as path from 'node:path'
|
|
import { WsBridge, type ServerMessage } from '../common/ws-bridge.js'
|
|
import { MessageDedup } from '../common/message-dedup.js'
|
|
import { enqueue } from '../common/chat-queue.js'
|
|
import { loadConfig } from '../common/config.js'
|
|
import {
|
|
formatImStatus,
|
|
formatPermissionRequest,
|
|
} from '../common/format.js'
|
|
import {
|
|
buildTelegramThinkingUpdate,
|
|
} from './format.js'
|
|
import { TelegramStreamDelivery } from './stream-delivery.js'
|
|
import {
|
|
formatPermissionDecisionStatus,
|
|
formatPermissionInstructions,
|
|
parsePermissionCommand,
|
|
parsePermitCallbackData,
|
|
type PermissionDecision,
|
|
} from '../common/permission.js'
|
|
import { SessionStore } from '../common/session-store.js'
|
|
import { createAdapterClient } from '../common/adapter-client.js'
|
|
import { restoreStoredSessionBinding, SESSION_RECONNECT_NOTICE, type SessionRestoreResult } from '../common/session-recovery.js'
|
|
import { SessionSelectionController } from '../common/session-selection.js'
|
|
import { syncImPermissionState } from '../common/permission-sync.js'
|
|
import { isAllowedUser, tryPair } from '../common/pairing.js'
|
|
import { TelegramMediaService } from './media.js'
|
|
import { AttachmentStore } from '../common/attachment/attachment-store.js'
|
|
import { checkAttachmentLimit } from '../common/attachment/attachment-limits.js'
|
|
import type { AttachmentRef } from '../common/ws-bridge.js'
|
|
import { ImageBlockWatcher } from '../common/attachment/image-block-watcher.js'
|
|
import type { PendingUpload } from '../common/attachment/attachment-types.js'
|
|
import { sendSafeOutboundImage } from '../common/attachment/outbound-image.js'
|
|
import { syncTelegramBotCommands } from './menu.js'
|
|
import { createTelegramRuntimeCommandController, registerAuthorizedTelegramCommand, registerTelegramExtendedCommands, registerTelegramSessionCommands, shouldProcessTelegramMessage, tryHandleTelegramSelectionCallback, tryHandleTelegramSessionInput } from './commands.js'
|
|
|
|
// ---------- init ----------
|
|
|
|
const config = loadConfig()
|
|
if (!config.telegram.botToken) {
|
|
console.error('[Telegram] Missing TELEGRAM_BOT_TOKEN. Set env or ~/.claude/adapters.json')
|
|
process.exit(1)
|
|
}
|
|
|
|
export const bot = new Bot(config.telegram.botToken)
|
|
bot.use((_ctx, next) => adapterMigrationLifecycle.isQuiescing ? Promise.resolve() : adapterMigrationLifecycle.track(next()))
|
|
const bridge = new WsBridge(config.serverUrl, 'tg')
|
|
const streamDelivery = new TelegramStreamDelivery(bot.api)
|
|
const dedup = new MessageDedup()
|
|
const sessionStore = new SessionStore()
|
|
const { httpClient, defaultWorkDir } = createAdapterClient(config, config.telegram)
|
|
const attachmentStore = new AttachmentStore()
|
|
const media = new TelegramMediaService(bot, attachmentStore)
|
|
|
|
const accumulatedThinkingText = new Map<string, string>()
|
|
// Track chats waiting for project selection
|
|
const pendingProjectSelection = new Map<string, boolean>()
|
|
const runtimeStates = new Map<string, ChatRuntimeState>()
|
|
const pendingPermissions = new Map<string, Set<string>>()
|
|
/** Per-chat outbound image watcher for Agent-produced markdown images. */
|
|
const tgImageWatchers = new Map<string, ImageBlockWatcher>()
|
|
|
|
function getTgWatcher(chatId: string): ImageBlockWatcher {
|
|
let w = tgImageWatchers.get(chatId)
|
|
if (!w) {
|
|
w = new ImageBlockWatcher()
|
|
tgImageWatchers.set(chatId, w)
|
|
}
|
|
return w
|
|
}
|
|
|
|
type ChatRuntimeState = {
|
|
state: 'idle' | 'thinking' | 'streaming' | 'tool_executing' | 'permission_pending'
|
|
verb?: string
|
|
model?: string
|
|
pendingPermissionCount: number
|
|
}
|
|
|
|
const isChatBusy = (chatId: string) => getRuntimeState(chatId).state !== 'idle' || Boolean(pendingPermissions.get(chatId)?.size)
|
|
const commandController = createTelegramRuntimeCommandController({
|
|
botApi: bot.api, httpClient, defaultWorkDir, bridge, sessionStore,
|
|
ensureExistingSession, clearTransientChatState,
|
|
clearOtherSelections: (chatId) => {
|
|
pendingProjectSelection.delete(chatId)
|
|
sessionSelection.clear(chatId)
|
|
},
|
|
isBusy: isChatBusy,
|
|
isAllowedUser: (userId) => isAllowedUser('telegram', userId),
|
|
handleServerMessage: (chatId, msg) => handleServerMessage(chatId, msg as ServerMessage),
|
|
setRuntimeModel: (chatId, modelId) => { getRuntimeState(chatId).model = modelId },
|
|
setRuntimeBusy: (chatId) => { getRuntimeState(chatId).state = 'thinking' },
|
|
})
|
|
const sessionSelection = new SessionSelectionController({
|
|
httpClient, bridge, sessionStore,
|
|
sendNotice: async (chatId, text) => { await bot.api.sendMessage(Number(chatId), text) },
|
|
onServerMessage: handleServerMessage,
|
|
clearTransientState: clearTransientChatState,
|
|
clearProjectSelection: (chatId) => {
|
|
pendingProjectSelection.delete(chatId)
|
|
commandController.clearPendingSelections(chatId)
|
|
},
|
|
isBusy: isChatBusy,
|
|
})
|
|
|
|
// ---------- helpers ----------
|
|
|
|
function getRuntimeState(chatId: string): ChatRuntimeState {
|
|
let state = runtimeStates.get(chatId)
|
|
if (!state) {
|
|
state = { state: 'idle', pendingPermissionCount: 0 }
|
|
runtimeStates.set(chatId, state)
|
|
}
|
|
return state
|
|
}
|
|
|
|
function clearTransientChatState(chatId: string): void {
|
|
streamDelivery.clear(chatId)
|
|
accumulatedThinkingText.delete(chatId)
|
|
const runtime = getRuntimeState(chatId)
|
|
runtime.state = 'idle'
|
|
runtime.verb = undefined
|
|
runtime.pendingPermissionCount = 0
|
|
pendingPermissions.delete(chatId)
|
|
tgImageWatchers.delete(chatId)
|
|
}
|
|
|
|
async function handlePermissionDecision(chatId: string, decision: PermissionDecision): Promise<void> {
|
|
const pending = pendingPermissions.get(chatId)
|
|
if (!pending?.has(decision.requestId)) {
|
|
await bot.api.sendMessage(Number(chatId), `未找到待确认的权限请求：${decision.requestId}`)
|
|
return
|
|
}
|
|
|
|
const sent = bridge.sendPermissionResponse(chatId, decision.requestId, decision.allowed, decision.rule)
|
|
if (sent) {
|
|
pending.delete(decision.requestId)
|
|
const runtime = getRuntimeState(chatId)
|
|
runtime.pendingPermissionCount = Math.max(0, runtime.pendingPermissionCount - 1)
|
|
}
|
|
await bot.api.sendMessage(
|
|
Number(chatId),
|
|
sent ? `${formatPermissionDecisionStatus(decision)}。` : '权限响应发送失败，请检查会话状态。',
|
|
)
|
|
}
|
|
|
|
async function ensureExistingSession(chatId: string): Promise<SessionRestoreResult> {
|
|
return await restoreStoredSessionBinding({
|
|
chatId,
|
|
bridge,
|
|
sessionStore,
|
|
httpClient,
|
|
onServerMessage: (msg) => handleServerMessage(chatId, msg),
|
|
logPrefix: '[Telegram]',
|
|
clearTransientState: () => clearTransientChatState(chatId),
|
|
})
|
|
}
|
|
|
|
async function buildStatusText(chatId: string): Promise<string> {
|
|
const result = await ensureExistingSession(chatId)
|
|
if (result.status === 'unavailable') return SESSION_RECONNECT_NOTICE
|
|
if (result.status === 'missing') return formatImStatus(null)
|
|
const stored = result.session
|
|
|
|
const runtime = getRuntimeState(chatId)
|
|
let projectName = path.basename(stored.workDir) || stored.workDir
|
|
let branch: string | null = null
|
|
|
|
try {
|
|
const gitInfo = await httpClient.getGitInfo(stored.sessionId)
|
|
projectName = gitInfo.repoName || path.basename(gitInfo.workDir) || projectName
|
|
branch = gitInfo.branch
|
|
} catch {
|
|
// Ignore git lookup failures and fall back to stored workDir
|
|
}
|
|
|
|
let taskCounts:
|
|
| {
|
|
total: number
|
|
pending: number
|
|
inProgress: number
|
|
completed: number
|
|
}
|
|
| undefined
|
|
|
|
try {
|
|
const tasks = await httpClient.getTasksForSession(stored.sessionId)
|
|
if (tasks.length > 0) {
|
|
taskCounts = {
|
|
total: tasks.length,
|
|
pending: tasks.filter((task) => task.status === 'pending').length,
|
|
inProgress: tasks.filter((task) => task.status === 'in_progress').length,
|
|
completed: tasks.filter((task) => task.status === 'completed').length,
|
|
}
|
|
}
|
|
} catch {
|
|
// Ignore task lookup failures in IM status summary
|
|
}
|
|
|
|
return formatImStatus({
|
|
sessionId: stored.sessionId,
|
|
projectName,
|
|
branch,
|
|
model: runtime.model,
|
|
state: runtime.state,
|
|
verb: runtime.verb,
|
|
pendingPermissionCount: runtime.pendingPermissionCount,
|
|
taskCounts,
|
|
})
|
|
}
|
|
|
|
// ---------- session management ----------
|
|
|
|
async function ensureSession(chatId: string): Promise<boolean> {
|
|
const result = await ensureExistingSession(chatId)
|
|
if (result.status === 'restored') return true
|
|
if (result.status === 'unavailable') {
|
|
await bot.api.sendMessage(Number(chatId), SESSION_RECONNECT_NOTICE)
|
|
return false
|
|
}
|
|
|
|
const workDir = defaultWorkDir
|
|
if (workDir) {
|
|
return await createSessionForChat(chatId, workDir)
|
|
}
|
|
|
|
await showProjectPicker(chatId)
|
|
return false
|
|
}
|
|
|
|
async function createSessionForChat(chatId: string, workDir: string): Promise<boolean> {
|
|
const numericChatId = Number(chatId)
|
|
try {
|
|
// Always tear down any stale WS connection before creating a new session.
|
|
// Without this, bridge.connectSession() below would short-circuit when an
|
|
// old OPEN connection still exists, leaving messages routed to the old session.
|
|
bridge.resetSession(chatId)
|
|
|
|
const sessionId = await httpClient.createSession(workDir)
|
|
sessionStore.set(chatId, sessionId, workDir)
|
|
bridge.connectSession(chatId, sessionId)
|
|
bridge.onServerMessage(chatId, (msg) => handleServerMessage(chatId, msg))
|
|
const opened = await bridge.waitForOpen(chatId)
|
|
if (!opened) {
|
|
await bot.api.sendMessage(numericChatId, '⚠️ 连接服务器超时，请重试。')
|
|
return false
|
|
}
|
|
return true
|
|
} catch (err) {
|
|
await bot.api.sendMessage(numericChatId,
|
|
`❌ 无法创建会话: ${err instanceof Error ? err.message : String(err)}`)
|
|
return false
|
|
}
|
|
}
|
|
|
|
async function showProjectPicker(chatId: string): Promise<void> {
|
|
sessionSelection.clear(chatId)
|
|
commandController.clearPendingSelections(chatId)
|
|
pendingProjectSelection.delete(chatId)
|
|
const numericChatId = Number(chatId)
|
|
try {
|
|
const projects = await httpClient.listRecentProjects()
|
|
if (projects.length === 0) {
|
|
await bot.api.sendMessage(numericChatId,
|
|
`没有找到最近的项目。发送 /new 会使用默认工作目录：${defaultWorkDir}\n也可以发送 /new /path/to/project 指定项目。`)
|
|
return
|
|
}
|
|
|
|
const lines = projects.slice(0, 10).map((p, i) =>
|
|
`${i + 1}. ${p.projectName}${p.branch ? ` (${p.branch})` : ''}\n ${p.realPath}`
|
|
)
|
|
pendingProjectSelection.set(chatId, true)
|
|
await bot.api.sendMessage(numericChatId,
|
|
`选择项目（回复编号）：\n\n${lines.join('\n\n')}\n\n💡 下次可直接 /new <编号、名称或绝对路径> 快速新建会话`)
|
|
} catch (err) {
|
|
await bot.api.sendMessage(numericChatId,
|
|
`❌ 无法获取项目列表: ${err instanceof Error ? err.message : String(err)}`)
|
|
}
|
|
}
|
|
|
|
// ---------- outbound media dispatch ----------
|
|
|
|
/** Upload a PendingUpload found in streaming output and send it via
|
|
* bot.api.sendPhoto as an independent message. Runs fire-and-forget
|
|
* from the stream handler so streaming text isn't blocked. */
|
|
async function dispatchOutboundMedia(chatId: string, pending: PendingUpload): Promise<void> {
|
|
const numericChatId = Number(chatId)
|
|
try {
|
|
const loaded = await sendSafeOutboundImage(pending, sessionStore.get(chatId)?.workDir, (buffer) => media.sendPhoto(numericChatId, buffer, pending.alt))
|
|
if (!loaded.ok) console.warn('[Telegram] Outbound image rejected:', loaded.reason)
|
|
} catch (err) {
|
|
console.error(
|
|
'[Telegram] dispatchOutboundMedia failed:',
|
|
err instanceof Error ? err.message : err,
|
|
)
|
|
}
|
|
}
|
|
|
|
// ---------- server message handler ----------
|
|
|
|
async function handleServerMessage(chatId: string, msg: ServerMessage): Promise<void> {
|
|
const numericChatId = Number(chatId)
|
|
const runtime = getRuntimeState(chatId)
|
|
|
|
if (syncImPermissionState(chatId, msg, runtime, pendingPermissions)) return
|
|
switch (msg.type) {
|
|
case 'connected':
|
|
break
|
|
|
|
case 'status':
|
|
runtime.state = msg.state
|
|
runtime.verb = typeof msg.verb === 'string' ? msg.verb : undefined
|
|
if (msg.state === 'thinking' && !streamDelivery.hasState(chatId)) {
|
|
await streamDelivery.ensurePlaceholder(chatId, '💭 思考中...')
|
|
accumulatedThinkingText.set(chatId, '')
|
|
}
|
|
break
|
|
|
|
case 'content_start':
|
|
if (msg.blockType === 'text') {
|
|
accumulatedThinkingText.delete(chatId)
|
|
await streamDelivery.handleEvent(chatId, { type: 'content_start', blockType: msg.blockType })
|
|
} else if (msg.blockType === 'tool_use') {
|
|
// Finalize current text placeholder before tool calls,
|
|
// so text after tools gets a fresh message
|
|
await streamDelivery.complete(chatId)
|
|
}
|
|
break
|
|
|
|
case 'content_delta':
|
|
if (msg.text) {
|
|
accumulatedThinkingText.delete(chatId)
|
|
await streamDelivery.handleEvent(chatId, { type: 'content_delta', text: msg.text })
|
|
const newUploads = getTgWatcher(chatId).feed(msg.text)
|
|
for (const pending of newUploads) {
|
|
void dispatchOutboundMedia(chatId, pending)
|
|
}
|
|
}
|
|
break
|
|
|
|
case 'thinking':
|
|
if (streamDelivery.getPlaceholderMessageId(chatId) !== undefined) {
|
|
const update = buildTelegramThinkingUpdate(
|
|
accumulatedThinkingText.get(chatId) ?? '',
|
|
msg.text,
|
|
)
|
|
accumulatedThinkingText.set(chatId, update.fullText)
|
|
try {
|
|
await bot.api.editMessageText(
|
|
numericChatId,
|
|
streamDelivery.getPlaceholderMessageId(chatId)!,
|
|
update.messageText,
|
|
)
|
|
} catch { /* ignore */ }
|
|
}
|
|
break
|
|
|
|
case 'tool_use_complete':
|
|
// Tool details are noise for IM users; visible in Desktop if needed.
|
|
break
|
|
|
|
case 'tool_result':
|
|
// Tool errors are handled internally by the AI (retries etc.)
|
|
// No need to notify the user for every failed attempt.
|
|
break
|
|
|
|
case 'permission_request': {
|
|
runtime.pendingPermissionCount += 1
|
|
runtime.state = 'permission_pending'
|
|
const pending = pendingPermissions.get(chatId) ?? new Set<string>()
|
|
pending.add(msg.requestId)
|
|
pendingPermissions.set(chatId, pending)
|
|
const text = `${formatPermissionRequest(msg.toolName, msg.input, msg.requestId)}\n\n${formatPermissionInstructions(msg.requestId)}`
|
|
const keyboard = new InlineKeyboard()
|
|
.text('✅ 允许', `permit:${msg.requestId}:yes`)
|
|
.text('♾️ 永久允许', `permit:${msg.requestId}:always`)
|
|
.row()
|
|
.text('❌ 拒绝', `permit:${msg.requestId}:no`)
|
|
await bot.api.sendMessage(numericChatId, text, { reply_markup: keyboard })
|
|
break
|
|
}
|
|
|
|
case 'message_complete':
|
|
runtime.state = 'idle'
|
|
runtime.verb = undefined
|
|
await streamDelivery.handleEvent(chatId, { type: 'message_complete' })
|
|
accumulatedThinkingText.delete(chatId)
|
|
break
|
|
|
|
case 'error':
|
|
runtime.state = 'idle'
|
|
runtime.verb = undefined
|
|
accumulatedThinkingText.delete(chatId)
|
|
// Auto-recover from stale thinking block signatures by creating a fresh session.
|
|
// This happens when the API key or provider changed since the session was created.
|
|
if (msg.message && /Invalid.*signature.*thinking/i.test(msg.message)) {
|
|
const stored = sessionStore.get(chatId)
|
|
const workDir = stored?.workDir || defaultWorkDir
|
|
if (workDir) {
|
|
await bot.api.sendMessage(numericChatId, '⚠️ 会话上下文已失效，正在自动重建...')
|
|
clearTransientChatState(chatId)
|
|
bridge.resetSession(chatId)
|
|
sessionStore.delete(chatId)
|
|
const ok = await createSessionForChat(chatId, workDir)
|
|
if (ok) {
|
|
await bot.api.sendMessage(numericChatId, '✅ 已重建会话，请重新发送消息。')
|
|
} else {
|
|
await bot.api.sendMessage(numericChatId, '❌ 重建会话失败，请发送 /new 手动新建。')
|
|
}
|
|
} else {
|
|
await bot.api.sendMessage(numericChatId, '⚠️ 会话上下文已失效，请发送 /new 新建会话。')
|
|
}
|
|
} else {
|
|
await bot.api.sendMessage(numericChatId, `❌ ${msg.message}`)
|
|
}
|
|
break
|
|
|
|
case 'system_notification':
|
|
if (msg.subtype === 'init' && msg.data && typeof msg.data === 'object') {
|
|
const model = (msg.data as Record<string, unknown>).model
|
|
if (typeof model === 'string' && model.trim()) {
|
|
runtime.model = model
|
|
}
|
|
}
|
|
break
|
|
}
|
|
}
|
|
|
|
// ---------- bot handlers ----------
|
|
|
|
registerTelegramExtendedCommands(bot, commandController)
|
|
registerTelegramSessionCommands(bot, (ctx, text) => routeUserMessage(ctx as Context, text, []))
|
|
|
|
/** Reset session state and start a new session for chatId.
|
|
* If `query` is provided, match a project by index or name;
|
|
* otherwise use the configured/default work directory. */
|
|
async function startNewSession(chatId: string, query?: string): Promise<void> {
|
|
const numericChatId = Number(chatId)
|
|
|
|
bridge.resetSession(chatId)
|
|
sessionStore.delete(chatId)
|
|
streamDelivery.clear(chatId)
|
|
pendingProjectSelection.delete(chatId)
|
|
sessionSelection.clear(chatId)
|
|
commandController.clearPendingSelections(chatId)
|
|
pendingPermissions.delete(chatId)
|
|
runtimeStates.delete(chatId)
|
|
tgImageWatchers.delete(chatId)
|
|
|
|
if (query) {
|
|
try {
|
|
const { project, ambiguous } = await httpClient.matchProject(query)
|
|
if (project) {
|
|
const ok = await createSessionForChat(chatId, project.realPath)
|
|
if (ok) {
|
|
await bot.api.sendMessage(numericChatId,
|
|
`✅ 已新建会话：${project.projectName}${project.branch ? ` (${project.branch})` : ''}`)
|
|
}
|
|
return
|
|
}
|
|
if (ambiguous) {
|
|
const list = ambiguous.map((p, i) => `${i + 1}. ${p.projectName} — ${p.realPath}`).join('\n')
|
|
await bot.api.sendMessage(numericChatId, `匹配到多个项目，请更精确：\n\n${list}`)
|
|
return
|
|
}
|
|
await bot.api.sendMessage(numericChatId, `未找到匹配 "${query}" 的项目。发送 /projects 查看完整列表。`)
|
|
} catch (err) {
|
|
await bot.api.sendMessage(numericChatId,
|
|
`❌ ${err instanceof Error ? err.message : String(err)}`)
|
|
}
|
|
} else {
|
|
const workDir = defaultWorkDir
|
|
if (workDir) {
|
|
const ok = await createSessionForChat(chatId, workDir)
|
|
if (ok) {
|
|
await bot.api.sendMessage(numericChatId, '✅ 已新建会话，可以开始对话了。')
|
|
}
|
|
} else {
|
|
await showProjectPicker(chatId)
|
|
}
|
|
}
|
|
}
|
|
|
|
const isAuthorizedTelegramUser = (userId: number) => isAllowedUser('telegram', userId)
|
|
|
|
registerAuthorizedTelegramCommand(bot, 'stop', isAuthorizedTelegramUser, (ctx) => {
|
|
const chatId = String(ctx.chat!.id)
|
|
void adapterMigrationLifecycle.track((async () => {
|
|
const result = await ensureExistingSession(chatId)
|
|
if (result.status !== 'restored') {
|
|
await ctx.reply(result.status === 'unavailable' ? SESSION_RECONNECT_NOTICE : formatImStatus(null))
|
|
return
|
|
}
|
|
bridge.sendStopGeneration(chatId)
|
|
await ctx.reply('⏹ 已发送停止信号。')
|
|
})())
|
|
})
|
|
|
|
registerAuthorizedTelegramCommand(bot, 'status', isAuthorizedTelegramUser, async (ctx) => {
|
|
const chatId = String(ctx.chat!.id)
|
|
await ctx.reply(await buildStatusText(chatId))
|
|
})
|
|
|
|
registerAuthorizedTelegramCommand(bot, 'clear', isAuthorizedTelegramUser, (ctx) => {
|
|
const chatId = String(ctx.chat!.id)
|
|
void adapterMigrationLifecycle.track((async () => {
|
|
const result = await ensureExistingSession(chatId)
|
|
if (result.status !== 'restored') {
|
|
await ctx.reply(result.status === 'unavailable' ? SESSION_RECONNECT_NOTICE : formatImStatus(null))
|
|
return
|
|
}
|
|
clearTransientChatState(chatId)
|
|
const sent = bridge.sendUserMessage(chatId, '/clear')
|
|
if (!sent) {
|
|
await ctx.reply('⚠️ 无法发送 /clear，请先发送 /new 重新连接会话。')
|
|
return
|
|
}
|
|
getRuntimeState(chatId).state = 'thinking'
|
|
await ctx.reply('🧹 已清空当前会话上下文。')
|
|
})())
|
|
})
|
|
|
|
for (const command of ['allow', 'always', 'allow-always', 'deny'] as const) {
|
|
bot.command(command, async (ctx) => {
|
|
await routeUserMessage(ctx, `/${command}${ctx.match ? ` ${ctx.match}` : ''}`, [])
|
|
})
|
|
}
|
|
|
|
/** Shared per-user-message pipeline: dedup, pairing check, project-pick
|
|
* routing, enqueue, ensureSession, sendUserMessage with attachments.
|
|
* Caller has already extracted text and attachments from the context. */
|
|
async function routeUserMessage(
|
|
ctx: Context,
|
|
text: string,
|
|
attachments: AttachmentRef[],
|
|
): Promise<void> {
|
|
if (!ctx.from || ctx.chat?.type !== 'private') return
|
|
const chatId = String(ctx.chat.id)
|
|
if (!shouldProcessTelegramMessage(dedup, chatId, ctx.message?.message_id)) return
|
|
|
|
const userId = ctx.from.id
|
|
|
|
if (!isAllowedUser('telegram', userId)) {
|
|
const displayName = [ctx.from.first_name, ctx.from.last_name].filter(Boolean).join(' ')
|
|
const success = tryPair(text.trim(), { userId, displayName }, 'telegram')
|
|
if (success) {
|
|
await ctx.reply('✅ 配对成功！现在可以开始聊天了。\n\n发送消息即可与 Claude 对话。')
|
|
} else {
|
|
await ctx.reply('🔒 未授权。请在 Claude Code 桌面端生成配对码后发送给我。')
|
|
}
|
|
return
|
|
}
|
|
|
|
// Captions remain conversation content even when an attachment download fails.
|
|
const hasAttachments = attachments.length > 0 || Boolean(
|
|
ctx.message?.photo || ctx.message?.document || ctx.message?.video || ctx.message?.audio || ctx.message?.voice,
|
|
)
|
|
await enqueue(chatId, async () => {
|
|
const permissionDecision = !hasAttachments
|
|
? parsePermissionCommand(text, pendingPermissions.get(chatId))
|
|
: null
|
|
if (permissionDecision) {
|
|
await handlePermissionDecision(chatId, permissionDecision)
|
|
return
|
|
}
|
|
|
|
if (await tryHandleTelegramSessionInput(chatId, text, hasAttachments, {
|
|
startNewSession,
|
|
showProjectPicker,
|
|
showResumeProjectPicker: commandController.showResumeProjectPicker,
|
|
handleSessionInput: (id, input) => sessionSelection.handleInput(id, input),
|
|
})) return
|
|
|
|
if (!hasAttachments && pendingProjectSelection.has(chatId)) {
|
|
if (text.trim()) await startNewSession(chatId, text.trim())
|
|
return
|
|
}
|
|
const ready = await ensureSession(chatId)
|
|
if (!ready) return
|
|
const effective =
|
|
text || (attachments.length > 0 ? '(用户发送了附件)' : '')
|
|
if (!effective && attachments.length === 0) return
|
|
const sent = bridge.sendUserMessage(chatId, effective, attachments.length ? attachments : undefined)
|
|
if (!sent) {
|
|
await bot.api.sendMessage(Number(chatId), '⚠️ 消息发送失败，连接可能已断开。请发送 /new 重新开始。')
|
|
} else {
|
|
getRuntimeState(chatId).state = 'thinking'
|
|
}
|
|
})
|
|
}
|
|
|
|
/** Scan ctx.message for photo/document/video/audio/voice, download
|
|
* each via TelegramMediaService, apply size/mime limits, and produce
|
|
* a ready-to-send AttachmentRef[] plus any rejection hints. */
|
|
async function collectAttachmentsFromCtx(
|
|
ctx: Context,
|
|
): Promise<{ attachments: AttachmentRef[]; rejections: string[] }> {
|
|
const msg = ctx.message
|
|
if (!msg || !ctx.chat) return { attachments: [], rejections: [] }
|
|
const sessionId = sessionStore.get(String(ctx.chat.id))?.sessionId ?? String(ctx.chat.id)
|
|
const attachments: AttachmentRef[] = []
|
|
const rejections: string[] = []
|
|
|
|
const runOne = async (
|
|
fileId: string,
|
|
fileName?: string,
|
|
mimeType?: string,
|
|
): Promise<void> => {
|
|
try {
|
|
const local = await media.downloadFile(fileId, sessionId, { fileName, mimeType })
|
|
const check = checkAttachmentLimit(local.kind, local.size, local.mimeType)
|
|
if (!check.ok) {
|
|
rejections.push(check.hint)
|
|
return
|
|
}
|
|
if (local.kind === 'image') {
|
|
attachments.push({
|
|
type: 'image',
|
|
name: local.name,
|
|
data: local.buffer.toString('base64'),
|
|
mimeType: local.mimeType,
|
|
})
|
|
} else {
|
|
attachments.push({
|
|
type: 'file',
|
|
name: local.name,
|
|
path: local.path,
|
|
mimeType: local.mimeType,
|
|
})
|
|
}
|
|
} catch (err) {
|
|
console.error('[Telegram] downloadFile failed:', err)
|
|
rejections.push('📎 附件下载失败,请稍后重试')
|
|
}
|
|
}
|
|
|
|
// Photos: grammY exposes an array of sizes, largest last.
|
|
if (msg.photo && msg.photo.length > 0) {
|
|
const largest = msg.photo[msg.photo.length - 1]!
|
|
await runOne(largest.file_id, `photo-${largest.file_unique_id}.jpg`, 'image/jpeg')
|
|
}
|
|
if (msg.document) {
|
|
await runOne(msg.document.file_id, msg.document.file_name, msg.document.mime_type)
|
|
}
|
|
if (msg.video) {
|
|
await runOne(msg.video.file_id, msg.video.file_name, msg.video.mime_type)
|
|
}
|
|
if (msg.audio) {
|
|
await runOne(msg.audio.file_id, msg.audio.file_name, msg.audio.mime_type)
|
|
}
|
|
if (msg.voice) {
|
|
await runOne(
|
|
msg.voice.file_id,
|
|
`voice-${msg.voice.file_unique_id}.ogg`,
|
|
msg.voice.mime_type ?? 'audio/ogg',
|
|
)
|
|
}
|
|
|
|
return { attachments, rejections }
|
|
}
|
|
|
|
bot.on('message:text', async (ctx) => {
|
|
await routeUserMessage(ctx, ctx.message.text, [])
|
|
})
|
|
|
|
bot.on(
|
|
['message:photo', 'message:document', 'message:video', 'message:audio', 'message:voice'],
|
|
async (ctx) => {
|
|
const caption = ctx.message.caption ?? ''
|
|
const { attachments, rejections } = await collectAttachmentsFromCtx(ctx)
|
|
for (const r of rejections) {
|
|
await ctx.reply(r).catch(() => {})
|
|
}
|
|
if (attachments.length === 0 && !caption.trim()) return
|
|
await routeUserMessage(ctx, caption, attachments)
|
|
},
|
|
)
|
|
|
|
bot.on('callback_query:data', async (ctx) => {
|
|
if (!ctx.from || ctx.chat?.type !== 'private') return
|
|
if (!dedup.tryRecord(`telegram:callback:${ctx.callbackQuery.id}`)) return
|
|
const data = ctx.callbackQuery.data
|
|
await enqueue(String(ctx.chat.id), async () => {
|
|
if (await tryHandleTelegramSelectionCallback(data, ctx, commandController)) return
|
|
|
|
if (!data.startsWith('permit:')) return
|
|
|
|
const decision = parsePermitCallbackData(data)
|
|
if (!decision) return
|
|
await commandController.handlePermissionCallback(ctx, decision, pendingPermissions, (chatId) => getRuntimeState(chatId).pendingPermissionCount = Math.max(0, getRuntimeState(chatId).pendingPermissionCount - 1))
|
|
})
|
|
})
|
|
|
|
// ---------- start ----------
|
|
|
|
export function stopTelegramAdapter(): void {
|
|
if (bot.isRunning()) void bot.stop()
|
|
bridge.destroy()
|
|
dedup.destroy()
|
|
}
|
|
|
|
export function startTelegramAdapter(): void {
|
|
console.log('[Telegram] Starting bot...')
|
|
console.log(`[Telegram] Server: ${config.serverUrl}`)
|
|
console.log(`[Telegram] Allowed users: ${config.telegram.allowedUsers.length === 0 ? 'paired users only' : config.telegram.allowedUsers.join(', ')}`)
|
|
void attachmentStore.gc().catch((err) => {
|
|
console.warn('[Telegram] AttachmentStore.gc failed:', err instanceof Error ? err.message : err)
|
|
})
|
|
void syncTelegramBotCommands(bot.api).then(() => console.log('[Telegram] Command menu synced')).catch((err) => console.warn('[Telegram] Command menu sync failed:', err instanceof Error ? err.message : err))
|
|
void bot.start({ onStart: () => console.log('[Telegram] Bot is running!') })
|
|
registerAdapterShutdown(async () => {
|
|
console.log('[Telegram] Shutting down...')
|
|
if (bot.isRunning()) await bot.stop()
|
|
bridge.destroy()
|
|
dedup.destroy()
|
|
})
|
|
}
|
|
|
|
// Desktop's shared sidecar imports this module with its explicit adapter flag.
|
|
if (import.meta.main || process.argv.includes('--telegram')) startTelegramAdapter()
|