romtuck/openclaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1589: final draft - pre log

Browse Source
This commit is contained in:
USO Status 2026-01-26 20:30:56 +01:00
parent 6a9301c27d
commit 005b601b13
22 changed files with 1291 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
scripts/run-test-gateway.sh Executable file
View File

@ -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"

15
src/auto-reply/reply/agent-runner-execution.ts
View File

@ -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<string, SessionEntry>;
storePath?: string;
resolvedVerboseLevel: VerboseLevel;
statusUpdateContext?: StatusUpdateRunContext;
}): Promise<AgentRunLoopResult> {
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") {

109
src/auto-reply/reply/agent-runner.ts
View File

@ -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 =====
}
}

12
src/auto-reply/reply/block-reply-pipeline.ts
View File

@ -69,7 +69,7 @@ export function createBlockReplyPipeline(params: {
onBlockReply: (
payload: ReplyPayload,
options?: { abortSignal?: AbortSignal; timeoutMs?: number },
) => Promise<void> | void;
) => Promise<string | void> | 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,
);

201
src/auto-reply/reply/status-updates-integration.ts Normal file
View File

@ -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<string, unknown>;
}): 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<string, unknown> },
): Promise<void> {
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<string | undefined> {
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<void> {
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,
};

346
src/auto-reply/reply/status-updates.ts Normal file
View File

@ -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<string | undefined>;
/** Edit an existing message with final content and checkmark. */
editFinal?: (text: string, messageId: string) => Promise<void>;
/** Check if the channel supports message editing. */
supportsEdit?: () => boolean;
/** Delete a status message (for cleanup). */
deleteStatus?: (messageId: string) => Promise<void>;
};
export type StatusUpdateState = {
phase: StatusPhase;
startedAt: number;
statusMessageId?: string;
lastUpdateAt: number;
elapsedSeconds: number;
isComplete: boolean;
};
const DEFAULT_CONFIG: Required<StatusUpdateConfig> = {
enabled: false,
mode: "edit",
updateIntervalMs: 5000,
showPhases: true,
showElapsedTime: true,
markFinalWithCheckmark: true,
elapsedTimeThresholdMs: 3000,
};
const PHASE_MESSAGES: Record<StatusPhase, string> = {
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<StatusPhase, string> = {
sending_query: "⏳",
receiving_reasoning: "🧠",
processing_tools: "🔧",
generating_response: "✍️",
complete: "✅",
};
export function resolveStatusUpdateConfig(
config?: Partial<StatusUpdateConfig>,
): Required<StatusUpdateConfig> {
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<StatusUpdateConfig>,
): 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<StatusUpdateConfig>;
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<void> {
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<void> {
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<string | undefined> {
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<void> {
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<void> {
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<string | undefined>;
editMessage: (text: string, messageId: string) => Promise<void>;
deleteMessage?: (messageId: string) => Promise<void>;
}): 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<string | undefined>;
}): StatusUpdateCallbacks {
return {
sendStatus: async (text) => {
return params.sendMessage(text);
},
supportsEdit: () => false,
};
}

9
src/auto-reply/types.ts
View File

@ -27,7 +27,10 @@ export type GetReplyOptions = {
isHeartbeat?: boolean;
onPartialReply?: (payload: ReplyPayload) => Promise<void> | void;
onReasoningStream?: (payload: ReplyPayload) => Promise<void> | void;
onBlockReply?: (payload: ReplyPayload, context?: BlockReplyContext) => Promise<void> | void;
onBlockReply?: (
payload: ReplyPayload,
context?: BlockReplyContext,
) => Promise<string | void> | string | void;
onToolResult?: (payload: ReplyPayload) => Promise<void> | 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;
};

8
src/config/schema.ts
View File

@ -262,6 +262,14 @@ const FIELD_LABELS: Record<string, string> = {
"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",

6
src/config/types.agent-defaults.ts
View File

@ -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). */

4
src/config/types.agents.ts
View File

@ -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?: {

25
src/config/types.base.ts
View File

@ -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;
};

3
src/config/zod-schema.agent-defaults.ts
View File

@ -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

3
src/config/zod-schema.agent-runtime.ts
View File

@ -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

32
src/config/zod-schema.core.ts
View File

@ -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();

6
src/telegram/bot-handlers.ts
View File

@ -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();

2
src/telegram/bot-message-dispatch.ts
View File

@ -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,

1
src/telegram/bot-native-commands.ts
View File

@ -341,6 +341,7 @@ export const registerTelegramNativeCommands = ({
replies: [payload],
chatId: String(chatId),
token: opts.token,
accountId,
runtime,
bot,
replyToMode,

5
src/telegram/bot/delivery.test.ts
View File

@ -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",

78
src/telegram/bot/delivery.ts
View File

@ -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

243
src/telegram/send.intermittent.test.ts Normal file
View File

@ -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<typeof import("../config/config.js")>();
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 },
}),
);
});
});

105
src/telegram/send.ts
View File

@ -42,6 +42,7 @@ type TelegramSendOpts = {
messageThreadId?: number;
/** Inline keyboard buttons (reply markup). */
buttons?: Array<Array<{ text: string; callback_data: string }>>;
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<string, unknown>,
@ -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<typeof mediaKindFromMime>) {
switch (kind) {
case "image":

34
src/telegram/sent-message-cache.ts
View File

@ -8,6 +8,7 @@ const TTL_MS = 24 * 60 * 60 * 1000; // 24 hours
type CacheEntry = {
messageIds: Set<number>;
timestamps: Map<number, number>;
lastSent?: { messageId: number; timestamp: number; isStatus: boolean };
};
const sentMessages = new Map<string, CacheEntry>();
@ -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.
*/