From 005b601b133287e75aabd7c693652ea04b7e413d Mon Sep 17 00:00:00 2001 From: USO Status Date: Mon, 26 Jan 2026 20:30:56 +0100 Subject: [PATCH] 1589: final draft - pre log --- scripts/run-test-gateway.sh | 68 ++++ .../reply/agent-runner-execution.ts | 15 + src/auto-reply/reply/agent-runner.ts | 109 +++++- src/auto-reply/reply/block-reply-pipeline.ts | 12 +- .../reply/status-updates-integration.ts | 201 ++++++++++ src/auto-reply/reply/status-updates.ts | 346 ++++++++++++++++++ src/auto-reply/types.ts | 9 +- src/config/schema.ts | 8 + src/config/types.agent-defaults.ts | 6 + src/config/types.agents.ts | 4 +- src/config/types.base.ts | 25 ++ src/config/zod-schema.agent-defaults.ts | 3 + src/config/zod-schema.agent-runtime.ts | 3 + src/config/zod-schema.core.ts | 32 ++ src/telegram/bot-handlers.ts | 6 + src/telegram/bot-message-dispatch.ts | 2 + src/telegram/bot-native-commands.ts | 1 + src/telegram/bot/delivery.test.ts | 5 + src/telegram/bot/delivery.ts | 78 +++- src/telegram/send.intermittent.test.ts | 243 ++++++++++++ src/telegram/send.ts | 105 +++++- src/telegram/sent-message-cache.ts | 34 +- 22 files changed, 1291 insertions(+), 24 deletions(-) create mode 100755 scripts/run-test-gateway.sh create mode 100644 src/auto-reply/reply/status-updates-integration.ts create mode 100644 src/auto-reply/reply/status-updates.ts create mode 100644 src/telegram/send.intermittent.test.ts diff --git a/scripts/run-test-gateway.sh b/scripts/run-test-gateway.sh new file mode 100755 index 000000000..b6f85ff12 --- /dev/null +++ b/scripts/run-test-gateway.sh @@ -0,0 +1,68 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +PORT="${CLAWDBOT_TEST_GATEWAY_PORT:-19001}" +STATE_DIR="${CLAWDBOT_TEST_STATE_DIR:-$HOME/.clawdbot-test}" +CONFIG_PATH="${CLAWDBOT_TEST_CONFIG_PATH:-$STATE_DIR/clawdbot.json}" +MAIN_CONFIG_PATH="${CLAWDBOT_MAIN_CONFIG_PATH:-$HOME/.clawdbot/clawdbot.json}" +TEST_WORKSPACE="${CLAWDBOT_TEST_WORKSPACE:-$HOME/clawd-test}" + +if ! command -v node >/dev/null 2>&1 || ! command -v pnpm >/dev/null 2>&1; then + if command -v nix-shell >/dev/null 2>&1 && [[ -z "${IN_NIX_SHELL:-}" ]]; then + exec nix-shell -p nodejs_22 pnpm --run "$0 $*" + fi + echo "Missing node/pnpm. Install them or run inside nix-shell." >&2 + exit 1 +fi + +if [[ ! -f "$MAIN_CONFIG_PATH" ]]; then + echo "Main config not found at $MAIN_CONFIG_PATH" >&2 + exit 1 +fi + +mkdir -p "$STATE_DIR" "$TEST_WORKSPACE" + +if [[ ! -f "$CONFIG_PATH" ]]; then + python - "$MAIN_CONFIG_PATH" "$CONFIG_PATH" "$PORT" "$TEST_WORKSPACE" <<'PY' +import json +import sys +from pathlib import Path + +main_path = Path(sys.argv[1]) +out_path = Path(sys.argv[2]) +port = int(sys.argv[3]) +workspace = sys.argv[4] + +with main_path.open() as f: + data = json.load(f) + +data.setdefault("gateway", {}) +data["gateway"]["port"] = port +data["gateway"]["bind"] = "loopback" +data["gateway"]["mode"] = "local" + +data.setdefault("agents", {}) +data["agents"].setdefault("defaults", {})["workspace"] = workspace + +channels = data.get("channels") or {} +if "telegram" in channels: + channels["telegram"]["enabled"] = False + data["channels"] = channels + +out_path.parent.mkdir(parents=True, exist_ok=True) +with out_path.open("w") as f: + json.dump(data, f, indent=2) + f.write("\n") +PY +fi + +cd "$ROOT_DIR" + +pnpm install +pnpm build + +export CLAWDBOT_STATE_DIR="$STATE_DIR" +export CLAWDBOT_CONFIG_PATH="$CONFIG_PATH" + +node dist/entry.js gateway --port "$PORT" diff --git a/src/auto-reply/reply/agent-runner-execution.ts b/src/auto-reply/reply/agent-runner-execution.ts index a428aa6da..bda5ee1fa 100644 --- a/src/auto-reply/reply/agent-runner-execution.ts +++ b/src/auto-reply/reply/agent-runner-execution.ts @@ -37,6 +37,13 @@ import type { FollowupRun } from "./queue.js"; import { parseReplyDirectives } from "./reply-directives.js"; import { applyReplyTagsToPayload, isRenderablePayload } from "./reply-payloads.js"; import type { TypingSignaler } from "./typing-mode.js"; +import { + handleAgentEventForStatus, + type StatusUpdateRunContext, +} from "./status-updates-integration.js"; +import { createSubsystemLogger } from "../../logging/subsystem.js"; + +const log = createSubsystemLogger("agent-runner-execution"); export type AgentRunLoopResult = | { @@ -77,6 +84,7 @@ export async function runAgentTurnWithFallback(params: { activeSessionStore?: Record; storePath?: string; resolvedVerboseLevel: VerboseLevel; + statusUpdateContext?: StatusUpdateRunContext; }): Promise { let didLogHeartbeatStrip = false; let autoCompactionCompleted = false; @@ -283,6 +291,13 @@ export async function runAgentTurnWithFallback(params: { } : undefined, onAgentEvent: async (evt) => { + // ===== STATUS UPDATES: Handle agent events ===== + if (params.statusUpdateContext) { + log.debug(`onAgentEvent: forwarding to status updates, stream=${evt.stream}`); + await handleAgentEventForStatus(params.statusUpdateContext, evt); + } + // ===== END STATUS UPDATES ===== + // Trigger typing when tools start executing. // Must await to ensure typing indicator starts before tool summaries are emitted. if (evt.stream === "tool") { diff --git a/src/auto-reply/reply/agent-runner.ts b/src/auto-reply/reply/agent-runner.ts index 227e6f17e..0a1e6d3f1 100644 --- a/src/auto-reply/reply/agent-runner.ts +++ b/src/auto-reply/reply/agent-runner.ts @@ -41,6 +41,17 @@ import { incrementCompactionCount } from "./session-updates.js"; import type { TypingController } from "./typing.js"; import { createTypingSignaler } from "./typing-mode.js"; import { emitDiagnosticEvent, isDiagnosticsEnabled } from "../../infra/diagnostic-events.js"; +import { + createAgentStatusController, + createStatusUpdateRunContext, + completeStatusUpdate, + cleanupStatusUpdate, + handleAgentEventForStatus, + noopStatusCallbacks, +} from "./status-updates-integration.js"; +import { createSubsystemLogger } from "../../logging/subsystem.js"; + +const log = createSubsystemLogger("agent-runner"); const BLOCK_REPLY_SEND_TIMEOUT_MS = 15_000; @@ -112,6 +123,10 @@ export async function runReplyAgent(params: { isHeartbeat, }); + // Ensure we have a runId for event tracking + const runId = opts?.runId ?? crypto.randomUUID(); + const optsWithRunId = opts?.runId ? opts : { ...opts, runId }; + const shouldEmitToolResult = createShouldEmitToolResult({ sessionKey, storePath, @@ -298,13 +313,80 @@ export async function runReplyAgent(params: { `Role ordering conflict (${reason}). Restarting session ${sessionKey} -> ${nextSessionId}.`, cleanupTranscripts: true, }); + + // ===== STATUS UPDATES INTEGRATION ===== + // Create status update controller for this agent run + const agentId = resolveAgentIdFromSessionKey(sessionKey); + + log.info(`Setting up status updates for agentId=${agentId}, channel=${replyToChannel}`); + + // Create callbacks using onBlockReply if available + // NOTE: Edit mode requires channel-specific edit APIs which aren't available + // at the agent-runner level. Use mode="inline" in config for now. + // Telegram/Discord support editing but need integration at dispatch level. + let lastStatusMessageId: string | undefined; + const statusCallbacks = optsWithRunId?.onBlockReply + ? { + sendStatus: async (text: string, _messageId?: string) => { + log.debug(`Sending status update: ${text}`); + // Send as new message via block reply + const res = await optsWithRunId.onBlockReply?.( + { text, isStatusMessage: true }, + { timeoutMs: 3000 }, + ); + if (typeof res === "string") { + lastStatusMessageId = res; + return res; + } + return undefined; + }, + editFinal: async (text: string, messageId: string) => { + log.debug(`Editing final status update: ${messageId} (tracked: ${lastStatusMessageId})`); + await optsWithRunId.onBlockReply?.( + { text, isStatusMessage: true, editMessageId: messageId }, + { timeoutMs: 3000 }, + ); + }, + supportsEdit: () => true, + } + : noopStatusCallbacks; + + const statusController = createAgentStatusController({ + cfg, + agentId, + callbacks: statusCallbacks, + }); + + const statusUpdateContext = createStatusUpdateRunContext(statusController); + log.info( + `Status update context created, controller=${statusController ? "present" : "undefined"}`, + ); + + // Subscribe to agent events for status phase updates + let removeEventListener: (() => void) | undefined; + if (statusUpdateContext.controller) { + const { onAgentEvent } = await import("../../infra/agent-events.js"); + removeEventListener = onAgentEvent((evt) => { + // Only handle events for this run + if (evt.runId === runId) { + handleAgentEventForStatus(statusUpdateContext, { + stream: evt.stream, + data: evt.data, + }).catch((err) => { + log.debug(`Error handling agent event for status: ${String(err)}`); + }); + } + }); + } + // ===== END STATUS UPDATES INTEGRATION ===== + try { const runStartedAt = Date.now(); const runOutcome = await runAgentTurnWithFallback({ commandBody, followupRun, sessionCtx, - opts, + opts: optsWithRunId, typingSignals, blockReplyPipeline, blockStreamingEnabled, @@ -322,6 +404,7 @@ export async function runReplyAgent(params: { activeSessionStore, storePath, resolvedVerboseLevel, + statusUpdateContext, }); if (runOutcome.kind === "final") { @@ -502,6 +585,20 @@ export async function runReplyAgent(params: { finalPayloads = appendUsageLine(finalPayloads, responseUsageLine); } + // ===== STATUS UPDATES: Complete and mark final response ===== + if (statusUpdateContext && replyPayloads.length > 0) { + const lastPayload = replyPayloads[replyPayloads.length - 1]; + if (lastPayload?.text) { + log.info(`[agent-runner] Completing status update with final text (fast-path)`); + const markedText = await completeStatusUpdate(statusUpdateContext, lastPayload.text); + if (markedText) { + // Just update the text. The delivery layer handles replacement. + replyPayloads[replyPayloads.length - 1] = { ...lastPayload, text: markedText }; + } + } + } + // ===== END STATUS UPDATES ===== + return finalizeWithFollowup( finalPayloads.length === 1 ? finalPayloads[0] : finalPayloads, queueKey, @@ -510,5 +607,15 @@ export async function runReplyAgent(params: { } finally { blockReplyPipeline?.stop(); typing.markRunComplete(); + + // ===== STATUS UPDATES: Cleanup ===== + if (removeEventListener) { + removeEventListener(); + } + if (statusUpdateContext) { + log.info(`Cleaning up status update context`); + await cleanupStatusUpdate(statusUpdateContext); + } + // ===== END STATUS UPDATES ===== } } diff --git a/src/auto-reply/reply/block-reply-pipeline.ts b/src/auto-reply/reply/block-reply-pipeline.ts index 00e08acba..de0db832d 100644 --- a/src/auto-reply/reply/block-reply-pipeline.ts +++ b/src/auto-reply/reply/block-reply-pipeline.ts @@ -69,7 +69,7 @@ export function createBlockReplyPipeline(params: { onBlockReply: ( payload: ReplyPayload, options?: { abortSignal?: AbortSignal; timeoutMs?: number }, - ) => Promise | void; + ) => Promise | string | void; timeoutMs: number; coalescing?: BlockStreamingCoalescing; buffer?: BlockReplyBuffer; @@ -102,10 +102,12 @@ export function createBlockReplyPipeline(params: { .then(async () => { if (aborted) return false; await withTimeout( - onBlockReply(payload, { - abortSignal: abortController.signal, - timeoutMs, - }) ?? Promise.resolve(), + Promise.resolve( + onBlockReply(payload, { + abortSignal: abortController.signal, + timeoutMs, + }), + ), timeoutMs, timeoutError, ); diff --git a/src/auto-reply/reply/status-updates-integration.ts b/src/auto-reply/reply/status-updates-integration.ts new file mode 100644 index 000000000..b10c8d9bd --- /dev/null +++ b/src/auto-reply/reply/status-updates-integration.ts @@ -0,0 +1,201 @@ +/** + * Status Updates Integration + * + * Integrates the status update controller with the agent runner + * and provides helpers for resolving configuration from the config. + */ + +import type { ClawdbotConfig } from "../../config/config.js"; +import type { StatusUpdateConfig } from "../../config/types.base.js"; +import type { ChannelCapabilities } from "../../channels/plugins/types.core.js"; +import { + createStatusUpdateController, + type StatusPhase, + type StatusUpdateCallbacks, + type StatusUpdateController, +} from "./status-updates.js"; +import { createSubsystemLogger } from "../../logging/subsystem.js"; + +const log = createSubsystemLogger("status-updates-integration"); + +/** + * Resolve status update configuration from the Clawdbot config. + */ +export function resolveStatusUpdateConfigFromConfig( + cfg: ClawdbotConfig, + agentId?: string, +): StatusUpdateConfig { + log.debug(`Resolving status update config for agentId=${agentId}`); + + // Check agent-specific config first + if (agentId && cfg.agents?.list) { + const agentCfg = cfg.agents.list.find((a) => a.id === agentId); + if (agentCfg?.statusUpdates) { + log.debug(`Found agent-specific config: ${JSON.stringify(agentCfg.statusUpdates)}`); + return agentCfg.statusUpdates; + } + } + + // Fall back to agent defaults + const defaultConfig = cfg.agents?.defaults?.statusUpdates ?? {}; + log.debug(`Using default config: ${JSON.stringify(defaultConfig)}`); + return defaultConfig; +} + +/** + * Check if a channel supports message editing. + */ +export function channelSupportsEdit(capabilities?: ChannelCapabilities): boolean { + return capabilities?.edit === true; +} + +/** + * Create a status update controller for an agent run. + * Returns undefined if status updates are disabled. + */ +export function createAgentStatusController(params: { + cfg: ClawdbotConfig; + agentId?: string; + callbacks: StatusUpdateCallbacks; +}): StatusUpdateController | undefined { + const { cfg, agentId, callbacks } = params; + log.info(`createAgentStatusController called for agentId=${agentId}`); + + const config = resolveStatusUpdateConfigFromConfig(cfg, agentId); + + if (!config.enabled) { + log.info(`Status updates disabled (enabled=${config.enabled})`); + return undefined; + } + + log.info( + `Creating status update controller with mode=${config.mode}, supportsEdit=${callbacks.supportsEdit?.()}`, + ); + const controller = createStatusUpdateController(config, callbacks); + log.info(`Status update controller created successfully`); + return controller; +} + +/** + * Map agent lifecycle events to status phases. + */ +export function mapAgentEventToPhase(event: { + stream: string; + data: Record; +}): StatusPhase | undefined { + const { stream, data } = event; + const phase = typeof data.phase === "string" ? data.phase : ""; + + if (stream === "lifecycle") { + if (phase === "start") return "sending_query"; + if (phase === "end") return "complete"; + } + + if (stream === "tool") { + if (phase === "start") return "processing_tools"; + } + + if (stream === "thinking" || stream === "reasoning") { + return "receiving_reasoning"; + } + + if (stream === "message") { + if (phase === "start") return "generating_response"; + } + + return undefined; +} + +/** + * Context for status updates within an agent run. + */ +export type StatusUpdateRunContext = { + controller?: StatusUpdateController; + startedAt: number; + currentPhase: StatusPhase; +}; + +/** + * Create a run context for status updates. + */ +export function createStatusUpdateRunContext( + controller?: StatusUpdateController, +): StatusUpdateRunContext { + log.info(`createStatusUpdateRunContext: controller=${controller ? "present" : "undefined"}`); + + const ctx = { + controller, + startedAt: Date.now(), + currentPhase: "sending_query" as StatusPhase, + }; + + if (controller) { + log.info(`Status update context created with controller, starting...`); + // Start the controller asynchronously + controller.start().catch((err) => { + log.debug(`Failed to start status controller: ${String(err)}`); + }); + } else { + log.info(`Status update context created WITHOUT controller (status updates disabled)`); + } + + return ctx; +} + +/** + * Handle an agent event and update the status phase if applicable. + */ +export async function handleAgentEventForStatus( + ctx: StatusUpdateRunContext, + event: { stream: string; data: Record }, +): Promise { + if (!ctx.controller) { + log.debug(`handleAgentEventForStatus: no controller, skipping event stream=${event.stream}`); + return; + } + + const phase = mapAgentEventToPhase(event); + log.debug( + `handleAgentEventForStatus: stream=${event.stream}, phase=${event.data.phase}, mappedPhase=${phase}, currentPhase=${ctx.currentPhase}`, + ); + + if (phase && phase !== ctx.currentPhase) { + log.info(`Status phase change: ${ctx.currentPhase} -> ${phase}`); + ctx.currentPhase = phase; + await ctx.controller.setPhase(phase); + } +} + +/** + * Mark the status as complete and optionally add checkmark to final text. + */ +export async function completeStatusUpdate( + ctx: StatusUpdateRunContext, + finalText?: string, +): Promise { + if (!ctx.controller) { + log.debug(`completeStatusUpdate: no controller, returning original text`); + return finalText; + } + + log.info(`completeStatusUpdate: finalizing status update`); + const result = await ctx.controller.complete(finalText); + log.debug(`completeStatusUpdate: result=${result?.substring(0, 50)}...`); + return result; +} + +/** + * Cleanup the status update controller. + */ +export async function cleanupStatusUpdate(ctx: StatusUpdateRunContext): Promise { + if (!ctx.controller) return; + await ctx.controller.cleanup(); +} + +/** + * Export a no-op callbacks object for channels that don't support status updates. + */ +export const noopStatusCallbacks: StatusUpdateCallbacks = { + sendStatus: async () => undefined, + supportsEdit: () => false, +}; diff --git a/src/auto-reply/reply/status-updates.ts b/src/auto-reply/reply/status-updates.ts new file mode 100644 index 000000000..25a829d79 --- /dev/null +++ b/src/auto-reply/reply/status-updates.ts @@ -0,0 +1,346 @@ +/** + * Status Updates Module + * + * Provides intermittent status messages during bot activity to keep users informed + * about what's happening while the AI processes their request. + * + * Features: + * - Configurable status message phases + * - Elapsed time tracking + * - Final response checkmark marking + * - Platform-specific message update handling + */ + +import { createSubsystemLogger } from "../../logging/subsystem.js"; + +const log = createSubsystemLogger("status-updates"); + +export type StatusPhase = + | "sending_query" + | "receiving_reasoning" + | "processing_tools" + | "generating_response" + | "complete"; + +export type StatusUpdateMode = "off" | "edit" | "inline"; + +export type StatusUpdateConfig = { + /** Enable status updates (default: false). */ + enabled?: boolean; + /** Mode for status updates: "off", "edit" (update same message), "inline" (new messages). */ + mode?: StatusUpdateMode; + /** Interval in milliseconds between elapsed time updates (default: 5000). */ + updateIntervalMs?: number; + /** Show phase descriptions like "Sending query..." (default: true). */ + showPhases?: boolean; + /** Show elapsed time in status messages (default: true). */ + showElapsedTime?: boolean; + /** Mark final responses with βœ… (default: true). */ + markFinalWithCheckmark?: boolean; + /** Minimum duration before showing elapsed time (ms, default: 3000). */ + elapsedTimeThresholdMs?: number; +}; + +export type StatusUpdateCallbacks = { + /** Send or update status message. Returns message ID if applicable. */ + sendStatus: (text: string, messageId?: string) => Promise; + /** Edit an existing message with final content and checkmark. */ + editFinal?: (text: string, messageId: string) => Promise; + /** Check if the channel supports message editing. */ + supportsEdit?: () => boolean; + /** Delete a status message (for cleanup). */ + deleteStatus?: (messageId: string) => Promise; +}; + +export type StatusUpdateState = { + phase: StatusPhase; + startedAt: number; + statusMessageId?: string; + lastUpdateAt: number; + elapsedSeconds: number; + isComplete: boolean; +}; + +const DEFAULT_CONFIG: Required = { + enabled: false, + mode: "edit", + updateIntervalMs: 5000, + showPhases: true, + showElapsedTime: true, + markFinalWithCheckmark: true, + elapsedTimeThresholdMs: 3000, +}; + +const PHASE_MESSAGES: Record = { + sending_query: "Sending query to AI model", + receiving_reasoning: "Processing reasoning data", + processing_tools: "Executing tools", + generating_response: "Generating response", + complete: "Complete", +}; + +const PHASE_EMOJI: Record = { + sending_query: "⏳", + receiving_reasoning: "🧠", + processing_tools: "πŸ”§", + generating_response: "✍️", + complete: "βœ…", +}; + +export function resolveStatusUpdateConfig( + config?: Partial, +): Required { + return { + ...DEFAULT_CONFIG, + ...config, + }; +} + +function formatElapsedTime(seconds: number): string { + const minutes = Math.floor(seconds / 60); + const remainingSeconds = seconds % 60; + const paddedSeconds = remainingSeconds.toString().padStart(2, "0"); + return `${minutes}:${paddedSeconds}`; +} + +function formatStatusMessage( + phase: StatusPhase, + elapsedSeconds: number, + config: Required, +): string { + const parts: string[] = []; + + if (config.showPhases) { + const emoji = PHASE_EMOJI[phase]; + const message = PHASE_MESSAGES[phase]; + parts.push(`${emoji} ${message}`); + } + + if (config.showElapsedTime && elapsedSeconds * 1000 >= config.elapsedTimeThresholdMs) { + const elapsed = formatElapsedTime(elapsedSeconds); + parts.push(`(${elapsed})`); + } + + return parts.join(" ") || "Processing..."; +} + +export class StatusUpdateController { + private config: Required; + private callbacks: StatusUpdateCallbacks; + private state: StatusUpdateState; + private updateTimer: NodeJS.Timeout | null = null; + private stopped = false; + private editedInPlace: boolean = false; + + constructor(config: StatusUpdateConfig, callbacks: StatusUpdateCallbacks) { + this.config = resolveStatusUpdateConfig(config); + this.callbacks = callbacks; + this.state = { + phase: "sending_query", + startedAt: Date.now(), + lastUpdateAt: Date.now(), + elapsedSeconds: 0, + isComplete: false, + }; + } + + isEnabled(): boolean { + return this.config.enabled && this.config.mode !== "off"; + } + + wasEditedInPlace(): boolean { + return this.editedInPlace; + } + + getStatusMessageId(): string | undefined { + return this.state.statusMessageId; + } + + supportsEdit(): boolean { + return this.callbacks.supportsEdit?.() ?? false; + } + + async start(): Promise { + if (!this.isEnabled() || this.stopped) return; + + log.debug("Starting status updates"); + + // Send initial status message + await this.sendUpdate(); + + // Start update timer if showing elapsed time + if (this.config.showElapsedTime) { + this.startUpdateTimer(); + } + } + + async setPhase(phase: StatusPhase): Promise { + if (!this.isEnabled() || this.stopped || this.state.isComplete) return; + + if (phase === this.state.phase) return; + + log.debug(`Status phase: ${this.state.phase} -> ${phase}`); + this.state.phase = phase; + + if (phase === "complete") { + this.state.isComplete = true; + this.stopUpdateTimer(); + return; // Suppress redundant sendUpdate for "complete" phase + } + + await this.sendUpdate(); + } + + async complete(finalText?: string): Promise { + if (this.stopped) return undefined; + + this.state.isComplete = true; + this.stopUpdateTimer(); + this.stopped = true; + + if (!this.isEnabled()) return finalText; + + // If we have a final text and config says to mark with checkmark + if (finalText && this.config.markFinalWithCheckmark) { + this.state.elapsedSeconds = Math.floor((Date.now() - this.state.startedAt) / 1000); + // NOTE: We rely on Telegram delivery to do the edit/replacement, so we just format + // the string correctly for the final response. + const marked = `${finalText.trimEnd()} _(${formatElapsedTime(this.state.elapsedSeconds)})_`; + + return marked; + } + + // Clean up status message if it exists and we're not editing + if (this.state.statusMessageId && this.callbacks.deleteStatus) { + try { + await this.callbacks.deleteStatus(this.state.statusMessageId); + } catch (err) { + log.debug(`Failed to delete status message: ${String(err)}`); + } + } + + return finalText; + } + + async cleanup(): Promise { + this.stopUpdateTimer(); + this.stopped = true; + + // Delete status message if it exists and response is being sent separately + if (this.state.statusMessageId && !this.state.isComplete && this.callbacks.deleteStatus) { + try { + await this.callbacks.deleteStatus(this.state.statusMessageId); + } catch (err) { + log.debug(`Failed to cleanup status message: ${String(err)}`); + } + } + } + + private async sendUpdate(): Promise { + if (this.stopped) return; + + const now = Date.now(); + this.state.elapsedSeconds = Math.floor((now - this.state.startedAt) / 1000); + this.state.lastUpdateAt = now; + + const text = formatStatusMessage(this.state.phase, this.state.elapsedSeconds, this.config); + + try { + if (this.config.mode === "edit" && this.supportsEdit() && this.state.statusMessageId) { + // Edit existing message + this.state.statusMessageId = await this.callbacks.sendStatus( + text, + this.state.statusMessageId, + ); + } else { + // Send new message (or first message in edit mode) + const messageId = await this.callbacks.sendStatus(text, this.state.statusMessageId); + if (messageId) { + this.state.statusMessageId = messageId; + } + } + } catch (err) { + log.debug(`Failed to send status update: ${String(err)}`); + } + } + + private startUpdateTimer(): void { + if (this.updateTimer) return; + + this.updateTimer = setInterval(async () => { + if (this.stopped || this.state.isComplete) { + this.stopUpdateTimer(); + return; + } + await this.sendUpdate(); + }, this.config.updateIntervalMs); + } + + private stopUpdateTimer(): void { + if (this.updateTimer) { + clearInterval(this.updateTimer); + this.updateTimer = null; + } + } +} + +export function createStatusUpdateController( + config: StatusUpdateConfig, + callbacks: StatusUpdateCallbacks, +): StatusUpdateController { + return new StatusUpdateController(config, callbacks); +} + +/** + * Mark a text response with the completion checkmark if enabled. + */ +export function markResponseComplete( + text: string | undefined, + config?: StatusUpdateConfig, +): string | undefined { + if (!text) return text; + + const resolved = resolveStatusUpdateConfig(config); + if (!resolved.enabled || !resolved.markFinalWithCheckmark) return text; + + // Don't double-mark + if (text.trimEnd().endsWith("βœ…")) return text; + + return `${text.trimEnd()} βœ…`; +} + +/** + * Create status update callbacks for channels that support message editing. + */ +export function createEditableStatusCallbacks(params: { + sendMessage: (text: string) => Promise; + editMessage: (text: string, messageId: string) => Promise; + deleteMessage?: (messageId: string) => Promise; +}): StatusUpdateCallbacks { + return { + sendStatus: async (text, messageId) => { + if (messageId) { + await params.editMessage(text, messageId); + return messageId; + } + return params.sendMessage(text); + }, + editFinal: params.editMessage, + supportsEdit: () => true, + deleteStatus: params.deleteMessage, + }; +} + +/** + * Create status update callbacks for channels that don't support editing. + */ +export function createInlineStatusCallbacks(params: { + sendMessage: (text: string) => Promise; +}): StatusUpdateCallbacks { + return { + sendStatus: async (text) => { + return params.sendMessage(text); + }, + supportsEdit: () => false, + }; +} diff --git a/src/auto-reply/types.ts b/src/auto-reply/types.ts index 250c14091..7e250a3f8 100644 --- a/src/auto-reply/types.ts +++ b/src/auto-reply/types.ts @@ -27,7 +27,10 @@ export type GetReplyOptions = { isHeartbeat?: boolean; onPartialReply?: (payload: ReplyPayload) => Promise | void; onReasoningStream?: (payload: ReplyPayload) => Promise | void; - onBlockReply?: (payload: ReplyPayload, context?: BlockReplyContext) => Promise | void; + onBlockReply?: ( + payload: ReplyPayload, + context?: BlockReplyContext, + ) => Promise | string | void; onToolResult?: (payload: ReplyPayload) => Promise | void; /** Called when the actual model is selected (including after fallback). * Use this to get model/provider/thinkLevel for responsePrefix template interpolation. */ @@ -51,5 +54,9 @@ export type ReplyPayload = { replyToCurrent?: boolean; /** Send audio as voice message (bubble) instead of audio file. Defaults to false. */ audioAsVoice?: boolean; + isStatusMessage?: boolean; isError?: boolean; + editMessageId?: string; + /** Whether this payload should replace the last status message (provider-specific logic). */ + replaceStatus?: boolean; }; diff --git a/src/config/schema.ts b/src/config/schema.ts index d61b5964e..a529829f4 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -262,6 +262,14 @@ const FIELD_LABELS: Record = { "agents.defaults.humanDelay.mode": "Human Delay Mode", "agents.defaults.humanDelay.minMs": "Human Delay Min (ms)", "agents.defaults.humanDelay.maxMs": "Human Delay Max (ms)", + "agents.defaults.statusUpdates": "Status Updates", + "agents.defaults.statusUpdates.enabled": "Enable Status Updates", + "agents.defaults.statusUpdates.mode": "Status Update Mode", + "agents.defaults.statusUpdates.updateIntervalMs": "Status Update Interval (ms)", + "agents.defaults.statusUpdates.showPhases": "Show Processing Phases", + "agents.defaults.statusUpdates.showElapsedTime": "Show Elapsed Time", + "agents.defaults.statusUpdates.markFinalWithCheckmark": "Mark Final with βœ…", + "agents.defaults.statusUpdates.elapsedTimeThresholdMs": "Elapsed Time Threshold (ms)", "agents.defaults.cliBackends": "CLI Backends", "commands.native": "Native Commands", "commands.nativeSkills": "Native Skill Commands", diff --git a/src/config/types.agent-defaults.ts b/src/config/types.agent-defaults.ts index 2a42d3623..1c736e814 100644 --- a/src/config/types.agent-defaults.ts +++ b/src/config/types.agent-defaults.ts @@ -2,6 +2,7 @@ import type { BlockStreamingChunkConfig, BlockStreamingCoalesceConfig, HumanDelayConfig, + StatusUpdateConfig, TypingMode, } from "./types.base.js"; import type { ChannelId } from "../channels/plugins/types.js"; @@ -161,6 +162,11 @@ export type AgentDefaultsConfig = { typingIntervalSeconds?: number; /** Typing indicator start mode (never|instant|thinking|message). */ typingMode?: TypingMode; + /** + * Intermittent status message configuration. + * Sends status updates during AI processing to keep users informed. + */ + statusUpdates?: StatusUpdateConfig; /** Periodic background heartbeat runs. */ heartbeat?: { /** Heartbeat interval (duration string, default unit: minutes; default: 30m). */ diff --git a/src/config/types.agents.ts b/src/config/types.agents.ts index f083c1897..b2c05a96f 100644 --- a/src/config/types.agents.ts +++ b/src/config/types.agents.ts @@ -1,5 +1,5 @@ import type { AgentDefaultsConfig } from "./types.agent-defaults.js"; -import type { HumanDelayConfig, IdentityConfig } from "./types.base.js"; +import type { HumanDelayConfig, IdentityConfig, StatusUpdateConfig } from "./types.base.js"; import type { GroupChatConfig } from "./types.messages.js"; import type { SandboxBrowserSettings, @@ -29,6 +29,8 @@ export type AgentConfig = { humanDelay?: HumanDelayConfig; /** Optional per-agent heartbeat overrides. */ heartbeat?: AgentDefaultsConfig["heartbeat"]; + /** Per-agent status update configuration. */ + statusUpdates?: StatusUpdateConfig; identity?: IdentityConfig; groupChat?: GroupChatConfig; subagents?: { diff --git a/src/config/types.base.ts b/src/config/types.base.ts index a84736571..08cf78689 100644 --- a/src/config/types.base.ts +++ b/src/config/types.base.ts @@ -164,3 +164,28 @@ export type IdentityConfig = { /** Avatar image: workspace-relative path, http(s) URL, or data URI. */ avatar?: string; }; + +/** + * Status Update Configuration + * + * Configures intermittent status messages sent during AI processing + * to keep users informed about what's happening. + */ +export type StatusUpdateMode = "off" | "edit" | "inline"; + +export type StatusUpdateConfig = { + /** Enable status updates (default: false). */ + enabled?: boolean; + /** Mode for status updates: "off", "edit" (update same message), "inline" (new messages). */ + mode?: StatusUpdateMode; + /** Interval in milliseconds between elapsed time updates (default: 5000). */ + updateIntervalMs?: number; + /** Show phase descriptions like "Sending query..." (default: true). */ + showPhases?: boolean; + /** Show elapsed time in status messages (default: true). */ + showElapsedTime?: boolean; + /** Mark final responses with βœ… (default: true). */ + markFinalWithCheckmark?: boolean; + /** Minimum duration before showing elapsed time (ms, default: 3000). */ + elapsedTimeThresholdMs?: number; +}; diff --git a/src/config/zod-schema.agent-defaults.ts b/src/config/zod-schema.agent-defaults.ts index c4b8a8f2c..300fcd1d3 100644 --- a/src/config/zod-schema.agent-defaults.ts +++ b/src/config/zod-schema.agent-defaults.ts @@ -11,6 +11,7 @@ import { BlockStreamingCoalesceSchema, CliBackendSchema, HumanDelaySchema, + StatusUpdateSchema, } from "./zod-schema.core.js"; export const AgentDefaultsSchema = z @@ -132,6 +133,8 @@ export const AgentDefaultsSchema = z z.literal("message"), ]) .optional(), + /** Configure intermittent status messages during AI processing. */ + statusUpdates: StatusUpdateSchema, heartbeat: HeartbeatSchema, maxConcurrent: z.number().int().positive().optional(), subagents: z diff --git a/src/config/zod-schema.agent-runtime.ts b/src/config/zod-schema.agent-runtime.ts index c733dcfa9..2a4dab3d4 100644 --- a/src/config/zod-schema.agent-runtime.ts +++ b/src/config/zod-schema.agent-runtime.ts @@ -5,6 +5,7 @@ import { GroupChatSchema, HumanDelaySchema, IdentitySchema, + StatusUpdateSchema, ToolsLinksSchema, ToolsMediaSchema, } from "./zod-schema.core.js"; @@ -397,6 +398,8 @@ export const AgentEntrySchema = z memorySearch: MemorySearchSchema, humanDelay: HumanDelaySchema.optional(), heartbeat: HeartbeatSchema, + /** Per-agent status update configuration. */ + statusUpdates: StatusUpdateSchema, identity: IdentitySchema, groupChat: GroupChatSchema, subagents: z diff --git a/src/config/zod-schema.core.ts b/src/config/zod-schema.core.ts index 4a8c80bcc..897d146cc 100644 --- a/src/config/zod-schema.core.ts +++ b/src/config/zod-schema.core.ts @@ -500,3 +500,35 @@ export const ProviderCommandsSchema = z }) .strict() .optional(); + +/** + * Status Updates Configuration Schema + * + * Configures intermittent status messages sent during AI processing + * to keep users informed about what's happening. + */ +export const StatusUpdateModeSchema = z.union([ + z.literal("off"), + z.literal("edit"), + z.literal("inline"), +]); + +export const StatusUpdateSchema = z + .object({ + /** Enable status updates (default: false). */ + enabled: z.boolean().optional(), + /** Mode for status updates: "off", "edit" (update same message), "inline" (new messages). */ + mode: StatusUpdateModeSchema.optional(), + /** Interval in milliseconds between elapsed time updates (default: 5000). */ + updateIntervalMs: z.number().int().positive().optional(), + /** Show phase descriptions like "Sending query..." (default: true). */ + showPhases: z.boolean().optional(), + /** Show elapsed time in status messages (default: true). */ + showElapsedTime: z.boolean().optional(), + /** Mark final responses with βœ… (default: true). */ + markFinalWithCheckmark: z.boolean().optional(), + /** Minimum duration before showing elapsed time (ms, default: 3000). */ + elapsedTimeThresholdMs: z.number().int().nonnegative().optional(), + }) + .strict() + .optional(); diff --git a/src/telegram/bot-handlers.ts b/src/telegram/bot-handlers.ts index 8dfcc5ac1..48916c541 100644 --- a/src/telegram/bot-handlers.ts +++ b/src/telegram/bot-handlers.ts @@ -12,6 +12,7 @@ import { resolveTelegramForumThreadId } from "./bot/helpers.js"; import type { TelegramMessage } from "./bot/types.js"; import { firstDefined, isSenderAllowed, normalizeAllowFromWithStore } from "./bot-access.js"; import { MEDIA_GROUP_TIMEOUT_MS, type MediaGroupEntry } from "./bot-updates.js"; +import { invalidateLastStatus } from "./sent-message-cache.js"; import { migrateTelegramGroupConfig } from "./group-migration.js"; import { resolveTelegramInlineButtonsScope } from "./inline-buttons.js"; import { readTelegramAllowFromStore } from "./pairing-store.js"; @@ -383,6 +384,10 @@ export const registerTelegramHandlers = ({ if (shouldSkipUpdate(ctx)) return; const chatId = msg.chat.id; + // Invalidate status message state on ANY inbound message. + // This ensures we never edit a previous turn's message. + invalidateLastStatus(chatId); + const isGroup = msg.chat.type === "group" || msg.chat.type === "supergroup"; const messageThreadId = (msg as { message_thread_id?: number }).message_thread_id; const isForum = (msg.chat as { is_forum?: boolean }).is_forum === true; @@ -477,6 +482,7 @@ export const registerTelegramHandlers = ({ // Text fragment handling - Telegram splits long pastes into multiple inbound messages (~4096 chars). // We buffer β€œnear-limit” messages and append immediately-following parts. const text = typeof msg.text === "string" ? msg.text : undefined; + const isCommandLike = (text ?? "").trim().startsWith("/"); if (text && !isCommandLike) { const nowMs = Date.now(); diff --git a/src/telegram/bot-message-dispatch.ts b/src/telegram/bot-message-dispatch.ts index 334c4c212..1bc345ed6 100644 --- a/src/telegram/bot-message-dispatch.ts +++ b/src/telegram/bot-message-dispatch.ts @@ -12,6 +12,7 @@ import { resolveMarkdownTableMode } from "../config/markdown-tables.js"; import { deliverReplies } from "./bot/delivery.js"; import { resolveTelegramDraftStreamingChunking } from "./draft-chunking.js"; import { createTelegramDraftStream } from "./draft-stream.js"; +import { editMessageTelegram } from "./send.js"; export const dispatchTelegramMessage = async ({ context, @@ -143,6 +144,7 @@ export const dispatchTelegramMessage = async ({ replies: [payload], chatId: String(chatId), token: opts.token, + accountId: route.accountId, runtime, bot, replyToMode, diff --git a/src/telegram/bot-native-commands.ts b/src/telegram/bot-native-commands.ts index 0f1cc1cb7..410448b3d 100644 --- a/src/telegram/bot-native-commands.ts +++ b/src/telegram/bot-native-commands.ts @@ -341,6 +341,7 @@ export const registerTelegramNativeCommands = ({ replies: [payload], chatId: String(chatId), token: opts.token, + accountId, runtime, bot, replyToMode, diff --git a/src/telegram/bot/delivery.test.ts b/src/telegram/bot/delivery.test.ts index ca1b3f4cd..c110723c1 100644 --- a/src/telegram/bot/delivery.test.ts +++ b/src/telegram/bot/delivery.test.ts @@ -32,6 +32,7 @@ describe("deliverReplies", () => { replies: [{ audioAsVoice: true }], chatId: "123", token: "tok", + accountId: "test", runtime, bot, replyToMode: "off", @@ -63,6 +64,7 @@ describe("deliverReplies", () => { replies: [{ mediaUrl: "https://example.com/note.ogg", audioAsVoice: true }], chatId: "123", token: "tok", + accountId: "test", runtime, bot, replyToMode: "off", @@ -93,6 +95,7 @@ describe("deliverReplies", () => { replies: [{ mediaUrl: "https://example.com/photo.jpg", text: "hi **boss**" }], chatId: "123", token: "tok", + accountId: "test", runtime, bot, replyToMode: "off", @@ -121,6 +124,7 @@ describe("deliverReplies", () => { replies: [{ text: "Check https://example.com" }], chatId: "123", token: "tok", + accountId: "test", runtime, bot, replyToMode: "off", @@ -149,6 +153,7 @@ describe("deliverReplies", () => { replies: [{ text: "Check https://example.com" }], chatId: "123", token: "tok", + accountId: "test", runtime, bot, replyToMode: "off", diff --git a/src/telegram/bot/delivery.ts b/src/telegram/bot/delivery.ts index 2d117d748..18b9dc520 100644 --- a/src/telegram/bot/delivery.ts +++ b/src/telegram/bot/delivery.ts @@ -1,3 +1,4 @@ +import { sendMessageTelegram, editMessageTelegram } from "../send.js"; import { type Bot, InputFile } from "grammy"; import { markdownToTelegramChunks, @@ -6,6 +7,7 @@ import { } from "../format.js"; import { chunkMarkdownTextWithMode, type ChunkMode } from "../../auto-reply/chunk.js"; import { splitTelegramCaption } from "../caption.js"; +import { getLastSentMessage, recordSentMessage } from "../sent-message-cache.js"; import type { ReplyPayload } from "../../auto-reply/types.js"; import type { ReplyToMode } from "../../config/config.js"; import type { MarkdownTableMode } from "../../config/types.base.js"; @@ -27,6 +29,7 @@ export async function deliverReplies(params: { replies: ReplyPayload[]; chatId: string; token: string; + accountId?: string; runtime: RuntimeEnv; bot: Bot; replyToMode: ReplyToMode; @@ -80,20 +83,71 @@ export async function deliverReplies(params: { ? [reply.mediaUrl] : []; if (mediaList.length === 0) { - const chunks = chunkText(reply.text || ""); - for (const chunk of chunks) { - await sendTelegramText(bot, chatId, chunk.html, runtime, { - replyToMessageId: - replyToId && (replyToMode === "all" || !hasReplied) ? replyToId : undefined, - messageThreadId, - textMode: "html", - plainText: chunk.text, - linkPreview, - }); - if (replyToId && !hasReplied) { - hasReplied = true; + // Check if the previous message was a status message. + // If so, we should edit it instead of sending a new message. + const lastSent = getLastSentMessage(chatId); + const shouldEditLast = + lastSent?.isStatus && lastSent.messageId && Date.now() - lastSent.timestamp < 120000; // 2 min threshold + + let sentMessageId: string | undefined; + + if (shouldEditLast && reply.text) { + const editId = String(lastSent!.messageId); + console.error( + `[CRITICAL LOG] [telegram/bot/delivery.ts] Editing previous STATUS message ${editId} with: ${JSON.stringify(reply.text)}`, + ); + try { + await editMessageTelegram(chatId, editId, reply.text, { + token: params.token, + accountId: params.accountId, + api: bot.api, + }); + sentMessageId = editId; + } catch (err) { + console.error( + `[CRITICAL LOG] [telegram/bot/delivery.ts] Edit failed, falling back to new message: ${String(err)}`, + ); + // Fall through to send new message } } + + if (!sentMessageId && reply.isStatusMessage && reply.text) { + console.error( + `[CRITICAL LOG] [telegram/bot/delivery.ts] Sending NEW status message: ${JSON.stringify(reply.text)}`, + ); + const result = await sendMessageTelegram(chatId, reply.text, { + token: params.token, + accountId: params.accountId, + api: bot.api, + isStatusMessage: true, + messageThreadId, + }); + sentMessageId = result.messageId; + } else if (!sentMessageId) { + const chunks = chunkText(reply.text || ""); + for (const chunk of chunks) { + console.error( + `[CRITICAL LOG] [telegram/bot/delivery.ts] Sending text chunk: ${JSON.stringify(chunk.text)}`, + ); + const msgId = await sendTelegramText(bot, chatId, chunk.html, runtime, { + replyToMessageId: + replyToId && (replyToMode === "all" || !hasReplied) ? replyToId : undefined, + messageThreadId, + textMode: "html", + plainText: chunk.text, + linkPreview, + }); + if (msgId) sentMessageId = String(msgId); + if (replyToId && !hasReplied) { + hasReplied = true; + } + } + } + + // Record the message state for next time + if (sentMessageId) { + recordSentMessage(chatId, Number(sentMessageId), !!reply.isStatusMessage); + } continue; } // media with optional caption on first item diff --git a/src/telegram/send.intermittent.test.ts b/src/telegram/send.intermittent.test.ts new file mode 100644 index 000000000..2d9ae3bd2 --- /dev/null +++ b/src/telegram/send.intermittent.test.ts @@ -0,0 +1,243 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const { botApi, botCtorSpy } = vi.hoisted(() => ({ + botApi: { + sendMessage: vi.fn(), + editMessageText: vi.fn(), + }, + botCtorSpy: vi.fn(), +})); + +vi.mock("grammy", () => ({ + Bot: class { + api = botApi; + constructor( + public token: string, + public options?: { + client?: { fetch?: typeof fetch; timeoutSeconds?: number }; + }, + ) { + botCtorSpy(token, options); + } + }, + InputFile: class {}, +})); + +const { loadConfig } = vi.hoisted(() => ({ + loadConfig: vi.fn(() => ({})), +})); +vi.mock("../config/config.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + loadConfig, + }; +}); + +const { getLastSentMessage, recordSentMessage, getLastStatusMessage, recordStatusMessage } = + vi.hoisted(() => ({ + getLastSentMessage: vi.fn(), + recordSentMessage: vi.fn(), + getLastStatusMessage: vi.fn(), + recordStatusMessage: vi.fn(), + })); + +vi.mock("./sent-message-cache.js", () => ({ + getLastSentMessage, + recordSentMessage, + getLastStatusMessage, + recordStatusMessage, +})); + +import { sendMessageTelegram } from "./send.js"; + +describe("sendMessageTelegram status message editing", () => { + beforeEach(() => { + loadConfig.mockReturnValue({}); + botApi.sendMessage.mockReset(); + botApi.editMessageText.mockReset(); + getLastSentMessage.mockReset(); + recordSentMessage.mockReset(); + getLastStatusMessage.mockReset(); + recordStatusMessage.mockReset(); + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("sends new message if isStatusMessage is false (reversion check)", async () => { + const chatId = "123"; + const text = "Hello"; + + botApi.sendMessage.mockResolvedValue({ + message_id: 100, + chat: { id: chatId }, + }); + // Even if we have a recent status message, it should be ignored if isStatusMessage is false + const now = 100000; + vi.setSystemTime(now); + getLastStatusMessage.mockReturnValue({ + messageId: 99, + timestamp: now - 1000, + }); + + await sendMessageTelegram(chatId, text, { token: "tok" }); + + expect(botApi.sendMessage).toHaveBeenCalledWith(chatId, text, expect.anything()); + expect(botApi.editMessageText).not.toHaveBeenCalled(); + expect(recordSentMessage).toHaveBeenCalledWith(chatId, 100); + expect(recordStatusMessage).not.toHaveBeenCalled(); + }); + + it("sends new message if no previous status message", async () => { + const chatId = "123"; + const text = "Status update"; + + botApi.sendMessage.mockResolvedValue({ + message_id: 100, + chat: { id: chatId }, + }); + getLastStatusMessage.mockReturnValue(undefined); + + await sendMessageTelegram(chatId, text, { token: "tok", isStatusMessage: true }); + + expect(botApi.sendMessage).toHaveBeenCalledWith(chatId, text, expect.anything()); + expect(botApi.editMessageText).not.toHaveBeenCalled(); + expect(recordStatusMessage).toHaveBeenCalledWith(chatId, 100); + }); + + it("edits previous status message if sent < 10s ago", async () => { + const chatId = "123"; + const text = "Updated status"; + const lastMsgId = 100; + + // Setup last sent message 5s ago + const now = 100000; + vi.setSystemTime(now); + getLastStatusMessage.mockReturnValue({ + messageId: lastMsgId, + timestamp: now - 5000, + }); + + botApi.editMessageText.mockResolvedValue(true); + + const res = await sendMessageTelegram(chatId, text, { token: "tok", isStatusMessage: true }); + + expect(botApi.editMessageText).toHaveBeenCalledWith( + chatId, + lastMsgId, + text, + expect.objectContaining({ parse_mode: "HTML" }), + ); + expect(botApi.sendMessage).not.toHaveBeenCalled(); + expect(res.messageId).toBe("100"); + // Should update timestamp for same messageId + expect(recordStatusMessage).toHaveBeenCalledWith(chatId, lastMsgId); + }); + + it("sends new message if status sent > 10s ago", async () => { + const chatId = "123"; + const text = "New status"; + const lastMsgId = 100; + + // Setup last sent message 11s ago + const now = 100000; + vi.setSystemTime(now); + getLastStatusMessage.mockReturnValue({ + messageId: lastMsgId, + timestamp: now - 11000, + }); + + botApi.sendMessage.mockResolvedValue({ + message_id: 101, + chat: { id: chatId }, + }); + + await sendMessageTelegram(chatId, text, { token: "tok", isStatusMessage: true }); + + expect(botApi.sendMessage).toHaveBeenCalled(); + expect(botApi.editMessageText).not.toHaveBeenCalled(); + expect(recordStatusMessage).toHaveBeenCalledWith(chatId, 101); + }); + + it("falls back to send if edit fails", async () => { + const chatId = "123"; + const text = "Updated status"; + const lastMsgId = 100; + + const now = 100000; + vi.setSystemTime(now); + getLastStatusMessage.mockReturnValue({ + messageId: lastMsgId, + timestamp: now - 5000, + }); + + botApi.editMessageText.mockRejectedValue(new Error("Message not found")); + botApi.sendMessage.mockResolvedValue({ + message_id: 102, + chat: { id: chatId }, + }); + + await sendMessageTelegram(chatId, text, { token: "tok", isStatusMessage: true }); + + expect(botApi.editMessageText).toHaveBeenCalled(); + expect(botApi.sendMessage).toHaveBeenCalled(); // Fallback + expect(recordStatusMessage).toHaveBeenCalledWith(chatId, 102); + }); + + it("treats 'message is not modified' error as success", async () => { + const chatId = "123"; + const text = "Same status"; + const lastMsgId = 100; + + const now = 100000; + vi.setSystemTime(now); + getLastStatusMessage.mockReturnValue({ + messageId: lastMsgId, + timestamp: now - 5000, + }); + + botApi.editMessageText.mockRejectedValue( + new Error( + "Bad Request: message is not modified: specified new message content and reply markup are exactly the same as a current content and reply markup of the message", + ), + ); + + const res = await sendMessageTelegram(chatId, text, { token: "tok", isStatusMessage: true }); + + expect(botApi.editMessageText).toHaveBeenCalled(); + expect(botApi.sendMessage).not.toHaveBeenCalled(); + expect(res.messageId).toBe("100"); + // Should still update timestamp (treated as success) + expect(recordStatusMessage).toHaveBeenCalledWith(chatId, lastMsgId); + }); + + it("includes buttons when editing status", async () => { + const chatId = "123"; + const text = "Status with buttons"; + const lastMsgId = 100; + const buttons = [[{ text: "Btn", callback_data: "data" }]]; + + const now = 100000; + vi.setSystemTime(now); + getLastStatusMessage.mockReturnValue({ + messageId: lastMsgId, + timestamp: now - 5000, + }); + + botApi.editMessageText.mockResolvedValue(true); + + await sendMessageTelegram(chatId, text, { token: "tok", buttons, isStatusMessage: true }); + + expect(botApi.editMessageText).toHaveBeenCalledWith( + chatId, + lastMsgId, + text, + expect.objectContaining({ + reply_markup: { inline_keyboard: buttons }, + }), + ); + }); +}); diff --git a/src/telegram/send.ts b/src/telegram/send.ts index 869e990c6..dac49edf3 100644 --- a/src/telegram/send.ts +++ b/src/telegram/send.ts @@ -42,6 +42,7 @@ type TelegramSendOpts = { messageThreadId?: number; /** Inline keyboard buttons (reply markup). */ buttons?: Array>; + isStatusMessage?: boolean; }; type TelegramSendResult = { @@ -202,6 +203,10 @@ export async function sendMessageTelegram( const linkPreviewEnabled = account.config.linkPreview ?? true; const linkPreviewOptions = linkPreviewEnabled ? undefined : { is_disabled: true }; + // Status message optimization: + // Moved to delivery.ts to centralize logic. + // We no longer attempt to optimize here to avoid race conditions and split-brain logic. + const sendTelegramText = async ( rawText: string, params?: Record, @@ -241,6 +246,11 @@ export async function sendMessageTelegram( throw wrapChatNotFound(err); }, ); + if (res?.message_id) { + console.error( + `[CRITICAL LOG] [telegram/send.ts] Sent message ${res.message_id} to ${chatId}. Payload: ${JSON.stringify({ text: rawText, params: sendParams })}`, + ); + } return res; }; @@ -324,7 +334,11 @@ export async function sendMessageTelegram( const mediaMessageId = String(result?.message_id ?? "unknown"); const resolvedChatId = String(result?.chat?.id ?? chatId); if (result?.message_id) { - recordSentMessage(chatId, result.message_id); + if (opts.isStatusMessage) { + recordSentMessage(chatId, result.message_id, true); + } else { + recordSentMessage(chatId, result.message_id, false); + } } recordChannelActivity({ channel: "telegram", @@ -366,7 +380,11 @@ export async function sendMessageTelegram( const res = await sendTelegramText(text, textParams, opts.plainText); const messageId = String(res?.message_id ?? "unknown"); if (res?.message_id) { - recordSentMessage(chatId, res.message_id); + if (opts.isStatusMessage) { + recordSentMessage(chatId, Number(res.message_id), true); + } else { + recordSentMessage(chatId, Number(res.message_id), false); + } } recordChannelActivity({ channel: "telegram", @@ -451,6 +469,89 @@ export async function deleteMessageTelegram( return { ok: true }; } +type TelegramEditOpts = { + token?: string; + accountId?: string; + verbose?: boolean; + api?: Bot["api"]; + retry?: RetryConfig; + textMode?: "markdown" | "html"; + buttons?: TelegramSendOpts["buttons"]; +}; + +export async function editMessageTelegram( + chatIdInput: string | number, + messageIdInput: string | number, + text: string, + opts: TelegramEditOpts = {}, +): Promise<{ ok: true }> { + const cfg = loadConfig(); + const account = resolveTelegramAccount({ + cfg, + accountId: opts.accountId, + }); + const token = resolveToken(opts.token, account); + const chatId = normalizeChatId(String(chatIdInput)); + const messageId = normalizeMessageId(messageIdInput); + const fetchImpl = resolveTelegramFetch(); + const client: ApiClientOptions | undefined = fetchImpl + ? { fetch: fetchImpl as unknown as ApiClientOptions["fetch"] } + : undefined; + const api = opts.api ?? new Bot(token, client ? { client } : undefined).api; + const request = createTelegramRetryRunner({ + retry: opts.retry, + configRetry: account.config.retry, + verbose: opts.verbose, + }); + const replyMarkup = buildInlineKeyboard(opts.buttons); + + const renderHtmlText = (rawText: string) => + renderTelegramHtmlText(rawText, { + tableMode: resolveMarkdownTableMode({ + cfg, + channel: "telegram", + accountId: opts.accountId, + }), + }); + + const textMode = opts.textMode ?? "html"; + const htmlText = textMode === "html" ? renderHtmlText(text) : text; + + try { + await request( + () => + api.editMessageText(chatId, messageId, htmlText, { + parse_mode: textMode === "html" ? "HTML" : "MarkdownV2", + ...(replyMarkup ? { reply_markup: replyMarkup } : {}), + }), + "editMessage", + ); + console.error( + `[CRITICAL LOG] [telegram/send.ts] Edited message ${messageId} in chat ${chatId}. Payload: ${JSON.stringify({ text, htmlText })}`, + ); + } catch (err) { + // If HTML parsing fails, fall back to plain text + const errText = formatErrorMessage(err); + if (PARSE_ERR_RE.test(errText)) { + if (opts.verbose) { + console.warn(`telegram HTML parse failed on edit, retrying as plain text: ${errText}`); + } + await request( + () => + api.editMessageText(chatId, messageId, text, { + ...(replyMarkup ? { reply_markup: replyMarkup } : {}), + }), + "editMessage-plain", + ); + logVerbose(`[telegram] Edited message ${messageId} in chat ${chatId} (plain text fallback)`); + } else { + throw err; + } + } + + return { ok: true }; +} + function inferFilename(kind: ReturnType) { switch (kind) { case "image": diff --git a/src/telegram/sent-message-cache.ts b/src/telegram/sent-message-cache.ts index 05c12ddf7..7a9f11869 100644 --- a/src/telegram/sent-message-cache.ts +++ b/src/telegram/sent-message-cache.ts @@ -8,6 +8,7 @@ const TTL_MS = 24 * 60 * 60 * 1000; // 24 hours type CacheEntry = { messageIds: Set; timestamps: Map; + lastSent?: { messageId: number; timestamp: number; isStatus: boolean }; }; const sentMessages = new Map(); @@ -29,21 +30,50 @@ function cleanupExpired(entry: CacheEntry): void { /** * Record a message ID as sent by the bot. */ -export function recordSentMessage(chatId: number | string, messageId: number): void { +export function recordSentMessage( + chatId: number | string, + messageId: number, + isStatus: boolean = false, +): void { const key = getChatKey(chatId); let entry = sentMessages.get(key); if (!entry) { entry = { messageIds: new Set(), timestamps: new Map() }; sentMessages.set(key, entry); } + const now = Date.now(); entry.messageIds.add(messageId); - entry.timestamps.set(messageId, Date.now()); + entry.timestamps.set(messageId, now); + entry.lastSent = { messageId, timestamp: now, isStatus }; // Periodic cleanup if (entry.messageIds.size > 100) { cleanupExpired(entry); } } +/** + * Invalidate the "status" flag of the last sent message. + * This prevents subsequent messages from overwriting it. + */ +export function invalidateLastStatus(chatId: number | string): void { + const key = getChatKey(chatId); + const entry = sentMessages.get(key); + if (entry?.lastSent) { + entry.lastSent.isStatus = false; + } +} + +/** + * Get the last sent message ID and timestamp for a chat. + */ +export function getLastSentMessage( + chatId: number | string, +): { messageId: number; timestamp: number; isStatus: boolean } | undefined { + const key = getChatKey(chatId); + const entry = sentMessages.get(key); + return entry?.lastSent; +} + /** * Check if a message was sent by the bot. */