Skip to content

Commit 741eafe

Browse files
obviyusobviyus
committed
fix(telegram): distinguish partial reply snapshots
1 parent 8880a58 commit 741eafe

2 files changed

Lines changed: 111 additions & 23 deletions

File tree

‎extensions/telegram/src/message-cache.test.ts‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -221,6 +221,57 @@ describe("telegram message cache", () => {
221221
}
222222
});
223223

224+
it("replaces authoritative edited message fields without stale caption carryover", () => {
225+
const cache = createTelegramMessageCache();
226+
const chat = { id: 7, type: "group", title: "Ops" } as const;
227+
cache.record({
228+
accountId: "default",
229+
chatId: 7,
230+
msg: {
231+
chat,
232+
message_id: 104,
233+
date: 1736380900,
234+
caption: "old caption",
235+
from: { id: 999, is_bot: true, first_name: "Bot" },
236+
photo: [
237+
{
238+
file_id: "generated-photo-2",
239+
file_unique_id: "generated-photo-unique-2",
240+
width: 640,
241+
height: 480,
242+
},
243+
],
244+
} as Message,
245+
});
246+
247+
const updated = cache.record({
248+
accountId: "default",
249+
chatId: 7,
250+
msg: {
251+
chat,
252+
message_id: 104,
253+
date: 1736380900,
254+
edit_date: 1736380910,
255+
from: { id: 999, is_bot: true, first_name: "Bot" },
256+
photo: [
257+
{
258+
file_id: "generated-photo-2",
259+
file_unique_id: "generated-photo-unique-2",
260+
width: 640,
261+
height: 480,
262+
},
263+
],
264+
} as Message,
265+
});
266+
267+
expect(updated).toMatchObject({
268+
messageId: "104",
269+
body: "<media:image>",
270+
mediaRef: "telegram:file/generated-photo-2",
271+
});
272+
expect(updated?.body).not.toBe("old caption");
273+
});
274+
224275
it("shares one persisted bucket across live cache instances", async () => {
225276
const storePath = `/tmp/openclaw-telegram-message-cache-shared-${process.pid}-${Date.now()}.json`;
226277
const persistedPath = resolveTelegramMessageCachePath(storePath);

‎extensions/telegram/src/message-cache.ts‎

Lines changed: 60 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,13 @@ type PersistedMessageReadResult = TelegramMessageCacheBucket & {
6363
needsRewrite: boolean;
6464
};
6565

66+
type TelegramMessageObservationMode = "authoritative" | "partial";
67+
68+
type TelegramCachedMessageObservation = {
69+
node: TelegramCachedMessageNode;
70+
mode: TelegramMessageObservationMode;
71+
};
72+
6673
const DEFAULT_MAX_MESSAGES = 5000;
6774
const COMPACT_THRESHOLD_RATIO = 2;
6875
const persistedMessageCacheBuckets = new Map<string, TelegramMessageCacheBucket>();
@@ -164,14 +171,18 @@ function resolveMessageThreadId(msg: Message): number | undefined {
164171
function normalizeMessageNodes(
165172
msg: Message,
166173
params: { threadId?: number },
167-
): TelegramCachedMessageNode[] {
168-
const nodes: TelegramCachedMessageNode[] = [];
174+
): TelegramCachedMessageObservation[] {
175+
const observations: TelegramCachedMessageObservation[] = [];
169176
const visited = new Set<string>();
170177
const nodeThreadId = (node: TelegramCachedMessageNode) => {
171178
const threadId = Number(node.threadId);
172179
return Number.isFinite(threadId) ? threadId : undefined;
173180
};
174-
const visit = (message: Message, inheritedThreadId?: number) => {
181+
const visit = (
182+
message: Message,
183+
inheritedThreadId: number | undefined,
184+
mode: TelegramMessageObservationMode,
185+
) => {
175186
const node = normalizeMessageNode(message, {
176187
threadId: resolveMessageThreadId(message) ?? inheritedThreadId,
177188
});
@@ -181,12 +192,12 @@ function normalizeMessageNodes(
181192
visited.add(node.messageId);
182193
const replyMessage = resolveEmbeddedReplyMessage(message);
183194
if (replyMessage?.message_id != null) {
184-
visit(replyMessage, nodeThreadId(node) ?? inheritedThreadId);
195+
visit(replyMessage, nodeThreadId(node) ?? inheritedThreadId, "partial");
185196
}
186-
nodes.push(node);
197+
observations.push({ node, mode });
187198
};
188-
visit(msg, params.threadId);
189-
return nodes;
199+
visit(msg, params.threadId, "authoritative");
200+
return observations;
190201
}
191202

192203
function isRecord(value: unknown): value is Record<string, unknown> {
@@ -215,6 +226,7 @@ function isTelegramSourceMessage(value: unknown): value is Message {
215226
function parsePersistedEntry(value: unknown): Array<{
216227
key: string;
217228
node: TelegramCachedMessageNode;
229+
mode: TelegramMessageObservationMode;
218230
}> {
219231
if (!isRecord(value) || !isString(value.key)) {
220232
return [];
@@ -229,10 +241,15 @@ function parsePersistedEntry(value: unknown): Array<{
229241
}
230242
const keyPrefix = value.key.slice(0, separatorIndex + 1);
231243
const threadId = Number(readOptionalString(value.node, "threadId"));
244+
const sourceMessageId = String(value.node.sourceMessage.message_id);
232245
return normalizeMessageNodes(
233246
value.node.sourceMessage,
234247
Number.isFinite(threadId) ? { threadId } : {},
235-
).map((node) => ({ key: `${keyPrefix}${node.messageId}`, node }));
248+
).map(({ node, mode }) => ({
249+
key: `${keyPrefix}${node.messageId}`,
250+
node,
251+
mode: node.messageId === sourceMessageId ? "authoritative" : mode,
252+
}));
236253
}
237254

238255
function findJsonArrayEnd(text: string): number {
@@ -337,26 +354,41 @@ function mergeTelegramSourceMessage(existing: Message, incoming: Message): Messa
337354
return merged;
338355
}
339356

357+
function mergeAuthoritativeTelegramSourceMessage(existing: Message, incoming: Message): Message {
358+
const existingReply = resolveEmbeddedReplyMessage(existing);
359+
const incomingReply = resolveEmbeddedReplyMessage(incoming);
360+
if (existingReply?.message_id != null && incomingReply?.message_id === existingReply.message_id) {
361+
return {
362+
...incoming,
363+
reply_to_message: mergeTelegramSourceMessage(existingReply, incomingReply),
364+
};
365+
}
366+
return incoming;
367+
}
368+
340369
function mergeCachedMessageNode(
341370
existing: TelegramCachedMessageNode,
342371
incoming: TelegramCachedMessageNode,
372+
mode: TelegramMessageObservationMode,
343373
): TelegramCachedMessageNode {
344374
const threadId = Number(incoming.threadId ?? existing.threadId);
345-
return normalizeRequiredMessageNode(
346-
mergeTelegramSourceMessage(existing.sourceMessage, incoming.sourceMessage),
347-
{
348-
...(Number.isFinite(threadId) ? { threadId } : {}),
349-
},
350-
);
375+
const sourceMessage =
376+
mode === "authoritative"
377+
? mergeAuthoritativeTelegramSourceMessage(existing.sourceMessage, incoming.sourceMessage)
378+
: mergeTelegramSourceMessage(existing.sourceMessage, incoming.sourceMessage);
379+
return normalizeRequiredMessageNode(sourceMessage, {
380+
...(Number.isFinite(threadId) ? { threadId } : {}),
381+
});
351382
}
352383

353384
function upsertCachedMessageNode(params: {
354385
messages: Map<string, TelegramCachedMessageNode>;
355386
key: string;
356387
node: TelegramCachedMessageNode;
388+
mode: TelegramMessageObservationMode;
357389
}): TelegramCachedMessageNode {
358390
const existing = params.messages.get(params.key);
359-
const node = existing ? mergeCachedMessageNode(existing, params.node) : params.node;
391+
const node = existing ? mergeCachedMessageNode(existing, params.node, params.mode) : params.node;
360392
params.messages.delete(params.key);
361393
params.messages.set(params.key, node);
362394
return node;
@@ -375,7 +407,12 @@ function readPersistedMessages(filePath: string, maxMessages: number): Persisted
375407
for (const value of persisted.values) {
376408
for (const entry of parsePersistedEntry(value)) {
377409
persistedEntryCount++;
378-
upsertCachedMessageNode({ messages, key: entry.key, node: entry.node });
410+
upsertCachedMessageNode({
411+
messages,
412+
key: entry.key,
413+
node: entry.node,
414+
mode: entry.mode,
415+
});
379416
trimMessages(messages, maxMessages);
380417
}
381418
}
@@ -515,16 +552,16 @@ export function createTelegramMessageCache(params?: {
515552

516553
return {
517554
record: ({ accountId, chatId, msg, threadId }) => {
518-
const entries = normalizeMessageNodes(msg, { threadId });
519-
const entry = entries.at(-1);
520-
if (!entry) {
555+
const observations = normalizeMessageNodes(msg, { threadId });
556+
const currentObservation = observations.at(-1);
557+
if (!currentObservation) {
521558
return null;
522559
}
523560
let recordedEntry: TelegramCachedMessageNode | null = null;
524-
for (const node of entries) {
561+
for (const { node, mode } of observations) {
525562
const key = telegramMessageCacheKey({ accountId, chatId, messageId: node.messageId });
526-
const cachedNode = upsertCachedMessageNode({ messages, key, node });
527-
if (node.messageId === entry.messageId) {
563+
const cachedNode = upsertCachedMessageNode({ messages, key, node, mode });
564+
if (node.messageId === currentObservation.node.messageId) {
528565
recordedEntry = cachedNode;
529566
}
530567
trimMessages(messages, maxMessages);
@@ -544,7 +581,7 @@ export function createTelegramMessageCache(params?: {
544581
logVerbose(`telegram: failed to persist message cache: ${String(error)}`);
545582
}
546583
}
547-
return recordedEntry ?? entry;
584+
return recordedEntry ?? currentObservation.node;
548585
},
549586
get,
550587
recentBefore: ({ accountId, chatId, messageId, threadId, limit }) => {

0 commit comments

Comments
 (0)