diff --git a/CHANGELOG.md b/CHANGELOG.md index ca226e8..7097300 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,15 @@ ## Unreleased +- `Status Errors`: Failures of the poll loop and of message delivery are classified into a short state with the next step (`invalid token · /telegram-setup`, `rate limited 12s · retrying`, `another poller is active · /telegram-status`, `Telegram unavailable`, `offline · retrying`); unrecognized messages stay verbatim, sanitized and trimmed to 100 characters. A non-idempotent send that hit a transport failure after the request left is reported as `delivery unconfirmed` instead of a definite failure. Impact: the status says what to do instead of showing raw runtime text. +- `Status Persistence`: The poll loop cleared the error status after its retry sleep, before the next attempt, so a slow retry (a 12 s rate-limit wait) showed as `connected`. The reset now happens after the next successful `getUpdates`, and recovery is reported only when `getUpdates` itself had failed, not when an update handler did. Impact: an error stays visible until polling works again. +- `Delivery Counters`: `lib/traffic.ts` counts undelivered messages (`recordFailed`, `onCallFailed`) and the status shows `✘N` next to the sent counter; `/telegram-status` adds `N failed`. `TelegramBridgeApiRuntime` gains `onCallFailed` beside `onCallSucceeded`. +- `Operator Notices`: New leaf-adjacent domain `lib/notices.ts` raises host toasts for owner paired (`onPaired` on the pairing runtime), connection recovered (`onRecovered` on the poll loop) and reply not delivered (throttled to one per 30 s). A failing toast is recorded as a runtime event and never reaches the send, pairing or polling path. +- `Queue Preview`: With turns waiting, the TUI widget adds a second line with the next two summaries in dispatch order (`Queue.formatQueuedTelegramItemsPreview`) and a remainder count; the plain `setStatus` entry keeps one line. +- `Working Message`: While a Telegram-originated turn runs, the omp spinner reads `Answering Telegram · ""` and is restored when the turn settles. `setExtensionWorkingMessage` skips hosts without `ui.setWorkingMessage`, because the text is cosmetic. +- `Widget Sanitization`: Widget lines bypass the host's status sanitizer, so `sanitizeTelegramStatusText` strips control characters and bidi overrides from every Telegram-controlled string (thread name, bot name, queued prompts, error text) before it reaches the terminal. +- `Status Line`: The line under the omp editor now shows a state glyph, the bot name, received/sent message counters and the time of the last message (`telegram ● connected · @bot · ⤵12 ⤴11 · ⏱ 14:32`), plus the next step for non-working states (`/telegram-setup`, `send /start to @bot`, `/telegram-connect`). Glyphs follow omp's `symbolPreset`. omp strips ANSI from extension status text, so in the TUI the line is rendered as a `belowEditor` widget that keeps the theme colors; other modes keep `setStatus`. Counters come from the new leaf domain `lib/traffic.ts`: received on every owner message, sent on every successful `sendMessage`/`sendRichMessage`/media call, and `/telegram-status` gains a `- messages:` line. Impact: connection state, traffic and recovery steps are visible without opening `/telegram-status`. +- `Pairing`: The poll loop persisted the config object it captured when polling started. The store swaps in a fresh object on every persist, so after the initial sync that stale copy lacked later changes and the next offset commit erased the paired owner: `telegram.json` never kept `allowedUserId`, the bridge re-paired on every message and the status flipped back to `awaiting pairing`. The runner now carries only the poll offset onto the store's current config. Impact: first-time pairing sticks, and another Telegram user can no longer be paired by writing before the owner's next message. - `omp Agent Directory`: Resolve the agent directory from the host's own `getAgentDir()` (registered by the composition root via `setHostAgentDirResolver`) instead of sniffing `execPath`/`argv[1]` for an `omp` prefix. omp installed through Bun runs as `bun …/@oh-my-pi/pi-coding-agent/dist/cli.js`, which matched neither, so with `telegram.json` in `~/.omp/agent` the bridge never connected and wrote its log and bus socket under `~/.pi/agent`. `PI_CODING_AGENT_DIR` still wins; with no host registered the fallback stays `~/.pi/agent`. Impact: a bot token saved by `/telegram-setup` lives in omp's own agent directory and is found again. `lib/paths.ts` stays free of Pi SDK imports, which cost about a second of startup per importing process. - `omp Settled Boundary`: omp never emits `agent_settled`, the only place the activity runtime finalized a turn. On omp the first terminal `agent_end` (no `willContinue: true`, which omp sets for retries, compaction continuations, todo/plan reminders and async wakes) now drives `onAgentSettled`; hosts that emit `agent_settled` natively are unchanged. Impact: Activity API consumers receive `agent-settled` and the activity id no longer leaks across turns. - `omp Current Model`: omp never emits `model_select`, so the stored current model went stale after a local `/model` switch, and omp builds command contexts by spreading a base context, which freezes `ctx.model` at the moment of `/telegram-connect`. On omp the live `ctx.models.current()` now wins over both, and the stored value no longer overrides it. Impact: the Telegram model menu marks the model omp is actually using. diff --git a/README.md b/README.md index 4cdc1bb..aac9413 100644 --- a/README.md +++ b/README.md @@ -130,6 +130,36 @@ Config is written to `/telegram.json`. The agent directory is `PI_COD Profile names: lowercase ASCII letters and digits, up to 32 characters; `default`, `main` and `active` are reserved. Profiles keep isolated polling, diagnostics, Threaded Mode state and local bus transport. +### Status line + +The bridge keeps one live line under the `omp` editor. In the interactive TUI it is a themed widget (`omp` strips color from plain extension status text, so the widget is what keeps the theme colors); in RPC, JSON and print modes it falls back to a plain status entry. The glyphs follow `omp`'s `symbolPreset` (`unicode`, `nerd` or `ascii`). + +```text +telegram ● connected · @my_bot · ⤵12 ⤴11 · ⏱ 14:32 +telegram ⟳ active +3 · @my_bot · ⤵14 ⤴11 · ⏱ 14:33 + next: "second prompt" · "third prompt" · +1 more +telegram ● connected · @my_bot · ⤵12 ⤴10 ✘2 · ⏱ 14:32 +telegram ⚠ awaiting pairing · send /start to @my_bot +telegram ○ disconnected · /telegram-connect +telegram ⦸ not configured · /telegram-setup +telegram ✘ invalid token · /telegram-setup +telegram ⟳ rate limited 12s · retrying +telegram ⟳ offline · retrying +telegram ⚠ another poller is active · /telegram-status +telegram ✘ error +``` + +- `⤵N` counts messages received from the paired owner and `⤴N` counts messages the bridge delivered to Telegram (replies, menus and attachments; typing indicators, drafts and edits are not counted). `✘N` appears only when some deliveries failed or could not be confirmed. `⏱` is the time of the latest message. Counters live in memory, start at zero when the extension loads, and are shown only while the bridge is up. +- The state word is the same ladder as before: `not configured`, `awaiting pairing`, `electing`, `disconnected`, then `active` while a turn runs or is queued, and `connected`, `leader` or `follower` otherwise. +- While turns wait, a second line previews the next two in dispatch order and how many more follow. +- Transport and API failures are classified into a short state with the next step: HTTP 401 (`invalid token`), 429 (`rate limited`), 409 (`another poller is active`), 5xx (`Telegram unavailable`) and network errors (`offline`). Anything unrecognized stays visible verbatim, trimmed to 100 characters. +- An error stays on screen until a poll succeeds again, so a slow retry is never shown as healthy. The API client waits out a rate limit itself, so `rate limited` appears once those waits are exhausted, not at the first 429. +- Toasts: `Telegram paired with user `, `Telegram reconnected`, and `Telegram: reply not delivered ()`. When a send hit a transport failure after the request left, the outcome is unknown and the toast says `reply delivery unconfirmed (check the chat)`. Delivery failure toasts are throttled to one per 30 seconds. +- While a turn that arrived from Telegram runs, the `omp` spinner reads `Answering Telegram · ""`. +- Widget lines bypass the host's text sanitizer, so every Telegram-controlled string shown there (thread name, bot name, queued prompts, error text) is stripped of control characters and bidi overrides. +- Messages a follower instance sends through the leader are counted by the leader process, not by the follower. +- `/telegram-status` repeats the counters as a `- messages:` line. + ### Tools available to the agent | Tool | Purpose | diff --git a/index.ts b/index.ts index 2fa90b1..80bbbcd 100644 --- a/index.ts +++ b/index.ts @@ -24,6 +24,7 @@ import * as MenuQueue from "./lib/menu-queue.ts"; import * as MenuSettings from "./lib/menu-settings.ts"; import * as Menu from "./lib/menu.ts"; import * as Model from "./lib/model.ts"; +import * as Notices from "./lib/notices.ts"; import * as Outbound from "./lib/outbound.ts"; import * as Ownership from "./lib/ownership.ts"; import * as Paths from "./lib/paths.ts"; @@ -43,6 +44,7 @@ import * as TelegramApi from "./lib/telegram-api.ts"; import * as TextGroups from "./lib/text-groups.ts"; import * as ThreadReconciler from "./lib/thread-reconciler.ts"; import * as TimeInjection from "./lib/time-injection.ts"; +import * as Traffic from "./lib/traffic.ts"; import * as Updates from "./lib/updates.ts"; import * as Voice from "./lib/voice.ts"; @@ -274,11 +276,13 @@ export default function (pi: Pi.ExtensionAPI) { getFollowerThreadName: telegramBusFollowerRegistrationState.getThreadName, getCurrentIdentity: getCurrentInstanceThreadIdentity, }); + const trafficCounters = Traffic.createTelegramTrafficCounters(); const statusRuntime = Status.createTelegramBridgeStatusRuntime< Pi.ExtensionContext, Queue.TelegramQueueItem >({ getConfig: configStore.get, + getTraffic: trafficCounters.snapshot, getActiveProfileName: configStore.getActiveProfileName, getDiagnosticPaths: Paths.getTelegramDiagnosticsDisplayPaths, isPollingActive: Polling.createTelegramPollingActivityReader( @@ -292,6 +296,7 @@ export default function (pi: Pi.ExtensionAPI) { hasPendingModelSwitch: pendingModelSwitchStore.has, getQueuedItems: telegramQueueStore.getQueuedItems, formatQueuedStatus: Queue.formatQueuedTelegramItemsStatus, + getQueuedPreview: Queue.formatQueuedTelegramItemsPreview, getRecentRuntimeEvents: runtimeEvents.getEvents, getRuntimeLockState: lockRuntime.getStatusLabel, ...threadStatusProjectionRuntime, @@ -323,6 +328,18 @@ export default function (pi: Pi.ExtensionAPI) { getCwd: Pi.getExtensionContextCwd, recordRuntimeEvent, }); + const notices = Notices.createTelegramNoticeRuntime({ + getContext: telegramSessionContextStore.get, + notify: Pi.notifyExtensionContext, + recordRuntimeEvent, + }); + const trafficBridge = Traffic.createTelegramTrafficStatusBridge({ + counters: trafficCounters, + getContext: telegramSessionContextStore.get, + updateStatus, + recordRuntimeEvent, + onDeliveryFailed: notices.deliveryFailed, + }); // --- Telegram API --- @@ -330,6 +347,8 @@ export default function (pi: Pi.ExtensionAPI) { TelegramApi.createDefaultTelegramBridgeApiRuntime({ getBotToken: configStore.getBotToken, recordRuntimeEvent, + onCallSucceeded: trafficBridge.onCallSucceeded, + onCallFailed: trafficBridge.onCallFailed, }); const telegramBusFollowerClients = BusFollower.createTelegramBusFollowerClientRuntime< @@ -645,6 +664,8 @@ export default function (pi: Pi.ExtensionAPI) { const inboundRouteRuntime = Routing.createTelegramInboundRouteRuntime({ configStore, callApi: callTelegramApi, + onAuthorizedMessage: trafficBridge.onAuthorizedMessage, + onPaired: notices.paired, getCurrentInstanceId() { return telegramInstanceId; }, @@ -822,6 +843,7 @@ export default function (pi: Pi.ExtensionAPI) { stopTypingLoop: typing.stop, updateStatus, recordRuntimeEvent, + onRecovered: notices.recovered, }); const recoverStaleTelegramTopicApiError = function ( apiBody: unknown, diff --git a/lib/bindings.ts b/lib/bindings.ts index 5dd0194..b1de408 100644 --- a/lib/bindings.ts +++ b/lib/bindings.ts @@ -329,6 +329,8 @@ export function registerTelegramLifecycleRuntimeHooks({ updateStatus, recordRuntimeEvent, }: TelegramLifecycleBindingDeps): void { + // Whether this bridge set the host spinner text, so it only clears its own. + let telegramWorkingMessageShown = false; const agentEndResetter = Runtime.createTelegramAgentEndResetter({ abort, typing, @@ -589,6 +591,14 @@ export function registerTelegramLifecycleRuntimeHooks({ await agentStartWithDedupReset(event, ctx); activityRuntime.onAgentStart(activeTurnRuntime.get()?.target); startAgentActivityTypingLoop(ctx); + const telegramTurn = activeTurnRuntime.get(); + if (telegramTurn) { + Pi.setExtensionWorkingMessage( + ctx, + Status.formatTelegramWorkingMessage(telegramTurn.statusSummary), + ); + telegramWorkingMessageShown = true; + } }, async onToolExecutionStart(event, ctx) { if (!isSessionContextActive(ctx)) return; @@ -638,6 +648,10 @@ export function registerTelegramLifecycleRuntimeHooks({ onAgentSettled(_event, ctx) { if (!isSessionContextActive(ctx)) return; activityRuntime.onAgentSettled(); + if (telegramWorkingMessageShown) { + telegramWorkingMessageShown = false; + Pi.setExtensionWorkingMessage(ctx, undefined); + } }, settleOnTerminalAgentEnd: Pi.isOmpHost(pi), onBeforeAgentStart: Prompts.createTelegramProactiveBeforeAgentStartHook({ diff --git a/lib/config.ts b/lib/config.ts index a8756fa..15c5f9a 100644 --- a/lib/config.ts +++ b/lib/config.ts @@ -777,6 +777,8 @@ export interface TelegramUserPairingRuntimeDeps { setAllowedUserId: (userId: number) => void; persistConfig: () => Promise; updateStatus: (ctx: TContext) => void; + /** Called once the user was accepted as owner and the config persisted. */ + onPaired?: (userId: number, ctx: TContext) => void; } export interface TelegramUserPairingRuntime { @@ -827,13 +829,16 @@ export function createTelegramUserPairingRuntime( deps: TelegramUserPairingRuntimeDeps, ): TelegramUserPairingRuntime { return { - pairIfNeeded: (userId, ctx) => - pairTelegramUserIfNeeded(userId, { + pairIfNeeded: async (userId, ctx) => { + const paired = await pairTelegramUserIfNeeded(userId, { allowedUserId: deps.getAllowedUserId(), ctx, setAllowedUserId: deps.setAllowedUserId, persistConfig: deps.persistConfig, updateStatus: deps.updateStatus, - }), + }); + if (paired) deps.onPaired?.(userId, ctx); + return paired; + }, }; } diff --git a/lib/notices.ts b/lib/notices.ts new file mode 100644 index 0000000..6e0ba3e --- /dev/null +++ b/lib/notices.ts @@ -0,0 +1,104 @@ +/** + * Telegram operator notices + * Zones: telegram ui, tui + * Owns the one-off host toasts (owner paired, connection recovered, reply not delivered) and their throttling + */ + +import { + sanitizeTelegramStatusText, + summarizeTelegramStatusError, +} from "./status.ts"; + +export type TelegramNoticeLevel = "info" | "warning" | "error"; + +/** A failed delivery raises at most one notice per this window. */ +export const TELEGRAM_DELIVERY_FAILED_NOTICE_COOLDOWN_MS = 30_000; + +export interface TelegramNoticeRuntimeDeps { + /** Current session context, when the bridge has one. */ + getContext: () => TContext | undefined; + notify: (ctx: TContext, message: string, level: TelegramNoticeLevel) => void; + recordRuntimeEvent: ( + category: string, + error: unknown, + details?: Record, + ) => void; + now?: () => number; +} + +export interface TelegramNoticeRuntime { + /** The first Telegram user was accepted as the bridge owner. */ + paired: (userId: number, ctx: TContext) => void; + /** Polling works again after at least one failed attempt. */ + recovered: (ctx: TContext) => void; + /** A message could not be delivered; throttled, uses the current context. */ + deliveryFailed: (error: unknown) => void; +} + +function describeTelegramDeliveryFailure(error: unknown): { + message: string; + level: TelegramNoticeLevel; +} { + const message = error instanceof Error ? error.message : String(error); + const summary = summarizeTelegramStatusError(message); + if (summary?.deliveryUnknown) { + return { + message: "Telegram: reply delivery unconfirmed (check the chat)", + level: "warning", + }; + } + if (summary) { + return { + message: `Telegram: reply not delivered (${summary.label})`, + level: summary.kind === "fatal" ? "error" : "warning", + }; + } + const raw = sanitizeTelegramStatusText(message); + const reason = raw.length > 80 ? `${raw.slice(0, 79)}…` : raw; + return { message: `Telegram: reply not delivered (${reason})`, level: "error" }; +} + +/** + * Toasts never propagate failures into the path that raised them: a stale + * session context only produces a recorded runtime event. + */ +export function createTelegramNoticeRuntime( + deps: TelegramNoticeRuntimeDeps, +): TelegramNoticeRuntime { + const now = deps.now ?? Date.now; + let lastDeliveryNoticeAtMs: number | undefined; + const show = ( + ctx: TContext, + message: string, + level: TelegramNoticeLevel, + ): void => { + try { + deps.notify(ctx, message, level); + } catch (error) { + deps.recordRuntimeEvent("status", error, { phase: "notice" }); + } + }; + return { + paired(userId, ctx) { + show(ctx, `Telegram paired with user ${userId}`, "info"); + }, + recovered(ctx) { + show(ctx, "Telegram reconnected", "info"); + }, + deliveryFailed(error) { + const ctx = deps.getContext(); + if (ctx === undefined) return; + const nowMs = now(); + if ( + lastDeliveryNoticeAtMs !== undefined && + nowMs - lastDeliveryNoticeAtMs < + TELEGRAM_DELIVERY_FAILED_NOTICE_COOLDOWN_MS + ) { + return; + } + lastDeliveryNoticeAtMs = nowMs; + const failure = describeTelegramDeliveryFailure(error); + show(ctx, failure.message, failure.level); + }, + }; +} diff --git a/lib/pi.ts b/lib/pi.ts index b8fa355..01ffb0f 100644 --- a/lib/pi.ts +++ b/lib/pi.ts @@ -260,6 +260,27 @@ export function isTerminalAgentEnd(event: AgentEndEvent): boolean { return !("willContinue" in event && event.willContinue === true); } +/** Show a one-off toast in the host UI; a no-op where the host has no UI. */ +export function notifyExtensionContext( + ctx: ExtensionContext, + message: string, + level: "info" | "warning" | "error", +): void { + ctx.ui.notify(message, level); +} + +/** + * Replace the host spinner text, or restore its default with `undefined`. + * Pi-compatible hosts do not all expose `setWorkingMessage`; the spinner text + * is cosmetic, so a host without it is skipped instead of failing the turn. + */ +export function setExtensionWorkingMessage( + ctx: ExtensionContext, + message?: string, +): void { + ctx.ui.setWorkingMessage?.(message); +} + export function isExtensionContextIdle(ctx: ExtensionContext): boolean { return ctx.isIdle(); } diff --git a/lib/polling.ts b/lib/polling.ts index a1eb736..9cd3d9d 100644 --- a/lib/polling.ts +++ b/lib/polling.ts @@ -163,6 +163,7 @@ export function createTelegramPollingControllerRuntime< updateStatus: deps.updateStatus, sleep: deps.sleep, maxUpdateFailures: deps.maxUpdateFailures, + onRecovered: deps.onRecovered, recordRuntimeEvent: deps.recordRuntimeEvent, }), updateStatus: deps.updateStatus, @@ -913,6 +914,8 @@ export interface TelegramPollLoopDeps< onStatusReset: () => void; sleep: (ms: number, signal?: AbortSignal) => Promise; maxUpdateFailures?: number; + /** Called on the first successful poll after a failed attempt. */ + onRecovered?: () => void; } export interface TelegramPollLoopRunnerDeps< @@ -930,6 +933,8 @@ export interface TelegramPollLoopRunnerDeps< updateStatus: (ctx: TContext, message?: string) => void; sleep?: (ms: number, signal?: AbortSignal) => Promise; maxUpdateFailures?: number; + /** Called on the first successful poll after a failed attempt. */ + onRecovered?: (ctx: TContext) => void; } export function sleepTelegramPollingRetry( @@ -971,7 +976,15 @@ export function createTelegramPollLoopRunner< config: deps.getConfig(), deleteWebhook: deps.deleteWebhook, getUpdates: deps.getUpdates, - persistConfig: deps.persistConfig, + // The loop keeps the config object captured when polling started, but the + // store swaps in a fresh object on every persist. Persisting that stale + // copy erased later changes such as the paired user, so only the poll + // offset is carried onto the store's current config. + persistConfig: (polled) => { + const current = deps.getConfig(); + current.lastUpdateId = polled.lastUpdateId; + return deps.persistConfig(current); + }, handleUpdate: deps.handleUpdate, onErrorStatus: (message) => { updateTelegramPollingStatusSafely(deps.updateStatus, ctx, { @@ -986,6 +999,15 @@ export function createTelegramPollLoopRunner< }, sleep, maxUpdateFailures: deps.maxUpdateFailures, + onRecovered: () => { + try { + deps.onRecovered?.(ctx); + } catch (error) { + deps.recordRuntimeEvent?.("polling", error, { + phase: "recovered-notice", + }); + } + }, recordRuntimeEvent: deps.recordRuntimeEvent, }); } @@ -1033,13 +1055,34 @@ export async function runTelegramPollLoop< const admittedUpdates = new Set(); let handledUpdateFailureRethrown = false; let consecutiveGetUpdatesConflicts = 0; + // An error status is on screen; it stays until polling works again, so a + // slow retry (for example a long rate-limit wait) never shows as healthy. + let errorShown = false; + // The last getUpdates failed, so the next successful one is a recovery. + // Failures of an update handler do not count: the transport kept working. + let transportDown = false; + const showError = (message: string): void => { + errorShown = true; + deps.onErrorStatus(message); + }; while (!deps.signal.aborted) { + let inGetUpdates = false; try { + inGetUpdates = true; const updates = await deps.getUpdates( buildTelegramLongPollRequest(deps.config.lastUpdateId), deps.signal, ); + inGetUpdates = false; consecutiveGetUpdatesConflicts = 0; + if (errorShown) { + errorShown = false; + deps.onStatusReset(); + } + if (transportDown) { + transportDown = false; + deps.onRecovered?.(); + } for (const update of updates) { if (admittedUpdates.has(update.update_id)) { deps.config.lastUpdateId = update.update_id; @@ -1068,7 +1111,7 @@ export async function runTelegramPollLoop< throw error; } const message = getTelegramPollingErrorMessage(error); - deps.onErrorStatus( + showError( `skipping Telegram update ${update.update_id} after ${failureCount} failures: ${message}`, ); admittedUpdates.add(update.update_id); @@ -1097,10 +1140,10 @@ export async function runTelegramPollLoop< continue; } consecutiveGetUpdatesConflicts = 0; - deps.onErrorStatus(getTelegramPollingErrorMessage(error)); + if (inGetUpdates) transportDown = true; + showError(getTelegramPollingErrorMessage(error)); await deps.sleep(TELEGRAM_POLLING_RETRY_MS, deps.signal); if (deps.signal.aborted) return; - deps.onStatusReset(); } } } diff --git a/lib/queue.ts b/lib/queue.ts index 10a36f1..06606e0 100644 --- a/lib/queue.ts +++ b/lib/queue.ts @@ -499,6 +499,25 @@ export function formatQueuedTelegramItemsStatus( return items.length === 0 ? "" : ` +${items.length}`; } +/** + * Summaries of the waiting items in dispatch order, for the status preview. + * Display must never break the status update, so a malformed item only + * costs the ordering. + */ +export function formatQueuedTelegramItemsPreview( + items: TelegramQueueItem[], +): string[] { + let ordered = items; + try { + ordered = [...items].sort(compareTelegramQueueItems); + } catch { + // Keep the store order. + } + return ordered + .map((item) => item.statusSummary.trim()) + .filter((summary) => summary.length > 0); +} + export function truncateTelegramQueueSummary( text: string, maxWords = 5, diff --git a/lib/routing.ts b/lib/routing.ts index f106660..6cf069e 100644 --- a/lib/routing.ts +++ b/lib/routing.ts @@ -554,6 +554,10 @@ export interface TelegramInboundRouteRuntimeDeps< method: string, body: Record, ) => Promise; + /** Called for every message from the paired owner, before it is dispatched. */ + onAuthorizedMessage?: (ctx: TContext) => void; + /** Called when the first Telegram user is accepted as the bridge owner. */ + onPaired?: (userId: number, ctx: TContext) => void; getCurrentInstanceId?: () => string | undefined; getMessageOwnership?: Updates.TelegramMessageOwnershipLookup; getTargetOwnership?: Updates.TelegramTargetOwnershipLookup; @@ -2074,6 +2078,7 @@ export function createTelegramInboundRouteRuntime< setAllowedUserId: deps.configStore.setAllowedUserId, persistConfig: deps.configStore.persist, updateStatus: deps.updateStatus, + onPaired: deps.onPaired, removePendingMediaGroupMessages: deps.mediaGroupRuntime.removeMessages, removeQueuedTelegramTurnsByMessageIds: deps.queueMutationRuntime.removeByMessageIds, @@ -2086,6 +2091,7 @@ export function createTelegramInboundRouteRuntime< handleAuthorizedTelegramCallbackQuery: callbackHandler, sendTextReply: deps.sendTextReply, handleAuthorizedTelegramMessage: async (message, ctx) => { + deps.onAuthorizedMessage?.(ctx); if (typeof message.message_thread_id === "number") { await deps.handleTelegramThreadTargetObserved?.( { diff --git a/lib/status.ts b/lib/status.ts index 9c3321a..d68b49c 100644 --- a/lib/status.ts +++ b/lib/status.ts @@ -4,6 +4,15 @@ * Builds usage, cost, and context summaries for the interactive Telegram status view */ +/** Message counters the status shows; structurally satisfied by the traffic domain. */ +export interface TelegramStatusTraffic { + received: number; + sent: number; + /** Deliveries that failed after the transport retries; absent means none. */ + failed?: number; + lastAtMs?: number; +} + export type TelegramStatusQueueLane = "control" | "priority" | "default"; export interface TelegramUsageStats { @@ -191,6 +200,7 @@ export interface TelegramBridgeStatusLineState { instanceSlot?: string; instanceThreadName?: string; lockState?: string; + traffic?: TelegramStatusTraffic; pollingActive: boolean; lastUpdateId?: number; activeSourceMessageIds?: number[]; @@ -215,12 +225,28 @@ export interface TelegramStatusBarTheme { token: "accent" | "error" | "muted" | "warning" | "success", text: string, ) => string; + /** + * omp symbol presets (unicode, nerd font, ascii) so the status follows the + * user's glyph set. Hosts that do not expose them get the unicode fallbacks. + */ + status?: { + enabled?: string; + shadowed?: string; + running?: string; + warning?: string; + error?: string; + disabled?: string; + }; + sep?: { dot?: string }; + icon?: { input?: string; output?: string; time?: string }; } export interface TelegramStatusBarState { hasBotToken: boolean; pollingActive: boolean; paired: boolean; + botUsername?: string; + traffic?: TelegramStatusTraffic; busRole?: TelegramBridgeBusRole; busLifecyclePhase?: TelegramBridgeBusLifecyclePhase; instanceSlot?: string; @@ -229,13 +255,22 @@ export interface TelegramStatusBarState { processing: boolean; processingStatus?: string; queuedStatus: string; + /** Summaries of the queued turns, in dispatch order. */ + queuedPreview?: string[]; error?: string; } export interface TelegramStatusRuntimeContext { + /** Host run mode; the colored widget is only used in the interactive TUI. */ + mode?: string; ui: { theme: TelegramStatusBarTheme; - setStatus: (key: string, text: string) => void; + setStatus(key: string, text: string | undefined): void; + setWidget?( + key: string, + content: string[] | undefined, + options?: { placement?: "aboveEditor" | "belowEditor" }, + ): void; }; } @@ -272,6 +307,7 @@ export interface TelegramBridgeStatusRuntimeDeps< hasPendingModelSwitch: () => boolean; getQueuedItems: () => TQueueItem[]; formatQueuedStatus: (items: TQueueItem[]) => string; + getQueuedPreview?: (items: TQueueItem[]) => string[]; getRecentRuntimeEvents: () => TelegramRuntimeEvent[]; getRuntimeLockState?: () => string; getBusRole?: () => TelegramBridgeBusRole | undefined; @@ -298,6 +334,7 @@ export interface TelegramBridgeStatusRuntimeDeps< getInstanceSlot?: () => string | undefined; getInstanceThreadName?: () => string | undefined; getNowMs?: () => number; + getTraffic?: () => TelegramStatusTraffic; } export interface TelegramBridgeStatusLineOptions { @@ -565,13 +602,19 @@ export function createTelegramStatusRuntime< const statusKey = deps.statusKey ?? "telegram"; return { updateStatus: (ctx, error) => { - ctx.ui.setStatus( - statusKey, - buildTelegramStatusBarText( - ctx.ui.theme, - deps.getStatusBarState(ctx, error), - ), + const lines = buildTelegramStatusBarLines( + ctx.ui.theme, + deps.getStatusBarState(ctx, error), ); + if (ctx.mode === "tui" && ctx.ui.setWidget) { + // The host strips ANSI from status text, so setStatus would drop the + // theme colors. String widgets keep them; clear the plain footer entry + // so the state is not shown twice. + ctx.ui.setWidget(statusKey, lines, { placement: "belowEditor" }); + ctx.ui.setStatus(statusKey, undefined); + return; + } + ctx.ui.setStatus(statusKey, lines[0]); }, getStatusLines: (options) => buildTelegramBridgeStatusLines(deps.getBridgeStatusLineState(), options), @@ -599,6 +642,8 @@ export function createTelegramBridgeStatusRuntime< hasBotToken: !!config.botToken, pollingActive: deps.isPollingActive(), paired: !!config.allowedUserId, + botUsername: config.botUsername, + traffic: deps.getTraffic?.(), busRole: deps.getBusRole?.(), busLifecyclePhase: deps.getBusLifecyclePhase?.(), instanceSlot: deps.getInstanceSlot?.(), @@ -618,6 +663,7 @@ export function createTelegramBridgeStatusRuntime< queuedItems: queuedItems.length, }), queuedStatus: deps.formatQueuedStatus(queuedItems), + queuedPreview: deps.getQueuedPreview?.(queuedItems), error, }; }, @@ -639,6 +685,7 @@ export function createTelegramBridgeStatusRuntime< instanceSlot: deps.getInstanceSlot?.(), instanceThreadName: deps.getInstanceThreadName?.(), lockState: deps.getRuntimeLockState?.(), + traffic: deps.getTraffic?.(), pollingActive: deps.isPollingActive(), lastUpdateId: config.lastUpdateId, activeSourceMessageIds: deps.getActiveSourceMessageIds(), @@ -761,38 +808,271 @@ function getTelegramStatusBarLabel(state: TelegramStatusBarState): string { return threadName; } +interface TelegramStatusBarGlyphs { + enabled: string; + shadowed: string; + running: string; + warning: string; + error: string; + disabled: string; + dot: string; + input: string; + output: string; + time: string; +} + +function resolveTelegramStatusBarGlyphs( + theme: TelegramStatusBarTheme, +): TelegramStatusBarGlyphs { + return { + enabled: theme.status?.enabled ?? "●", + shadowed: theme.status?.shadowed ?? "○", + running: theme.status?.running ?? "⟳", + warning: theme.status?.warning ?? "⚠", + error: theme.status?.error ?? "✘", + disabled: theme.status?.disabled ?? "⦸", + dot: theme.sep?.dot ?? " · ", + input: theme.icon?.input ?? "↓", + output: theme.icon?.output ?? "↑", + time: theme.icon?.time ?? "⏱", + }; +} + +function formatTelegramStatusBarClock(timestampMs: number): string { + const date = new Date(timestampMs); + const hours = String(date.getHours()).padStart(2, "0"); + const minutes = String(date.getMinutes()).padStart(2, "0"); + return `${hours}:${minutes}`; +} + +/** Bot name, message counters and last activity, each behind a muted dot. */ +function buildTelegramStatusBarDetails( + theme: TelegramStatusBarTheme, + glyphs: TelegramStatusBarGlyphs, + state: TelegramStatusBarState, +): string { + const parts: string[] = []; + if (state.botUsername) { + parts.push(`@${sanitizeTelegramStatusText(state.botUsername)}`); + } + if (state.traffic) { + const failed = state.traffic.failed + ? ` ${glyphs.error}${state.traffic.failed}` + : ""; + parts.push( + `${glyphs.input}${state.traffic.received} ${glyphs.output}${state.traffic.sent}${failed}`, + ); + if (state.traffic.lastAtMs !== undefined) { + parts.push( + `${glyphs.time} ${formatTelegramStatusBarClock(state.traffic.lastAtMs)}`, + ); + } + } + return parts + .map((part) => theme.fg("muted", `${glyphs.dot}${part}`)) + .join(""); +} + +function buildTelegramTrafficLines( + state: Pick, +): string[] { + if (!state.traffic) return []; + const last = + state.traffic.lastAtMs === undefined + ? "" + : `, last ${formatTelegramStatusBarClock(state.traffic.lastAtMs)}`; + return [ + `- messages: ${state.traffic.received} received, ${state.traffic.sent} sent${state.traffic.failed ? `, ${state.traffic.failed} failed` : ""}${last}`, + ]; +} + +/** + * Strip control characters and bidi overrides. Widget lines bypass the host's + * status sanitizer, and thread names, bot names, queue summaries and error text + * can all originate from Telegram, so none may carry terminal escapes. + */ +export function sanitizeTelegramStatusText(text: string): string { + return text + .replace( + /[\u0000-\u001f\u007f-\u009f\u200e\u200f\u202a-\u202e\u2066-\u2069]/g, + " ", + ) + .replace(/ +/g, " ") + .trim(); +} + +export interface TelegramStatusErrorSummary { + /** Short state text such as "invalid token". */ + label: string; + /** What the operator can do next, or "retrying" while the bridge retries. */ + hint?: string; + /** fatal needs the operator, attention needs a look, retrying heals itself. */ + kind: "fatal" | "attention" | "retrying"; + /** + * The failed call may still have reached Telegram: a non-idempotent send hit + * a transport failure after the request left, so the outcome is unknown. + */ + deliveryUnknown?: boolean; +} + +/** + * Turn a raw transport or API failure into a short state and next step. + * Returns undefined for anything unrecognized so the raw text stays visible. + */ +export function summarizeTelegramStatusError( + message: string, +): TelegramStatusErrorSummary | undefined { + if (/may have committed before transport failed/i.test(message)) { + return { + kind: "attention", + label: "delivery unconfirmed", + deliveryUnknown: true, + }; + } + if (/HTTP 401\b|\bUnauthorized\b/i.test(message)) { + return { kind: "fatal", label: "invalid token", hint: "/telegram-setup" }; + } + if (/HTTP 429\b|Too Many Requests|retry[ _]after/i.test(message)) { + const seconds = /retry[ _]after[:= ]+(\d+)/i.exec(message)?.[1]; + return { + kind: "retrying", + label: seconds ? `rate limited ${seconds}s` : "rate limited", + hint: "retrying", + }; + } + if ( + /HTTP 409\b|\bConflict\b|terminated by other getUpdates/i.test(message) + ) { + return { + kind: "attention", + label: "another poller is active", + hint: "/telegram-status", + }; + } + if ( + /HTTP 5\d\d\b|Bad Gateway|Service Unavailable|Internal Server Error/i.test( + message, + ) + ) { + return { kind: "retrying", label: "Telegram unavailable", hint: "retrying" }; + } + if ( + /Unable to connect|ECONNREFUSED|ECONNRESET|ENOTFOUND|EAI_AGAIN|ETIMEDOUT|fetch failed|network|timed? out|socket hang up/i.test( + message, + ) + ) { + return { kind: "retrying", label: "offline", hint: "retrying" }; + } + return undefined; +} + +function formatTelegramStatusBarError( + theme: TelegramStatusBarTheme, + glyphs: TelegramStatusBarGlyphs, + label: string, + error: string, +): string { + const summary = summarizeTelegramStatusError(error); + if (!summary) { + const raw = sanitizeTelegramStatusText(error); + const shown = raw.length > 100 ? `${raw.slice(0, 99)}…` : raw; + return `${label} ${theme.fg("error", `${glyphs.error} error`)} ${theme.fg("muted", shown)}`; + } + const token = summary.kind === "fatal" ? "error" : "warning"; + const glyph = + summary.kind === "fatal" + ? glyphs.error + : summary.kind === "retrying" + ? glyphs.running + : glyphs.warning; + const hint = summary.hint + ? theme.fg("muted", `${glyphs.dot}${summary.hint}`) + : ""; + return `${label} ${theme.fg(token, `${glyph} ${summary.label}`)}${hint}`; +} + export function buildTelegramStatusBarText( theme: TelegramStatusBarTheme, state: TelegramStatusBarState, ): string { - const label = theme.fg("accent", getTelegramStatusBarLabel(state)); - if (state.error) { - return `${label} ${theme.fg("error", "error")} ${theme.fg("muted", state.error)}`; - } + const glyphs = resolveTelegramStatusBarGlyphs(theme); + const label = theme.fg( + "accent", + sanitizeTelegramStatusText(getTelegramStatusBarLabel(state)), + ); + const hint = (text: string) => theme.fg("muted", `${glyphs.dot}${text}`); const queued = state.queuedStatus ? theme.fg("success", state.queuedStatus) : ""; - if (!state.hasBotToken) - return `${label} ${theme.fg("muted", "not configured")}${queued}`; - if (!state.paired) - return `${label} ${theme.fg("warning", "awaiting pairing")}${queued}`; - if (state.busLifecyclePhase === "electing") - return `${label} ${theme.fg("warning", "electing")}${queued}`; - if (!state.pollingActive && state.busRole !== "follower") - return `${theme.fg("accent", "telegram")} ${theme.fg("muted", "disconnected")}${queued}`; + if (state.error) { + return formatTelegramStatusBarError(theme, glyphs, label, state.error); + } + if (!state.hasBotToken) { + return `${label} ${theme.fg("muted", `${glyphs.disabled} not configured`)}${queued}${hint("/telegram-setup")}`; + } + if (!state.paired) { + const bot = state.botUsername + ? `@${sanitizeTelegramStatusText(state.botUsername)}` + : "the bot"; + return `${label} ${theme.fg("warning", `${glyphs.warning} awaiting pairing`)}${queued}${hint(`send /start to ${bot}`)}`; + } + if (state.busLifecyclePhase === "electing") { + return `${label} ${theme.fg("warning", `${glyphs.running} electing`)}${queued}`; + } + if (!state.pollingActive && state.busRole !== "follower") { + return `${theme.fg("accent", "telegram")} ${theme.fg("muted", `${glyphs.shadowed} disconnected`)}${queued}${hint("/telegram-connect")}`; + } + const details = buildTelegramStatusBarDetails(theme, glyphs, state); if (state.processing) { const processingStatus = state.queuedStatus ? "active" : (state.processingStatus ?? "processing"); const processingToken = processingStatus === "active" ? "warning" : "accent"; - return `${label} ${theme.fg(processingToken, processingStatus)}${queued}`; + return `${label} ${theme.fg(processingToken, `${glyphs.running} ${processingStatus}`)}${queued}${details}`; } - if (state.busRole === "follower") - return `${label} ${theme.fg("success", "follower")}${queued}`; - if (state.busRole === "leader") - return `${label} ${theme.fg("success", "leader")}`; - return `${label} ${theme.fg("success", "connected")}`; + const role = state.busRole ?? "connected"; + return `${label} ${theme.fg("success", `${glyphs.enabled} ${role}`)}${queued}${details}`; +} + +/** Up to two queued turns and a remainder count, only while the bridge is up. */ +function buildTelegramStatusBarQueueLine( + theme: TelegramStatusBarTheme, + state: TelegramStatusBarState, +): string | undefined { + const operational = + state.hasBotToken && + state.paired && + !state.error && + state.busLifecyclePhase !== "electing" && + (state.pollingActive || state.busRole === "follower"); + const preview = (state.queuedPreview ?? []) + .map(sanitizeTelegramStatusText) + .filter((summary) => summary.length > 0); + if (!operational || preview.length === 0) return undefined; + const glyphs = resolveTelegramStatusBarGlyphs(theme); + const shown = preview.slice(0, 2).map((summary) => `"${summary}"`); + const more = + preview.length > 2 ? `${glyphs.dot}+${preview.length - 2} more` : ""; + return theme.fg("muted", ` next: ${shown.join(glyphs.dot)}${more}`); +} + +/** The main status line, plus a queue preview line when turns are waiting. */ +export function buildTelegramStatusBarLines( + theme: TelegramStatusBarTheme, + state: TelegramStatusBarState, +): string[] { + const queueLine = buildTelegramStatusBarQueueLine(theme, state); + return [ + buildTelegramStatusBarText(theme, state), + ...(queueLine ? [queueLine] : []), + ]; +} + +/** Host spinner text while a turn that arrived from Telegram is running. */ +export function formatTelegramWorkingMessage(summary: string): string { + const text = sanitizeTelegramStatusText(summary); + return text ? `Answering Telegram · "${text}"` : "Answering Telegram"; } function formatTelegramBridgeBotStatus( @@ -1079,6 +1359,7 @@ function buildTelegramBridgeCompactStatusLines( `- bot: ${formatTelegramBridgeBotStatus(state)}`, ...(state.activeProfileName ? [`- profile: ${state.activeProfileName}`] : []), `- user: ${state.allowedUserId ?? "not paired"}`, + ...buildTelegramTrafficLines(state), ...(state.botThreadMode ? [`- thread mode: ${state.botThreadMode}`] : []), ...(state.busRole ? [`- role: ${state.busRole}`] : []), ...(state.busLifecyclePhase @@ -1134,6 +1415,7 @@ export function buildTelegramBridgeDiagnosticStatusLines( `- bot: ${formatTelegramBridgeBotStatus(state)}`, ...(state.activeProfileName ? [`- profile: ${state.activeProfileName}`] : []), `- allowed user: ${state.allowedUserId ?? "not paired"}`, + ...buildTelegramTrafficLines(state), ...(state.botThreadMode ? [ `- thread mode: ${state.botThreadMode}${state.botThreadModeAction ? ` reconcile=${state.botThreadModeAction}` : ""}`, diff --git a/lib/telegram-api.ts b/lib/telegram-api.ts index 7f78fb9..f6738fc 100644 --- a/lib/telegram-api.ts +++ b/lib/telegram-api.ts @@ -375,6 +375,10 @@ export interface TelegramBridgeApiRuntimeDeps { error: unknown, details?: Record, ) => void; + /** Called after a Bot API call or multipart upload succeeded. */ + onCallSucceeded?: (method: string) => void; + /** Called after a Bot API call or multipart upload failed for good. */ + onCallFailed?: (method: string, error: unknown) => void; } export interface TelegramBridgeApiRuntime { @@ -1246,6 +1250,8 @@ export function createTelegramAssistantDraftSender(deps: { export function createDefaultTelegramBridgeApiRuntime(deps: { getBotToken: () => string | undefined; recordRuntimeEvent: TelegramBridgeApiRuntimeDeps["recordRuntimeEvent"]; + onCallSucceeded?: TelegramBridgeApiRuntimeDeps["onCallSucceeded"]; + onCallFailed?: TelegramBridgeApiRuntimeDeps["onCallFailed"]; }): TelegramBridgeApiRuntime { return createTelegramBridgeApiRuntime({ client: createTelegramApiClient(deps.getBotToken, { @@ -1255,6 +1261,8 @@ export function createDefaultTelegramBridgeApiRuntime(deps: { maxFileSizeBytes: TELEGRAM_INBOUND_FILE_MAX_BYTES, tempFileMaxAgeMs: TELEGRAM_TEMP_FILE_MAX_AGE_MS, recordRuntimeEvent: deps.recordRuntimeEvent, + onCallSucceeded: deps.onCallSucceeded, + onCallFailed: deps.onCallFailed, }); } @@ -1266,16 +1274,20 @@ export function createTelegramBridgeApiRuntime( body: Record, options?: TelegramApiCallOptions, ): Promise => { + let response: TResponse; try { - return await deps.client.call(method, body, options); + response = await deps.client.call(method, body, options); } catch (error) { deps.recordRuntimeEvent( "api", error, withTelegramTransportDiagnostics(error, { method }), ); + deps.onCallFailed?.(method, error); throw error; } + deps.onCallSucceeded?.(method); + return response; }; return { call: callRecorded, @@ -1285,16 +1297,17 @@ export function createTelegramBridgeApiRuntime( * photos, documents, animations, etc.). * Errors are recorded under the "multipart" category for diagnostics. */ - callMultipart: async ( - method, - fields, - fileField, - filePath, - fileName, - options, - ) => { + callMultipart: async ( + method: string, + fields: Record, + fileField: string, + filePath: string, + fileName: string, + options?: TelegramApiCallOptions, + ): Promise => { + let response: TResponse; try { - return await deps.client.callMultipart( + response = await deps.client.callMultipart( method, fields, fileField, @@ -1308,8 +1321,11 @@ export function createTelegramBridgeApiRuntime( error, withTelegramTransportDiagnostics(error, { method, fileName }), ); + deps.onCallFailed?.(method, error); throw error; } + deps.onCallSucceeded?.(method); + return response; }, /** diff --git a/lib/traffic.ts b/lib/traffic.ts new file mode 100644 index 0000000..24dd27c --- /dev/null +++ b/lib/traffic.ts @@ -0,0 +1,134 @@ +/** + * Telegram bridge traffic counters + * Zones: telegram transport, status + * Owns in-memory received/sent message counts and last-activity time for the status bar + * + * This domain is pure: it counts, it does not read config, call Telegram, or import other domains. + */ + +export interface TelegramTrafficSnapshot { + /** Authorized messages received from the paired owner since the bridge loaded. */ + received: number; + /** Messages the bridge delivered to Telegram since the bridge loaded. */ + sent: number; + /** Messages that could not be delivered after the transport retries. */ + failed: number; + /** Time of the latest received or sent message, when there was one. */ + lastAtMs?: number; +} + +export interface TelegramTrafficCounters { + recordReceived: () => void; + recordSent: () => void; + recordFailed: () => void; + snapshot: () => TelegramTrafficSnapshot; +} + +/** + * Bot API methods that deliver a message the owner can read. Typing actions, + * drafts, edits, reactions and callback answers are not messages. + */ +const TELEGRAM_OUTBOUND_MESSAGE_METHODS: Readonly> = { + sendMessage: true, + sendRichMessage: true, + sendPhoto: true, + sendDocument: true, + sendAudio: true, + sendVideo: true, + sendVoice: true, + sendAnimation: true, + sendVideoNote: true, + sendMediaGroup: true, + sendSticker: true, +}; + +export function isTelegramOutboundMessageMethod(method: string): boolean { + return Object.hasOwn(TELEGRAM_OUTBOUND_MESSAGE_METHODS, method); +} + +export function createTelegramTrafficCounters( + now: () => number = Date.now, +): TelegramTrafficCounters { + let received = 0; + let sent = 0; + let failed = 0; + let lastAtMs: number | undefined; + return { + recordReceived() { + received += 1; + lastAtMs = now(); + }, + recordSent() { + sent += 1; + lastAtMs = now(); + }, + recordFailed() { + failed += 1; + }, + snapshot: () => ({ + received, + sent, + failed, + ...(lastAtMs === undefined ? {} : { lastAtMs }), + }), + }; +} + +export interface TelegramTrafficStatusBridgeDeps { + counters: TelegramTrafficCounters; + /** Current session context, when the bridge has one. */ + getContext: () => TContext | undefined; + updateStatus: (ctx: TContext) => void; + recordRuntimeEvent: ( + category: string, + error: unknown, + details?: Record, + ) => void; + /** Told about every message delivery that failed, e.g. to raise a notice. */ + onDeliveryFailed?: (error: unknown) => void; +} + +export interface TelegramTrafficStatusBridge { + /** Observer for successful Bot API calls; counts delivered messages. */ + onCallSucceeded: (method: string) => void; + /** Observer for failed Bot API calls; counts undelivered messages. */ + onCallFailed: (method: string, error: unknown) => void; + /** Observer for messages accepted from the paired owner. */ + onAuthorizedMessage: (ctx: TContext) => void; +} + +/** + * Counts traffic and refreshes the status bar so the counters stay current. A + * status refresh failure (for example a stale session context) is recorded and + * never propagates into the send or receive path that triggered it. + */ +export function createTelegramTrafficStatusBridge( + deps: TelegramTrafficStatusBridgeDeps, +): TelegramTrafficStatusBridge { + const refreshStatus = (ctx: TContext): void => { + try { + deps.updateStatus(ctx); + } catch (error) { + deps.recordRuntimeEvent("status", error, { phase: "traffic-refresh" }); + } + }; + return { + onCallSucceeded(method) { + if (!isTelegramOutboundMessageMethod(method)) return; + deps.counters.recordSent(); + const ctx = deps.getContext(); + if (ctx !== undefined) refreshStatus(ctx); + }, + onCallFailed(method, error) { + if (!isTelegramOutboundMessageMethod(method)) return; + deps.counters.recordFailed(); + deps.onDeliveryFailed?.(error); + const ctx = deps.getContext(); + if (ctx !== undefined) refreshStatus(ctx); + }, + onAuthorizedMessage(ctx) { + deps.counters.recordReceived(); + refreshStatus(ctx); + }, + }; +} diff --git a/lib/updates.ts b/lib/updates.ts index 8418fdf..121f58b 100644 --- a/lib/updates.ts +++ b/lib/updates.ts @@ -850,6 +850,7 @@ export function createTelegramPairedUpdateRuntime< setAllowedUserId: deps.setAllowedUserId, persistConfig: deps.persistConfig, updateStatus: deps.updateStatus, + onPaired: deps.onPaired, }).pairIfNeeded, answerCallbackQuery: deps.answerCallbackQuery, answerGuestQuery: deps.answerGuestQuery, diff --git a/tests/config.test.ts b/tests/config.test.ts index 83cc77d..7fecffe 100644 --- a/tests/config.test.ts +++ b/tests/config.test.ts @@ -683,6 +683,29 @@ test("Telegram config pairing runtime binds config and status ports", async () = assert.deepEqual(events, ["set:7", "persist", "status:ctx"]); }); +test("Telegram config pairing runtime reports the new owner only after persisting", async () => { + const events: string[] = []; + let allowedUserId: number | undefined; + const runtime = createTelegramUserPairingRuntime({ + getAllowedUserId: () => allowedUserId, + setAllowedUserId: (userId) => { + allowedUserId = userId; + }, + persistConfig: async () => { + events.push("persist"); + }, + updateStatus: () => { + events.push("status"); + }, + onPaired: (userId, ctx: string) => { + events.push(`paired:${userId}:${ctx}`); + }, + }); + assert.equal(await runtime.pairIfNeeded(7, "ctx"), true); + assert.equal(await runtime.pairIfNeeded(8, "ctx"), false); + assert.deepEqual(events, ["persist", "status", "paired:7:ctx"]); +}); + test("Bot token input prefers stored config over env vars", () => { const value = getTelegramBotTokenInputDefault( { diff --git a/tests/notices.test.ts b/tests/notices.test.ts new file mode 100644 index 0000000..d22a259 --- /dev/null +++ b/tests/notices.test.ts @@ -0,0 +1,111 @@ +/** + * Regression tests for the Telegram operator notices + * Covers the toast wording, severity, throttling and failure isolation. + */ + +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + createTelegramNoticeRuntime, + TELEGRAM_DELIVERY_FAILED_NOTICE_COOLDOWN_MS, +} from "../lib/notices.ts"; + +function createNoticeHarness(options: { notifyThrows?: boolean } = {}) { + const notices: string[] = []; + const recorded: string[] = []; + let nowMs = 1_000_000; + let context: string | undefined = "session"; + const runtime = createTelegramNoticeRuntime({ + getContext: () => context, + notify: (ctx: string, message, level) => { + if (options.notifyThrows) throw new Error("stale ctx"); + notices.push(`${level}|${ctx}|${message}`); + }, + recordRuntimeEvent: (category, error, details) => { + recorded.push(`${category}:${(error as Error).message}:${details?.phase}`); + }, + now: () => nowMs, + }); + return { + runtime, + notices, + recorded, + advance: (ms: number) => { + nowMs += ms; + }, + dropContext: () => { + context = undefined; + }, + }; +} + +test("Notices announce pairing and recovery as info toasts", () => { + const { runtime, notices } = createNoticeHarness(); + runtime.paired(777, "ctx-a"); + runtime.recovered("ctx-b"); + assert.deepEqual(notices, [ + "info|ctx-a|Telegram paired with user 777", + "info|ctx-b|Telegram reconnected", + ]); +}); + +test("Delivery failures explain the reason and pick a severity from it", () => { + const harness = createNoticeHarness(); + harness.runtime.deliveryFailed( + new Error("Telegram API sendMessage failed: HTTP 429: Too Many Requests: retry after 12"), + ); + harness.advance(TELEGRAM_DELIVERY_FAILED_NOTICE_COOLDOWN_MS); + harness.runtime.deliveryFailed(new Error("Unable to connect. Is the computer able to access the url?")); + harness.advance(TELEGRAM_DELIVERY_FAILED_NOTICE_COOLDOWN_MS); + harness.runtime.deliveryFailed(new Error("Telegram API sendMessage failed: HTTP 401: Unauthorized")); + harness.advance(TELEGRAM_DELIVERY_FAILED_NOTICE_COOLDOWN_MS); + harness.runtime.deliveryFailed(new Error("message is too long")); + harness.advance(TELEGRAM_DELIVERY_FAILED_NOTICE_COOLDOWN_MS); + harness.runtime.deliveryFailed( + new Error("Telegram API sendRichMessage may have committed before transport failed."), + ); + assert.deepEqual(harness.notices, [ + "warning|session|Telegram: reply not delivered (rate limited 12s)", + "warning|session|Telegram: reply not delivered (offline)", + "error|session|Telegram: reply not delivered (invalid token)", + "error|session|Telegram: reply not delivered (message is too long)", + "warning|session|Telegram: reply delivery unconfirmed (check the chat)", + ]); +}); + +test("Delivery failure notices are throttled and need a live context", () => { + const harness = createNoticeHarness(); + harness.runtime.deliveryFailed(new Error("fetch failed")); + harness.advance(TELEGRAM_DELIVERY_FAILED_NOTICE_COOLDOWN_MS - 1); + harness.runtime.deliveryFailed(new Error("fetch failed")); + assert.equal(harness.notices.length, 1); + harness.advance(1); + harness.runtime.deliveryFailed(new Error("fetch failed")); + assert.equal(harness.notices.length, 2); + harness.advance(TELEGRAM_DELIVERY_FAILED_NOTICE_COOLDOWN_MS); + harness.dropContext(); + harness.runtime.deliveryFailed(new Error("fetch failed")); + assert.equal(harness.notices.length, 2); +}); + +test("A failing toast is recorded and never reaches the caller", () => { + const harness = createNoticeHarness({ notifyThrows: true }); + assert.doesNotThrow(() => { + harness.runtime.paired(777, "ctx"); + harness.runtime.recovered("ctx"); + harness.runtime.deliveryFailed(new Error("fetch failed")); + }); + assert.deepEqual(harness.recorded, [ + "status:stale ctx:notice", + "status:stale ctx:notice", + "status:stale ctx:notice", + ]); +}); + +test("Delivery failure notices carry no control sequences from raw error text", () => { + const harness = createNoticeHarness(); + harness.runtime.deliveryFailed(new Error("bad\u001b]0;pwned\u0007\u001b[31m text")); + assert.equal(harness.notices.length, 1); + assert.doesNotMatch(harness.notices[0], /[\u0000-\u001f\u007f-\u009f]/); +}); diff --git a/tests/pi.test.ts b/tests/pi.test.ts index 2200d70..96e3040 100644 --- a/tests/pi.test.ts +++ b/tests/pi.test.ts @@ -19,6 +19,8 @@ import { hasExtensionContextPendingMessages, isExtensionContextIdle, isExtensionContextPassiveRunMode, + notifyExtensionContext, + setExtensionWorkingMessage, } from "../lib/pi.ts"; type PiRuntimeApiHarness = Parameters< @@ -169,3 +171,24 @@ test("Pi context model prefers omp's live models facade over a frozen ctx.model" assert.equal(getExtensionContextModel(ompCtx), live); assert.equal(getExtensionContextModel(piCtx), frozen); }); + +test("Host UI helpers forward to the context and tolerate a host without a spinner API", () => { + const calls: string[] = []; + const full = { + ui: { + notify: (message: string, level: string) => { + calls.push(`notify:${level}:${message}`); + }, + setWorkingMessage: (message?: string) => { + calls.push(`working:${message}`); + }, + }, + } as unknown as ExtensionContext; + notifyExtensionContext(full, "hello", "warning"); + setExtensionWorkingMessage(full, "busy"); + setExtensionWorkingMessage(full, undefined); + assert.deepEqual(calls, ["notify:warning:hello", "working:busy", "working:undefined"]); + + const withoutSpinner = { ui: {} } as unknown as ExtensionContext; + assert.doesNotThrow(() => setExtensionWorkingMessage(withoutSpinner, "busy")); +}); diff --git a/tests/polling.test.ts b/tests/polling.test.ts index cd8d57b..cc227cb 100644 --- a/tests/polling.test.ts +++ b/tests/polling.test.ts @@ -4,6 +4,9 @@ */ import assert from "node:assert/strict"; +import { mkdtempSync, readFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import test from "node:test"; import { @@ -29,6 +32,7 @@ import { stopTelegramPollingRuntime, TELEGRAM_ALLOWED_UPDATES, } from "../lib/polling.ts"; +import { createTelegramConfigStore } from "../lib/config.ts"; const TEST_CONTEXT = "ctx"; @@ -601,6 +605,103 @@ test("Poll loop runner binds config, status, and transport ports", async () => { assert.deepEqual(events, ["deleteWebhook", "handle:ctx:6", "persist:6"]); }); +test("Poll loop keeps the paired user when the offset is committed after the initial sync replaced the stored config", async () => { + const dir = mkdtempSync(join(tmpdir(), "telegram-pairing-")); + const configPath = join(dir, "telegram.json"); + const store = createTelegramConfigStore({ agentDir: dir, configPath }); + await store.load(); + store.update((config) => { + config.botToken = "123:abc"; + }); + await store.persist(); + let calls = 0; + const runPollLoop = createTelegramPollLoopRunner({ + getConfig: store.get, + deleteWebhook: async () => {}, + getUpdates: async () => { + calls += 1; + if (calls === 1) return [{ update_id: 1 }]; // initial sync + if (calls === 2) return [{ update_id: 2 }]; // the owner's /start + throw new DOMException("stop", "AbortError"); + }, + persistConfig: store.persist, + handleUpdate: async () => { + store.setAllowedUserId(777); + await store.persist(); + }, + updateStatus: () => {}, + sleep: async () => {}, + }); + await runPollLoop("ctx", new AbortController().signal); + assert.equal(store.getAllowedUserId(), 777); + const saved = JSON.parse(readFileSync(configPath, "utf8")); + assert.equal(saved.allowedUserId, 777); + assert.equal(saved.lastUpdateId, 2); +}); + +test("Poll loop reports recovery once, only after a failed attempt", async () => { + const run = async (failures: number): Promise => { + const events: string[] = []; + let calls = 0; + const runPollLoop = createTelegramPollLoopRunner({ + getConfig: () => ({ botToken: "123:abc", lastUpdateId: 1 }), + deleteWebhook: async () => {}, + getUpdates: async () => { + calls += 1; + if (calls <= failures) throw new Error("network down"); + if (calls <= failures + 2) return []; + throw new DOMException("stop", "AbortError"); + }, + persistConfig: async () => {}, + handleUpdate: async () => {}, + updateStatus: (_ctx: string, message?: string) => { + events.push(`status:${message ?? "ok"}`); + }, + onRecovered: (ctx: string) => { + events.push(`recovered:${ctx}`); + }, + sleep: async () => {}, + }); + await runPollLoop("ctx", new AbortController().signal); + return events; + }; + assert.deepEqual(await run(2), [ + "status:network down", + "status:network down", + "status:ok", + "recovered:ctx", + ]); + assert.deepEqual(await run(0), []); +}); + +test("Poll loop keeps polling when the recovery notice itself fails", async () => { + const recorded: string[] = []; + let calls = 0; + const runPollLoop = createTelegramPollLoopRunner({ + getConfig: () => ({ botToken: "123:abc", lastUpdateId: 1 }), + deleteWebhook: async () => {}, + getUpdates: async () => { + calls += 1; + if (calls === 1) throw new Error("network down"); + if (calls === 2) return []; + throw new DOMException("stop", "AbortError"); + }, + persistConfig: async () => {}, + handleUpdate: async () => {}, + updateStatus: () => {}, + onRecovered: () => { + throw new Error("toast broke"); + }, + sleep: async () => {}, + recordRuntimeEvent: (category, error, details) => { + recorded.push(`${category}:${(error as Error).message}:${details?.phase}`); + }, + }); + await runPollLoop("ctx", new AbortController().signal); + assert.equal(calls, 3); + assert.ok(recorded.includes("polling:toast broke:recovered-notice"), recorded.join("|")); +}); + test("Poll loop runner ignores stale-context status failures while retrying", async () => { const config = { botToken: "123:abc", lastUpdateId: 1 }; const events: string[] = []; @@ -612,6 +713,7 @@ test("Poll loop runner ignores stale-context status failures while retrying", as getUpdates: async () => { calls += 1; if (calls === 1) throw new Error("network down"); + if (calls === 2) return []; throw new DOMException("stop", "AbortError"); }, persistConfig: async () => {}, @@ -637,6 +739,74 @@ test("Poll loop runner ignores stale-context status failures while retrying", as ]); }); +test("Poll loop keeps the error status until a poll succeeds", async () => { + const events: string[] = []; + let calls = 0; + await runTelegramPollLoop({ + ctx: TEST_CONTEXT, + signal: new AbortController().signal, + config: { botToken: "123:abc", lastUpdateId: 1 }, + deleteWebhook: async () => {}, + getUpdates: async () => { + calls += 1; + if (calls <= 2) throw new Error("rate limited"); + if (calls === 3) return []; + throw new DOMException("stop", "AbortError"); + }, + persistConfig: async () => {}, + handleUpdate: async () => {}, + onErrorStatus: (message) => { + events.push(`error:${message}`); + }, + onStatusReset: () => { + events.push("reset"); + }, + sleep: async () => { + events.push("sleep"); + }, + }); + assert.deepEqual(events, [ + "error:rate limited", + "sleep", + "error:rate limited", + "sleep", + "reset", + ]); +}); + +test("Poll loop does not report recovery when only an update handler failed", async () => { + const events: string[] = []; + let calls = 0; + await runTelegramPollLoop({ + ctx: TEST_CONTEXT, + signal: new AbortController().signal, + config: { botToken: "123:abc", lastUpdateId: 1 }, + maxUpdateFailures: 3, + deleteWebhook: async () => {}, + getUpdates: async () => { + calls += 1; + if (calls <= 3) return [{ update_id: 6 }]; + throw new DOMException("stop", "AbortError"); + }, + persistConfig: async () => {}, + handleUpdate: async () => { + throw new Error("handler failed"); + }, + onErrorStatus: (message) => { + events.push(`error:${message}`); + }, + onStatusReset: () => { + events.push("reset"); + }, + onRecovered: () => { + events.push("recovered"); + }, + sleep: async () => {}, + }); + assert.equal(events.includes("recovered"), false, events.join("|")); + assert.ok(events.includes("reset"), events.join("|")); +}); + test("Poll loop initializes lastUpdateId and processes updates", async () => { const handled: number[] = []; const config: { botToken: string; lastUpdateId?: number } = { @@ -893,6 +1063,7 @@ test("Poll loop reports retryable errors and sleeps before retrying", async () = if (calls === 1) { throw new Error("network down"); } + if (calls === 2) return []; throw new DOMException("stop", "AbortError"); }, persistConfig: async () => {}, diff --git a/tests/queue.test.ts b/tests/queue.test.ts index 925a808..e542eb7 100644 --- a/tests/queue.test.ts +++ b/tests/queue.test.ts @@ -48,6 +48,7 @@ import { executeTelegramControlItemRuntime, executeTelegramQueueDispatchPlan, formatQueuedTelegramItemsStatus, + formatQueuedTelegramItemsPreview, getNextTelegramToolExecutionCount, getTelegramQueueItemAdmissionMode, getTelegramQueueLaneContract, @@ -405,6 +406,42 @@ test("Control-lane items sort before priority and default prompt items", () => { ); }); +test("Queued item preview lists summaries in dispatch order and skips empty ones", () => { + const defaultPrompt: TelegramQueueItem = createQueueTestPromptTurn({ + queueOrder: 10, + queueLane: "default", + laneOrder: 0, + statusSummary: "default", + }); + const emptyPrompt: TelegramQueueItem = createQueueTestPromptTurn({ + queueOrder: 13, + queueLane: "default", + laneOrder: 1, + statusSummary: " ", + }); + const priorityPrompt: TelegramQueueItem = createQueueTestPromptTurn({ + queueOrder: 11, + queueLane: "priority", + laneOrder: 0, + statusSummary: "priority", + }); + const controlItem: TelegramQueueItem = createQueueTestControlItem({ + queueOrder: 12, + statusSummary: "control", + }); + const stored = [defaultPrompt, emptyPrompt, controlItem, priorityPrompt]; + assert.deepEqual(formatQueuedTelegramItemsPreview(stored), [ + "control", + "priority", + "default", + ]); + assert.deepEqual( + stored.map((item) => item.statusSummary), + ["default", " ", "control", "priority"], + ); + assert.deepEqual(formatQueuedTelegramItemsPreview([]), []); +}); + test("Queue mutation helpers remove prompt items by Telegram message id", () => { const promptItem: TelegramQueueItem = createQueueTestPromptTurn({ replyToMessageId: 1, diff --git a/tests/status.test.ts b/tests/status.test.ts index 301302d..cd58999 100644 --- a/tests/status.test.ts +++ b/tests/status.test.ts @@ -10,6 +10,7 @@ import { buildTelegramBridgeStatusLines, buildTelegramRuntimeEventLines, buildTelegramStatusBarText, + buildTelegramStatusBarLines, clearTelegramStatusLineProviders, createTelegramBridgeStatusRuntime, createTelegramRuntimeDiagnosticsSnapshotScheduler, @@ -19,6 +20,9 @@ import { createTelegramStatusSnapshot, createTelegramStatusRuntime, getTelegramStatusBarProcessingStatus, + formatTelegramWorkingMessage, + sanitizeTelegramStatusText, + summarizeTelegramStatusError, recordStructuredTelegramRuntimeEvent, recordTelegramRuntimeEvent, registerTelegramStatusLineProvider, @@ -111,248 +115,165 @@ test("Status runtime diagnostics scheduler coalesces snapshot persists", async ( assert.deepEqual(errors, []); }); -test("Status bar text renders bridge connection and queue states", () => { - const theme = { - fg: (token: string, text: string) => `<${token}>${text}`, +const statusBarTheme = { + fg: (token: string, text: string) => `<${token}>${text}`, +}; + +function plainStatusText(text: string): string { + return text.replace(/<\/?[a-z]+>/g, ""); +} + +function escapeRegExp(text: string): string { + return text.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); +} + +test("Status bar text keeps the label, state color and queue for every connection state", () => { + const base = { + hasBotToken: true, + pollingActive: true, + paired: true, + compactionInProgress: false, + processing: false, + queuedStatus: "", }; - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: false, - pollingActive: false, - paired: false, - compactionInProgress: false, - processing: false, - queuedStatus: "", - }), - "telegram not configured", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: true, - paired: true, - compactionInProgress: false, - processing: true, - queuedStatus: " +1", - }), - "telegram active +1", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: true, - paired: true, - compactionInProgress: false, - processing: true, - processingStatus: "dispatching", - queuedStatus: " +1", - }), - "telegram active +1", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: true, - paired: true, - compactionInProgress: false, - processing: true, - processingStatus: "active", - queuedStatus: "", - }), - "telegram active", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: true, - paired: true, - compactionInProgress: true, - processing: false, - queuedStatus: "", - }), - "telegram connected", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: true, - paired: true, - compactionInProgress: true, - processing: true, - processingStatus: "active", - queuedStatus: "", - }), - "telegram active", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: false, - paired: true, - instanceThreadName: "Aurora", - compactionInProgress: false, - processing: true, - processingStatus: "queued", - queuedStatus: " +2", - }), - "telegram disconnected +2", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: false, - paired: true, - busRole: "follower", - compactionInProgress: false, - processing: false, - queuedStatus: "", - }), - "telegram follower", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: false, - paired: true, - busRole: "follower", - instanceThreadName: "Amber", - compactionInProgress: false, - processing: true, - processingStatus: "active", - queuedStatus: "", - }), - "Amber active", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: false, - paired: true, - busRole: "follower", - busLifecyclePhase: "electing", - compactionInProgress: false, - processing: false, - queuedStatus: "", - }), - "telegram electing", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: false, - paired: true, - busLifecyclePhase: "electing", - compactionInProgress: false, - processing: false, - queuedStatus: "", - }), - "telegram electing", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: false, - paired: true, - busRole: "follower", - instanceThreadName: "Follower", - compactionInProgress: false, - processing: false, - queuedStatus: "", - }), - "telegram follower", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: false, - paired: true, - busRole: "follower", - instanceThreadName: "Lname", - compactionInProgress: false, - processing: false, - queuedStatus: "", - }), - "Lname follower", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: false, - paired: true, - busRole: "follower", - instanceSlot: "O", - instanceThreadName: "extensions Follower", - compactionInProgress: false, - processing: false, - queuedStatus: "", - }), - "extensions Follower follower", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: false, - paired: true, - busRole: "follower", - instanceSlot: "O", - instanceThreadName: "Oname", - compactionInProgress: false, - processing: false, - queuedStatus: "", - }), - "Oname follower", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: true, - paired: true, - busRole: "leader", - compactionInProgress: false, - processing: false, - queuedStatus: "", - }), - "telegram leader", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: true, - paired: true, - busRole: "leader", - instanceThreadName: "🌙 A-identity", - compactionInProgress: false, - processing: false, - queuedStatus: "", - }), - "🌙 A-identity leader", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: true, - paired: false, - compactionInProgress: false, - processing: true, - processingStatus: "queued", - queuedStatus: " +1", - }), - "telegram awaiting pairing +1", - ); - assert.equal( - buildTelegramStatusBarText(theme, { - hasBotToken: true, - pollingActive: true, - paired: true, - compactionInProgress: false, - processing: false, - queuedStatus: "", + const cases: Array<{ + state: Partial[1]>; + label: string; + token: string; + word: string; + queued?: string; + }> = [ + { state: { hasBotToken: false, pollingActive: false, paired: false }, label: "telegram", token: "muted", word: "not configured" }, + { state: { processing: true, queuedStatus: " +1" }, label: "telegram", token: "warning", word: "active", queued: " +1" }, + { state: { processing: true, processingStatus: "dispatching", queuedStatus: " +1" }, label: "telegram", token: "warning", word: "active", queued: " +1" }, + { state: { processing: true, processingStatus: "active" }, label: "telegram", token: "warning", word: "active" }, + { state: { compactionInProgress: true }, label: "telegram", token: "success", word: "connected" }, + { state: { compactionInProgress: true, processing: true, processingStatus: "active" }, label: "telegram", token: "warning", word: "active" }, + { state: { pollingActive: false, instanceThreadName: "Aurora", processing: true, processingStatus: "queued", queuedStatus: " +2" }, label: "telegram", token: "muted", word: "disconnected", queued: " +2" }, + { state: { pollingActive: false, busRole: "follower" }, label: "telegram", token: "success", word: "follower" }, + { state: { pollingActive: false, busRole: "follower", instanceThreadName: "Amber", processing: true, processingStatus: "active" }, label: "Amber", token: "warning", word: "active" }, + { state: { pollingActive: false, busRole: "follower", busLifecyclePhase: "electing" }, label: "telegram", token: "warning", word: "electing" }, + { state: { pollingActive: false, busLifecyclePhase: "electing" }, label: "telegram", token: "warning", word: "electing" }, + { state: { pollingActive: false, busRole: "follower", instanceThreadName: "Follower" }, label: "telegram", token: "success", word: "follower" }, + { state: { pollingActive: false, busRole: "follower", instanceThreadName: "Lname" }, label: "Lname", token: "success", word: "follower" }, + { state: { pollingActive: false, busRole: "follower", instanceSlot: "O", instanceThreadName: "extensions Follower" }, label: "extensions Follower", token: "success", word: "follower" }, + { state: { pollingActive: false, busRole: "follower", instanceSlot: "O", instanceThreadName: "Oname" }, label: "Oname", token: "success", word: "follower" }, + { state: { busRole: "leader" }, label: "telegram", token: "success", word: "leader" }, + { state: { busRole: "leader", instanceThreadName: "🌙 A-identity" }, label: "🌙 A-identity", token: "success", word: "leader" }, + { state: { paired: false, processing: true, processingStatus: "queued", queuedStatus: " +1" }, label: "telegram", token: "warning", word: "awaiting pairing", queued: " +1" }, + ]; + for (const { state, label, token, word, queued } of cases) { + const text = buildTelegramStatusBarText(statusBarTheme, { ...base, ...state }); + const expected = new RegExp( + `^${escapeRegExp(label)} <${token}>\\S+ ${word}${queued ? `${escapeRegExp(queued)}` : ""}`, + ); + assert.match(text, expected, JSON.stringify(state)); + } + assert.match( + buildTelegramStatusBarText(statusBarTheme, { + ...base, error: "typing failed", }), - "telegram error typing failed", + /^telegram<\/accent> \S+ error<\/error> typing failed<\/muted>$/, ); }); +test("Status bar shows bot, message counters and last activity only while the bridge is up", () => { + const lastAtMs = new Date(2026, 9, 1, 14, 32).getTime(); + const connected = { + hasBotToken: true, + pollingActive: true, + paired: true, + compactionInProgress: false, + processing: false, + queuedStatus: "", + botUsername: "demo_bot", + traffic: { received: 12, sent: 11, lastAtMs }, + }; + const up = plainStatusText(buildTelegramStatusBarText(statusBarTheme, connected)); + assert.match(up, /connected/); + assert.match(up, /@demo_bot/); + assert.match(up, /12 \S*11/); + assert.match(up, /14:32/); + + const withoutActivity = plainStatusText( + buildTelegramStatusBarText(statusBarTheme, { + ...connected, + traffic: { received: 0, sent: 0 }, + }), + ); + assert.match(withoutActivity, /0 \S*0/); + assert.doesNotMatch(withoutActivity, /\d\d:\d\d/); + + const working = plainStatusText( + buildTelegramStatusBarText(statusBarTheme, { + ...connected, + processing: true, + processingStatus: "active", + }), + ); + assert.match(working, /active/); + assert.match(working, /@demo_bot/); + assert.match(working, /12 \S*11/); + + const offline = plainStatusText( + buildTelegramStatusBarText(statusBarTheme, { ...connected, pollingActive: false }), + ); + assert.doesNotMatch(offline, /@demo_bot|14:32/); + assert.match(offline, /disconnected/); +}); + +test("Status bar tells the operator how to leave a non-working state", () => { + const base = { + hasBotToken: true, + pollingActive: false, + paired: true, + compactionInProgress: false, + processing: false, + queuedStatus: "", + }; + assert.match( + plainStatusText(buildTelegramStatusBarText(statusBarTheme, { ...base, hasBotToken: false, paired: false })), + /\/telegram-setup/, + ); + const pairing = plainStatusText( + buildTelegramStatusBarText(statusBarTheme, { ...base, paired: false, botUsername: "demo_bot" }), + ); + assert.match(pairing, /\/start/); + assert.match(pairing, /@demo_bot/); + assert.match( + plainStatusText(buildTelegramStatusBarText(statusBarTheme, base)), + /\/telegram-connect/, + ); +}); + +test("Status bar follows the host symbol preset and falls back without one", () => { + const state = { + hasBotToken: true, + pollingActive: true, + paired: true, + compactionInProgress: false, + processing: false, + queuedStatus: "", + botUsername: "demo_bot", + traffic: { received: 3, sent: 2, lastAtMs: new Date(2026, 9, 1, 9, 5).getTime() }, + }; + const ascii = plainStatusText( + buildTelegramStatusBarText( + { + ...statusBarTheme, + status: { enabled: "[x]" }, + sep: { dot: " - " }, + icon: { input: "in:", output: "out:", time: "t:" }, + }, + state, + ), + ); + assert.equal(ascii, "telegram [x] connected - @demo_bot - in:3 out:2 - t: 09:05"); + const fallback = plainStatusText(buildTelegramStatusBarText(statusBarTheme, state)); + assert.match(fallback, /^telegram \S+ connected · @demo_bot · \S*3 \S*2 · \S+ 09:05$/u); +}); + test("Status runtime updates the status bar and exposes bridge lines", () => { const events: string[] = []; const ctx = { @@ -390,9 +311,9 @@ test("Status runtime updates the status bar and exposes bridge lines", () => { }), }); runtime.updateStatus(ctx, "demo error"); - assert.equal( + assert.match( events[0], - "telegram:telegram error demo error", + /^telegram:telegram<\/accent> \S+ error<\/error> demo error<\/muted>$/, ); assert.deepEqual(runtime.getStatusLines().slice(0, 3), [ "connection:", @@ -401,6 +322,237 @@ test("Status runtime updates the status bar and exposes bridge lines", () => { ]); }); +test("Status runtime keeps theme colors in the TUI through a widget and stays plain elsewhere", () => { + const createContext = (mode: string | undefined) => { + const events: string[] = []; + return { + events, + ctx: { + mode, + ui: { + theme: statusBarTheme, + setStatus: (key: string, text: string | undefined) => { + events.push(`status:${key}:${text}`); + }, + setWidget: ( + key: string, + content: string[] | undefined, + options?: { placement?: string }, + ) => { + events.push(`widget:${key}:${options?.placement}:${content?.join("|")}`); + }, + }, + }, + }; + }; + const runtime = createTelegramStatusRuntime({ + getStatusBarState: () => ({ + hasBotToken: true, + pollingActive: true, + paired: true, + compactionInProgress: false, + processing: false, + queuedStatus: "", + }), + getBridgeStatusLineState: () => { + throw new Error("not used"); + }, + }); + + const tui = createContext("tui"); + runtime.updateStatus(tui.ctx); + assert.equal(tui.events.length, 2); + assert.match(tui.events[0], /^widget:telegram:belowEditor:telegram<\/accent> \S+ connected<\/success>$/); + assert.equal(tui.events[1], "status:telegram:undefined"); + + for (const mode of ["rpc", "json", "print", undefined]) { + const other = createContext(mode); + runtime.updateStatus(other.ctx); + assert.equal(other.events.length, 1, String(mode)); + assert.match(other.events[0], /^status:telegram:telegram<\/accent> \S+ connected<\/success>$/); + } + + const noWidgetApi = createContext("tui"); + delete (noWidgetApi.ctx.ui as { setWidget?: unknown }).setWidget; + runtime.updateStatus(noWidgetApi.ctx); + assert.equal(noWidgetApi.events.length, 1); + assert.match(noWidgetApi.events[0], /^status:telegram:/); +}); + +test("Error summaries turn transport failures into a short state and next step", () => { + const cases: Array<[string, string | undefined, string | undefined, string | undefined]> = [ + ["Telegram API getUpdates failed: HTTP 401: Unauthorized", "fatal", "invalid token", "/telegram-setup"], + ["Telegram API sendMessage failed: HTTP 429: Too Many Requests: retry after 12", "retrying", "rate limited 12s", "retrying"], + ["Telegram API sendMessage failed: HTTP 429", "retrying", "rate limited", "retrying"], + ["Conflict: terminated by other getUpdates request; make sure that only one bot instance is running", "attention", "another poller is active", "/telegram-status"], + ["Telegram API getUpdates failed: HTTP 502: Bad Gateway", "retrying", "Telegram unavailable", "retrying"], + ["Unable to connect. Is the computer able to access the url?", "retrying", "offline", "retrying"], + ["fetch failed", "retrying", "offline", "retrying"], + ["Telegram API sendRichMessage may have committed before transport failed.", "attention", "delivery unconfirmed", undefined], + ["typing failed", undefined, undefined, undefined], + ["settingsManager.reload is not a function", undefined, undefined, undefined], + ]; + for (const [message, kind, label, hint] of cases) { + const summary = summarizeTelegramStatusError(message); + assert.equal(summary?.kind, kind, message); + assert.equal(summary?.label, label, message); + assert.equal(summary?.hint, hint, message); + } +}); + +test("Status bar shows classified errors with their next step and keeps unknown ones verbatim", () => { + const base = { + hasBotToken: true, + pollingActive: true, + paired: true, + compactionInProgress: false, + processing: false, + queuedStatus: "", + }; + const fatal = buildTelegramStatusBarText(statusBarTheme, { + ...base, + error: "Telegram API getUpdates failed: HTTP 401: Unauthorized", + }); + assert.match(fatal, /\S+ invalid token<\/error>/); + assert.match(plainStatusText(fatal), /\/telegram-setup/); + const retrying = buildTelegramStatusBarText(statusBarTheme, { + ...base, + error: "Unable to connect. Is the computer able to access the url?", + }); + assert.match(retrying, /\S+ offline<\/warning>/); + assert.match(plainStatusText(retrying), /retrying/); + const attention = buildTelegramStatusBarText(statusBarTheme, { + ...base, + error: "HTTP 409 Conflict", + }); + assert.match(attention, /\S+ another poller is active<\/warning>/); + assert.match(plainStatusText(attention), /\/telegram-status/); + assert.match( + buildTelegramStatusBarText(statusBarTheme, { ...base, error: "typing failed" }), + /\S+ error<\/error> typing failed<\/muted>/, + ); + const long = plainStatusText( + buildTelegramStatusBarText(statusBarTheme, { ...base, error: "x".repeat(200) }), + ); + assert.ok(long.includes("…")); + assert.equal(long.includes("x".repeat(101)), false); +}); + +test("Status bar lines carry no terminal control sequences from Telegram-controlled text", () => { + const hostile = "\u001b]0;pwned\u0007\u001b[31mred\u001b[0m\u202e"; + const base = { + hasBotToken: true, + pollingActive: true, + paired: true, + compactionInProgress: false, + processing: true, + queuedStatus: " +1", + instanceThreadName: `thread${hostile}`, + botUsername: `bot${hostile}`, + queuedPreview: [`queued${hostile}`], + }; + const unsafe = /[\u0000-\u001f\u007f-\u009f\u202a-\u202e]/; + for (const state of [ + base, + { ...base, error: `boom${hostile}` }, + { ...base, paired: false }, + ]) { + for (const line of buildTelegramStatusBarLines(statusBarTheme, state)) { + assert.doesNotMatch(line, unsafe, JSON.stringify(line)); + } + } + assert.equal(sanitizeTelegramStatusText(" a\u001b[0m b\u0000 "), "a [0m b"); +}); + +test("Status bar previews the next queued turns only while the bridge is up", () => { + const base = { + hasBotToken: true, + pollingActive: true, + paired: true, + compactionInProgress: false, + processing: true, + queuedStatus: " +3", + }; + const three = buildTelegramStatusBarLines(statusBarTheme, { + ...base, + queuedPreview: ["one", "two", "three"], + }); + assert.equal(three.length, 2); + assert.match(plainStatusText(three[1]), /next: "one" · "two" · \+1 more/); + const two = buildTelegramStatusBarLines(statusBarTheme, { + ...base, + queuedPreview: ["one", "two"], + }); + assert.match(plainStatusText(two[1]), /next: "one" · "two"$/); + for (const state of [ + { ...base, queuedPreview: [] }, + { ...base, queuedPreview: ["one"], pollingActive: false }, + { ...base, queuedPreview: ["one"], paired: false }, + { ...base, queuedPreview: ["one"], error: "typing failed" }, + ]) { + assert.equal(buildTelegramStatusBarLines(statusBarTheme, state).length, 1, JSON.stringify(state)); + } +}); + +test("Status bar surfaces failed deliveries only when there are some", () => { + const state = { + hasBotToken: true, + pollingActive: true, + paired: true, + compactionInProgress: false, + processing: false, + queuedStatus: "", + traffic: { received: 12, sent: 11, failed: 2 }, + }; + assert.match(plainStatusText(buildTelegramStatusBarText(statusBarTheme, state)), /12 \S*11 \S*2/); + const none = plainStatusText( + buildTelegramStatusBarText(statusBarTheme, { + ...state, + traffic: { received: 12, sent: 11, failed: 0 }, + }), + ); + assert.equal(none.includes("✘"), false); +}); + +test("Status runtime puts the queue preview in the widget and keeps the plain status to one line", () => { + const runtime = createTelegramStatusRuntime({ + getStatusBarState: () => ({ + hasBotToken: true, + pollingActive: true, + paired: true, + compactionInProgress: false, + processing: true, + queuedStatus: " +1", + queuedPreview: ["one"], + }), + getBridgeStatusLineState: () => { + throw new Error("not used"); + }, + }); + const widgets: string[][] = []; + const statuses: Array = []; + const ui = { + theme: statusBarTheme, + setStatus: (_key: string, text: string | undefined) => { + statuses.push(text); + }, + setWidget: (_key: string, content: string[] | undefined) => { + widgets.push(content ?? []); + }, + }; + runtime.updateStatus({ mode: "tui", ui }); + assert.equal(widgets[0].length, 2); + assert.match(plainStatusText(widgets[0][1]), /next: "one"/); + runtime.updateStatus({ mode: "rpc", ui }); + assert.equal(statuses.at(-1)?.includes("next:"), false); +}); + +test("Working message names the Telegram prompt and ignores control characters", () => { + assert.equal(formatTelegramWorkingMessage("say pong"), 'Answering Telegram · "say pong"'); + assert.equal(formatTelegramWorkingMessage(" "), "Answering Telegram"); + assert.equal(formatTelegramWorkingMessage("a\u001b[0mb"), 'Answering Telegram · "a [0mb"'); +}); + test("Status lines expose thread reconciliation state", () => { const lines = buildTelegramBridgeStatusLines({ botUsername: "demo_bot", @@ -550,9 +702,9 @@ test("Bridge status runtime stays active while tools run after queue changes", ( }, }, }); - assert.equal( + assert.match( events[0], - "telegram:telegram active", + /^telegram:telegram<\/accent> \S+ active<\/warning>/, ); }); @@ -589,9 +741,9 @@ test("Bridge status runtime builds status state from live ports", () => { }, }, }); - assert.equal( + assert.match( events[0], - "telegram:telegram active +1", + /^telegram:telegram<\/accent> \S+ active<\/warning> \+1<\/success>/, ); assert.deepEqual(runtime.getStatusLines(), [ "connection:", diff --git a/tests/traffic.test.ts b/tests/traffic.test.ts new file mode 100644 index 0000000..bc01cc7 --- /dev/null +++ b/tests/traffic.test.ts @@ -0,0 +1,163 @@ +/** + * Regression tests for the Telegram traffic counters + * Covers what counts as a message, last-activity tracking, failed deliveries, and the /telegram-status line. + */ + +import assert from "node:assert/strict"; +import test from "node:test"; + +import { buildTelegramBridgeStatusLines } from "../lib/status.ts"; +import { + createTelegramTrafficCounters, + createTelegramTrafficStatusBridge, + isTelegramOutboundMessageMethod, +} from "../lib/traffic.ts"; + +test("Traffic counters count every direction and remember the latest message", () => { + let nowMs = 1_000; + const counters = createTelegramTrafficCounters(() => nowMs); + assert.deepEqual(counters.snapshot(), { received: 0, sent: 0, failed: 0 }); + counters.recordReceived(); + nowMs = 2_000; + counters.recordSent(); + counters.recordSent(); + nowMs = 3_000; + counters.recordFailed(); + assert.deepEqual(counters.snapshot(), { + received: 1, + sent: 2, + failed: 1, + lastAtMs: 2_000, + }); +}); + +test("Only calls that deliver a readable message count as sent", () => { + for (const method of ["sendMessage", "sendRichMessage", "sendPhoto", "sendDocument", "sendVoice"]) { + assert.equal(isTelegramOutboundMessageMethod(method), true, method); + } + for (const method of [ + "sendChatAction", + "sendMessageDraft", + "editMessageText", + "setMessageReaction", + "answerCallbackQuery", + "getUpdates", + "toString", + ]) { + assert.equal(isTelegramOutboundMessageMethod(method), false, method); + } +}); + +test("Status lines report the message counters and the time of the last one", () => { + const baseState = { + botUsername: "demo_bot", + allowedUserId: 7, + pollingActive: true, + pendingDispatch: false, + compactionInProgress: false, + activeToolExecutions: 0, + pendingModelSwitch: false, + queuedItems: [], + recentRuntimeEvents: [], + }; + const lastAtMs = new Date(2026, 9, 1, 14, 32).getTime(); + const clean = buildTelegramBridgeStatusLines({ + ...baseState, + traffic: { received: 12, sent: 11, failed: 0, lastAtMs }, + }); + assert.ok( + clean.includes("- messages: 12 received, 11 sent, last 14:32"), + clean.join("\n"), + ); + const withFailures = buildTelegramBridgeStatusLines({ + ...baseState, + traffic: { received: 12, sent: 11, failed: 2, lastAtMs }, + }); + assert.ok( + withFailures.includes("- messages: 12 received, 11 sent, 2 failed, last 14:32"), + withFailures.join("\n"), + ); + const withoutTraffic = buildTelegramBridgeStatusLines(baseState); + assert.equal(withoutTraffic.some((line) => line.startsWith("- messages:")), false); +}); + +test("Traffic bridge counts delivered messages and refreshes the status with the current context", () => { + const counters = createTelegramTrafficCounters(); + const refreshed: string[] = []; + let context: string | undefined = "session-a"; + const bridge = createTelegramTrafficStatusBridge({ + counters, + getContext: () => context, + updateStatus: (ctx: string) => { + refreshed.push(ctx); + }, + recordRuntimeEvent: () => {}, + }); + bridge.onCallSucceeded("sendChatAction"); + bridge.onCallSucceeded("editMessageText"); + assert.deepEqual(counters.snapshot(), { received: 0, sent: 0, failed: 0 }); + assert.deepEqual(refreshed, []); + bridge.onCallSucceeded("sendRichMessage"); + assert.equal(counters.snapshot().sent, 1); + assert.deepEqual(refreshed, ["session-a"]); + context = undefined; + bridge.onCallSucceeded("sendMessage"); + assert.equal(counters.snapshot().sent, 2); + assert.deepEqual(refreshed, ["session-a"]); + bridge.onAuthorizedMessage("session-b"); + assert.equal(counters.snapshot().received, 1); + assert.deepEqual(refreshed, ["session-a", "session-b"]); +}); + +test("Traffic bridge counts undelivered messages, reports them and refreshes the status", () => { + const counters = createTelegramTrafficCounters(); + const refreshed: string[] = []; + const failures: string[] = []; + const bridge = createTelegramTrafficStatusBridge({ + counters, + getContext: () => "session", + updateStatus: (ctx: string) => { + refreshed.push(ctx); + }, + recordRuntimeEvent: () => {}, + onDeliveryFailed: (error) => { + failures.push((error as Error).message); + }, + }); + bridge.onCallFailed("getUpdates", new Error("poll failed")); + bridge.onCallFailed("sendChatAction", new Error("typing failed")); + assert.equal(counters.snapshot().failed, 0); + assert.deepEqual(failures, []); + assert.deepEqual(refreshed, []); + bridge.onCallFailed("sendMessage", new Error("HTTP 500")); + assert.equal(counters.snapshot().failed, 1); + assert.equal(counters.snapshot().sent, 0); + assert.deepEqual(failures, ["HTTP 500"]); + assert.deepEqual(refreshed, ["session"]); +}); + +test("Traffic bridge records a failed status refresh instead of breaking the message path", () => { + const counters = createTelegramTrafficCounters(); + const recorded: string[] = []; + const bridge = createTelegramTrafficStatusBridge({ + counters, + getContext: () => "session", + updateStatus: () => { + throw new Error("stale ctx"); + }, + recordRuntimeEvent: (category, error, details) => { + recorded.push(`${category}:${(error as Error).message}:${details?.phase}`); + }, + }); + bridge.onCallSucceeded("sendMessage"); + bridge.onAuthorizedMessage("session"); + bridge.onCallFailed("sendMessage", new Error("HTTP 500")); + assert.equal(counters.snapshot().received, 1); + assert.equal(counters.snapshot().sent, 1); + assert.equal(counters.snapshot().failed, 1); + assert.deepEqual(recorded, [ + "status:stale ctx:traffic-refresh", + "status:stale ctx:traffic-refresh", + "status:stale ctx:traffic-refresh", + ]); +});