Repository navigation
fix(runtime): make Telegram bot connections operational #3105
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -9,7 +9,7 @@ import { | |
| nonTextMessageAck, | ||
| plaintextHelpReply, | ||
| } from '@maka/core/bot-events'; | ||
| import { generalizedErrorMessage } from '@maka/core/redaction'; | ||
| import { generalizedErrorMessage, redactSecrets } from '@maka/core/redaction'; | ||
| import type { BotIncomingMessage, BotRegistry, BotReplyStream } from '@maka/runtime/bots'; | ||
| import type { BotSessionAdapter, BotSessionTurnResult } from './bot-session-adapter.js'; | ||
| import { isBotSessionUnavailableError } from './bot-session-adapter.js'; | ||
|
|
@@ -205,6 +205,7 @@ export function createBotIncomingMainService(deps: BotIncomingMainServiceDeps): | |
| ): Promise<void> { | ||
| if (closed) return; | ||
| let replyStream: BotReplyStream | null = null; | ||
| let processingStage = 'command-routing'; | ||
| // PR-BOT-EPHEMERAL-REPLY-0: TTL for system notices (help / reset ack / | ||
| // fallback errors). Five minutes is long enough for the user to read | ||
| // and process the notice on mobile; short enough that bot DMs do not | ||
|
|
@@ -248,13 +249,15 @@ export function createBotIncomingMainService(deps: BotIncomingMainServiceDeps): | |
| let sessionId = botConversationSessions.get(conversationKey); | ||
| try { | ||
| if (!sessionId) { | ||
| processingStage = 'session-create'; | ||
| sessionId = await createBotConversationSession( | ||
| conversationKey, | ||
| message, | ||
| SYSTEM_NOTICE_TTL_MS, | ||
| ); | ||
| if (!sessionId) return; | ||
| } else { | ||
| processingStage = 'session-prepare'; | ||
| let rebound = false; | ||
| try { | ||
| const permissionModeOk = await ensureBotSessionExploreMode( | ||
|
|
@@ -285,6 +288,7 @@ export function createBotIncomingMainService(deps: BotIncomingMainServiceDeps): | |
| } | ||
|
|
||
| const turnId = randomUUID(); | ||
| processingStage = 'turn-start'; | ||
| const replyOptions = message.sourceMessageId | ||
| ? { replyToMessageId: message.sourceMessageId } | ||
| : undefined; | ||
|
|
@@ -344,6 +348,7 @@ export function createBotIncomingMainService(deps: BotIncomingMainServiceDeps): | |
| // bridge layer drops the field for non-Telegram platforms / multi-chunk | ||
| // continuation pieces. | ||
| if (reply.trim()) { | ||
| processingStage = 'reply-send'; | ||
| // Actual agent reply: NO ephemeral TTL. The answer must stay | ||
| // visible — auto-deleting it would defeat the bot's purpose. | ||
| const sent = replyStream | ||
|
|
@@ -370,6 +375,12 @@ export function createBotIncomingMainService(deps: BotIncomingMainServiceDeps): | |
| } catch (error) { | ||
| await replyStream?.abort().catch(() => {}); | ||
| if (closed) return; | ||
| const errorName = error instanceof Error ? error.name : typeof error; | ||
| const errorCode = readErrorCode(error); | ||
| const errorDetail = redactSecrets(error instanceof Error ? error.message : String(error)); | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [P2] This new log line reaches for const errorDetail = redactSecrets(error instanceof Error ? error.message : String(error));
console.error(`[bot-incoming] conversation processing failed: stage=${processingStage} … detail=${errorDetail}`);
Measured against The patterns cover What was not demonstrated, stated plainly: undici transport failures were exercised directly — Why it is still worth fixing:
Either add a Telegram-shaped pattern to Independent untrusted-input & credential line, bound to |
||
| console.error( | ||
| `[bot-incoming] conversation processing failed: stage=${processingStage} name=${errorName} code=${errorCode ?? 'unknown'} detail=${errorDetail}`, | ||
| ); | ||
| const detail = isSessionWorkspaceUnavailableError(error) | ||
| ? '工作目录不可用,请在桌面端选择有效目录后重试' | ||
| : generalizedErrorMessage(error, '机器人对话处理失败'); | ||
|
|
@@ -412,3 +423,9 @@ function botReply(result: BotSessionTurnResult): string { | |
| if (result.kind === 'errored') return `Maka 处理失败:${result.reason}`; | ||
| return result.text; | ||
| } | ||
|
|
||
| function readErrorCode(error: unknown): string | undefined { | ||
| if (!error || typeof error !== 'object' || !('code' in error)) return undefined; | ||
| const code = (error as { code?: unknown }).code; | ||
| return typeof code === 'string' || typeof code === 'number' ? String(code) : undefined; | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -87,16 +87,25 @@ export function registerSettingsBotsIpc( | |
| }); | ||
| return toSettingsTestResult(provider, result); | ||
| }); | ||
| deps.ipcMain.handle('settings:bots:listStatuses', () => | ||
| tryResult(async () => deps.botRegistry.allStatuses(), 'BOTS_STATUS_FAILED'), | ||
| ); | ||
| deps.ipcMain.handle('settings:bots:listStatuses', () => deps.botRegistry.allStatuses()); | ||
| deps.ipcMain.handle('settings:bots:restart', (_event, provider: BotProvider) => | ||
| tryResult(async () => { | ||
| await deps.botRegistry.applySettings( | ||
| (await deps.settingsStore.get()).botChat, | ||
| (async () => { | ||
| const settings = await deps.settingsStore.get(); | ||
| const status = await deps.botRegistry.restart( | ||
| provider, | ||
| settings.botChat.channels[provider], | ||
| ); | ||
| return deps.botRegistry.getStatus(provider); | ||
| }, 'BOTS_RESTART_FAILED'), | ||
| if (!status.running) { | ||
| const reason = status.reason ?? 'unknown'; | ||
| console.error( | ||
| `[BotRegistry] ${provider} restart did not enter listening: ${reason}`, | ||
|
Comment on lines
+98
to
+101
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 2. Restart discards failure status Disposition: fix-now. The restart IPC handler rejects normal non-running BotStatus results, so the renderer’s declared Promise<BotStatus> contract and its explicit status.running === false UI path are bypassed; it retains stale status and shows only the generic start-failed toast. Agent Prompt
|
||
| ); | ||
| // The renderer intentionally hides unmapped raw status reasons. Reject | ||
| // here so diagnostics retain the actionable, already-generalized cause. | ||
| throw new Error(`${provider} listener startup failed: ${reason}`); | ||
| } | ||
| return status; | ||
| })(), | ||
| ); | ||
| deps.ipcMain.handle( | ||
| 'settings:bots:onboarding:start', | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -76,6 +76,7 @@ export function botSettingsRequireRestart( | |
| previous.appId !== next.appId || | ||
| previous.appSecret !== next.appSecret || | ||
| previous.domain !== next.domain || | ||
| previous.webhookUrl !== next.webhookUrl | ||
| previous.webhookUrl !== next.webhookUrl || | ||
| previous.proxyUrl !== next.proxyUrl | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [P2 · category ①] proxyUrl is now a restart trigger for every BotProvider, but this PR only passes the channel override into Telegram requests. Discord REST, gateway lookup, typing, and WebSocket connection still use proxiedFetch/WebSocket without settings.proxyUrl, while the shared Settings detail still exposes that field for Discord. Changing a Discord proxy therefore tears down and reconnects the gateway—creating a recoverable message window—yet leaves routing unchanged, so the saved setting is ineffective. Either scope this restart predicate/field to Telegram, or plumb the override through the complete Discord REST and gateway path with coverage. |
||
| ); | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -48,6 +48,27 @@ export class BotRegistry extends EventEmitter { | |
| return next; | ||
| } | ||
|
|
||
| /** Force a provider restart even when its persisted settings are unchanged. */ | ||
| async restart(platform: BotPlatform, settings: BotChannelSettings): Promise<BotStatus> { | ||
| const restart = async (): Promise<BotStatus> => { | ||
| const existing = this.bridges.get(platform); | ||
| if (existing) { | ||
| await existing.stop().catch(() => {}); | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. P1 — Cancel or invalidate in-flight startup before replacing the bridge. |
||
| this.detach(existing); | ||
| this.bridges.delete(platform); | ||
| } | ||
| this.statuses.delete(platform); | ||
| await this.reconcileOne(platform, settings); | ||
| return this.getStatus(platform); | ||
| }; | ||
| const next = this.applyQueue.then(restart, restart); | ||
| this.applyQueue = next.then( | ||
| () => undefined, | ||
| () => undefined, | ||
| ); | ||
| return next; | ||
| } | ||
|
|
||
| getStatus(platform: BotPlatform): BotStatus { | ||
| return ( | ||
| this.bridges.get(platform)?.getStatus() ?? | ||
|
|
@@ -134,18 +155,7 @@ export class BotRegistry extends EventEmitter { | |
| ).updateSettings; | ||
| if (update && !update.call(existing, settings).needsRestart) return; | ||
| await existing.stop().catch(() => {}); | ||
| // Drop our 'message' / 'statusChange' listeners on the old | ||
| // bridge before dereferencing it. Some bridges (Discord | ||
| // gateway, WeChat poll) hold async tasks that can still emit | ||
| // after `stop()` returns; without removeAllListeners those | ||
| // emissions would race-call onIncomingMessage / onStatusChange | ||
| // against a bridge the registry already considers gone. | ||
| try { | ||
| (existing as BotBridge & EventEmitter).removeAllListeners('message'); | ||
| (existing as BotBridge & EventEmitter).removeAllListeners('statusChange'); | ||
| } catch { | ||
| // best-effort — non-EventEmitter bridges don't have listeners to clear. | ||
| } | ||
| this.detach(existing); | ||
|
coderabbitai[bot] marked this conversation as resolved.
Resolved
|
||
| } | ||
| this.statuses.delete(platform); | ||
|
|
||
|
|
@@ -174,6 +184,18 @@ export class BotRegistry extends EventEmitter { | |
| ); | ||
| } | ||
|
|
||
| private detach(bridge: BotBridge): void { | ||
| // Some bridges retain async tasks after stop(); detach registry listeners | ||
| // before dereferencing them so stale emissions cannot race the replacement. | ||
| try { | ||
| const emitter = bridge as BotBridge & EventEmitter; | ||
| emitter.removeAllListeners('message'); | ||
| emitter.removeAllListeners('statusChange'); | ||
| } catch { | ||
| // Non-EventEmitter bridges have no listeners to clear. | ||
| } | ||
| } | ||
|
|
||
| private wire(bridge: BotBridge): void { | ||
| const emitter = bridge as BotBridge & EventEmitter; | ||
| emitter.on('message', (message: BotIncomingMessage) => this.deps.onIncomingMessage(message)); | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,3 +1,4 @@ | ||
| import type { ProxySettings } from '@maka/core/settings/network-settings'; | ||
| import { fetch, type Dispatcher, type RequestInit as UndiciRequestInit } from 'undici'; | ||
| import { matchesBypassList } from '../network/bypass-matcher.js'; | ||
| import { buildProxyDispatcher } from '../network/proxy-dispatcher.js'; | ||
|
|
@@ -12,6 +13,8 @@ export type ProxiedFetchInit = Omit< | |
| > & { | ||
| signal?: AbortSignal | null; | ||
| timeoutMs?: number; | ||
| /** Per-channel proxy override (for example Telegram's saved proxy URL). */ | ||
| proxyUrl?: string; | ||
| }; | ||
|
|
||
| export async function proxiedFetch( | ||
|
|
@@ -20,12 +23,12 @@ export async function proxiedFetch( | |
| ): Promise<Response> { | ||
| const url = | ||
| typeof input === 'string' ? input : input instanceof URL ? input.toString() : input.url; | ||
| const proxy = resolveActiveProxy(); | ||
| const { timeoutMs = DEFAULT_TIMEOUT_MS, signal, proxyUrl, ...fetchInit } = init; | ||
| const proxy = proxyUrl?.trim() ? proxySettingsFromUrl(proxyUrl) : resolveActiveProxy(); | ||
|
Comment on lines
+26
to
+27
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 1. Default proxy hijacks telegram proxiedFetch now treats every nonblank channel proxyUrl as an explicit override, while each default Telegram channel is pre-populated with http://127.0.0.1:7890, causing fresh and previously defaulted installations without that local proxy to bypass direct/global networking and fail even with valid credentials. This affects connection tests, startup, polling, sends, typing, deletion, and streaming drafts, contradicting the intended behavior of applying a per-channel override only when configured by the user. Agent Prompt
|
||
| let dispatcher: Dispatcher | undefined; | ||
| if (proxy && !matchesBypassList(new URL(url).hostname, proxy.bypassList)) { | ||
| dispatcher = buildProxyDispatcher(proxy) as Dispatcher; | ||
| } | ||
| const { timeoutMs = DEFAULT_TIMEOUT_MS, signal, ...fetchInit } = init; | ||
| const timeoutEnabled = timeoutMs > 0; | ||
| const controller = new AbortController(); | ||
| const requestSignal = signal ? AbortSignal.any([signal, controller.signal]) : controller.signal; | ||
|
|
@@ -91,6 +94,32 @@ export async function proxiedFetch( | |
| return response; | ||
| } | ||
|
|
||
| function proxySettingsFromUrl(value: string): ProxySettings { | ||
| const url = new URL(value); | ||
| const type = | ||
| url.protocol === 'socks5:' | ||
| ? 'socks5' | ||
| : url.protocol === 'https:' | ||
| ? 'https' | ||
| : url.protocol === 'http:' | ||
| ? 'http' | ||
| : undefined; | ||
| if (!type) throw new Error(`Unsupported proxy protocol: ${url.protocol}`); | ||
| const port = Number(url.port || (type === 'https' ? 443 : type === 'socks5' ? 1080 : 80)); | ||
| if (!url.hostname || !Number.isInteger(port) || port < 1 || port > 65_535) { | ||
| throw new Error('Invalid proxy URL'); | ||
| } | ||
| return { | ||
| enabled: true, | ||
| type, | ||
| host: url.hostname, | ||
| port, | ||
| ...(url.username ? { username: decodeURIComponent(url.username) } : {}), | ||
| ...(url.password ? { password: decodeURIComponent(url.password) } : {}), | ||
| bypassList: [], | ||
| }; | ||
| } | ||
|
|
||
| Object.defineProperty(proxiedFetch, FETCH_PROXY_SNAPSHOT, { | ||
| get: resolveActiveProxy, | ||
| enumerable: false, | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.