Skip to content

Commit 3ad465d

Browse files
obviyusobviyus
committed
fix(telegram): persist ambient transcript rows
Persist room-event observations as durable bare user transcript rows and carry an ambient transcript watermark through session state so Telegram chat windows only include the unpersisted gap. Fixes #99257
1 parent b47fe7a commit 3ad465d

29 files changed

Lines changed: 1057 additions & 81 deletions

‎extensions/telegram/src/bot-deps.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@ import {
1919
getSessionEntry,
2020
listSessionEntries,
2121
readSessionUpdatedAt,
22+
readAmbientTranscriptWatermark,
23+
resolveAmbientTranscriptWatermarkKey,
2224
resolveStorePath,
2325
} from "openclaw/plugin-sdk/session-store-runtime";
2426
import { loadSessionStore } from "openclaw/plugin-sdk/session-store-runtime";
@@ -40,6 +42,8 @@ export type TelegramBotDeps = {
4042
listSessionEntries?: typeof listSessionEntries;
4143
loadSessionStore?: typeof loadSessionStore;
4244
readSessionUpdatedAt?: typeof readSessionUpdatedAt;
45+
readAmbientTranscriptWatermark?: typeof readAmbientTranscriptWatermark;
46+
resolveAmbientTranscriptWatermarkKey?: typeof resolveAmbientTranscriptWatermarkKey;
4347
recordInboundSession?: typeof recordInboundSession;
4448
recordChannelActivity?: typeof recordChannelActivity;
4549
resolveInboundLastRouteSessionKey?: typeof resolveInboundLastRouteSessionKey;
@@ -86,6 +90,12 @@ export const defaultTelegramBotDeps: TelegramBotDeps = {
8690
get readSessionUpdatedAt() {
8791
return readSessionUpdatedAt;
8892
},
93+
get readAmbientTranscriptWatermark() {
94+
return readAmbientTranscriptWatermark;
95+
},
96+
get resolveAmbientTranscriptWatermarkKey() {
97+
return resolveAmbientTranscriptWatermarkKey;
98+
},
8999
get recordInboundSession() {
90100
return recordInboundSession;
91101
},

‎extensions/telegram/src/bot-handlers.runtime.ts‎

Lines changed: 144 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,8 @@ import {
4343
getSessionEntry,
4444
listSessionEntries,
4545
patchSessionEntry,
46+
readAmbientTranscriptWatermark,
47+
resolveAmbientTranscriptWatermarkKey,
4648
} from "openclaw/plugin-sdk/session-store-runtime";
4749
import { normalizeStringEntries } from "openclaw/plugin-sdk/string-coerce-runtime";
4850
import { expandTelegramAllowFromWithAccessGroups } from "./access-groups.js";
@@ -77,6 +79,7 @@ import type {
7779
TelegramMessageContextOptions,
7880
TelegramPromptContextEntry,
7981
} from "./bot-message-context.types.js";
82+
import type { TelegramAmbientTranscriptWatermark } from "./bot-message-context.types.js";
8083
import { parseTelegramNativeCommandCallbackData } from "./bot-native-commands.js";
8184
import type { RegisterTelegramHandlerParams } from "./bot-native-commands.js";
8285
import {
@@ -93,6 +96,7 @@ import {
9396
} from "./bot-updates.js";
9497
import { resolveMedia } from "./bot/delivery.resolve-media.js";
9598
import {
99+
buildSenderName,
96100
getTelegramTextParts,
97101
hasBotMention,
98102
buildTelegramThreadParams,
@@ -106,6 +110,7 @@ import {
106110
resolveTelegramThreadSpec,
107111
loadTelegramPairingStoreIfNeeded,
108112
resolveTelegramBotHasTopicsEnabled,
113+
resolveTelegramMediaPlaceholder,
109114
TelegramPairingStoreReadError,
110115
shouldUseTelegramDmThreadSession,
111116
withResolvedTelegramForumFlag,
@@ -130,6 +135,7 @@ import {
130135
evaluateTelegramGroupPolicyAccess,
131136
} from "./group-access.js";
132137
import { resolveTelegramScopedGroupConfig } from "./group-config-helpers.js";
138+
import { isTelegramHistoryEntryAfterAmbientWatermark } from "./group-history-window.js";
133139
import { migrateTelegramGroupConfig } from "./group-migration.js";
134140
import {
135141
resolveTelegramCommandIngressAuthorization,
@@ -265,6 +271,7 @@ export const registerTelegramHandlers = ({
265271
threadId?: number;
266272
messages: Array<{ msg: Message; ctx: TelegramContext; receivedAtMs: number }>;
267273
promptContextMinTimestampMs?: number;
274+
promptContextAmbientWatermark?: TelegramAmbientTranscriptWatermark;
268275
dispatchDedupeKeys: string[];
269276
spooledReplayParticipants: TelegramSpooledReplayDeferredParticipant[];
270277
timer: ReturnType<typeof setTimeout>;
@@ -300,6 +307,7 @@ export const registerTelegramHandlers = ({
300307
botUsername?: string;
301308
threadId?: number;
302309
promptContextMinTimestampMs?: number;
310+
promptContextAmbientWatermark?: TelegramAmbientTranscriptWatermark;
303311
dispatchDedupeKeys: string[];
304312
spooledReplayParticipant?: TelegramSpooledReplayDeferredParticipant;
305313
};
@@ -326,9 +334,18 @@ export const registerTelegramHandlers = ({
326334
typeof timestampMs === "number" && Number.isFinite(timestampMs) ? timestampMs : undefined;
327335
const promptContextBoundaryOptions = (
328336
timestampMs?: number,
329-
): Pick<TelegramMessageContextOptions, "promptContextMinTimestampMs"> => {
337+
ambientWatermark?: TelegramAmbientTranscriptWatermark,
338+
): Pick<
339+
TelegramMessageContextOptions,
340+
"promptContextMinTimestampMs" | "promptContextAmbientWatermark"
341+
> => {
330342
const promptContextMinTimestampMs = normalizePromptContextMinTimestampMs(timestampMs);
331-
return promptContextMinTimestampMs === undefined ? {} : { promptContextMinTimestampMs };
343+
return {
344+
...(promptContextMinTimestampMs === undefined ? {} : { promptContextMinTimestampMs }),
345+
...(ambientWatermark === undefined
346+
? {}
347+
: { promptContextAmbientWatermark: ambientWatermark }),
348+
};
332349
};
333350
const latestPromptContextMinTimestampMs = (
334351
...timestamps: Array<number | undefined>
@@ -343,6 +360,11 @@ export const registerTelegramHandlers = ({
343360
}
344361
return latest;
345362
};
363+
const latestPromptContextAmbientWatermark = (
364+
...watermarks: Array<TelegramAmbientTranscriptWatermark | undefined>
365+
): TelegramAmbientTranscriptWatermark | undefined => {
366+
return watermarks.findLast((watermark) => watermark !== undefined);
367+
};
346368
const mergeDispatchDedupeKeys = (...groups: Array<readonly string[] | undefined>) => [
347369
...new Set(normalizeStringEntries(groups.flatMap((group) => group ?? []))),
348370
];
@@ -435,6 +457,23 @@ export const registerTelegramHandlers = ({
435457
message: Message,
436458
): TelegramContext => ({ message, me: ctx.me, getFile: ctx.getFile.bind(ctx) });
437459

460+
const formatTelegramAmbientTranscriptLine = (msg: Message): string => {
461+
const text = getTelegramTextParts(msg).text.trim();
462+
const body =
463+
text || resolveTelegramMediaPlaceholder(msg) || "[User sent media without caption]";
464+
const messageId = msg.message_id ? `#${msg.message_id}` : undefined;
465+
const sender = buildSenderName(msg);
466+
const prefix = [messageId, sender].filter(Boolean).join(" ");
467+
return prefix ? `${prefix}: ${body}` : body;
468+
};
469+
470+
const formatTelegramAmbientTranscriptBody = (
471+
messages: readonly Message[],
472+
): string | undefined => {
473+
const lines = messages.map(formatTelegramAmbientTranscriptLine);
474+
return lines.length > 0 ? lines.join("\n") : undefined;
475+
};
476+
438477
const MULTI_SELECT_PREFIX = "OC_MULTI|";
439478
const MULTI_SELECT_TOGGLE_PREFIX = `${MULTI_SELECT_PREFIX}toggle|`;
440479
const SELECT_PREFIX = "OC_SELECT|";
@@ -589,7 +628,10 @@ export const registerTelegramHandlers = ({
589628
options: {
590629
receivedAtMs: last.receivedAtMs,
591630
ingressBuffer: "inbound-debounce",
592-
...promptContextBoundaryOptions(last.promptContextMinTimestampMs),
631+
...promptContextBoundaryOptions(
632+
last.promptContextMinTimestampMs,
633+
last.promptContextAmbientWatermark,
634+
),
593635
...spooledReplayOptions(spooledReplayParticipants),
594636
},
595637
dispatchDedupeKeys: last.dispatchDedupeKeys,
@@ -610,6 +652,9 @@ export const registerTelegramHandlers = ({
610652
const promptContextMinTimestampMs = latestPromptContextMinTimestampMs(
611653
...entries.map((entry) => entry.promptContextMinTimestampMs),
612654
);
655+
const promptContextAmbientWatermark = latestPromptContextAmbientWatermark(
656+
...entries.map((entry) => entry.promptContextAmbientWatermark),
657+
);
613658
const baseCtx = first.ctx;
614659
const syntheticMessage = buildSyntheticTextMessage({
615660
base: first.msg,
@@ -625,9 +670,15 @@ export const registerTelegramHandlers = ({
625670
storeAllowFrom: first.storeAllowFrom,
626671
options: {
627672
...(messageIdOverride ? { messageIdOverride } : {}),
673+
ambientTranscriptBody: formatTelegramAmbientTranscriptBody(
674+
entries.map((entry) => entry.msg),
675+
),
628676
receivedAtMs: first.receivedAtMs,
629677
ingressBuffer: "inbound-debounce",
630-
...promptContextBoundaryOptions(promptContextMinTimestampMs),
678+
...promptContextBoundaryOptions(
679+
promptContextMinTimestampMs,
680+
promptContextAmbientWatermark,
681+
),
631682
...spooledReplayOptions(spooledReplayParticipants),
632683
},
633684
dispatchDedupeKeys: mergeDispatchDedupeKeys(
@@ -786,6 +837,31 @@ export const registerTelegramHandlers = ({
786837
};
787838
};
788839

840+
const resolvePromptContextAmbientWatermark = (params: {
841+
chatId: number | string;
842+
isGroup: boolean;
843+
resolvedThreadId?: number;
844+
sessionKey: string;
845+
storePath: string;
846+
}): TelegramAmbientTranscriptWatermark | undefined => {
847+
if (!params.isGroup) {
848+
return undefined;
849+
}
850+
const key = (
851+
telegramDeps.resolveAmbientTranscriptWatermarkKey ?? resolveAmbientTranscriptWatermarkKey
852+
)({
853+
channel: "telegram",
854+
accountId,
855+
conversationId: String(params.chatId),
856+
...(params.resolvedThreadId !== undefined ? { threadId: params.resolvedThreadId } : {}),
857+
});
858+
return (telegramDeps.readAmbientTranscriptWatermark ?? readAmbientTranscriptWatermark)({
859+
storePath: params.storePath,
860+
sessionKey: params.sessionKey,
861+
key,
862+
});
863+
};
864+
789865
const mediaMayNeedDownloadForMentionDetection = (msg: Message): boolean => {
790866
const textParts = getTelegramTextParts(msg);
791867
if (textParts.text.trim()) {
@@ -1016,7 +1092,10 @@ export const registerTelegramHandlers = ({
10161092
promptContextMessageSelection,
10171093
storeAllowFrom: entry.storeAllowFrom,
10181094
options: {
1019-
...promptContextBoundaryOptions(entry.promptContextMinTimestampMs),
1095+
...promptContextBoundaryOptions(
1096+
entry.promptContextMinTimestampMs,
1097+
entry.promptContextAmbientWatermark,
1098+
),
10201099
...spooledReplayOptions(entry.spooledReplayParticipants),
10211100
},
10221101
dispatchDedupeKeys: entry.dispatchDedupeKeys,
@@ -1068,9 +1147,15 @@ export const registerTelegramHandlers = ({
10681147
storeAllowFrom,
10691148
options: {
10701149
messageIdOverride: String(last.msg.message_id),
1150+
ambientTranscriptBody: formatTelegramAmbientTranscriptBody(
1151+
entry.messages.map((message) => message.msg),
1152+
),
10711153
receivedAtMs: first.receivedAtMs,
10721154
ingressBuffer: "text-fragment",
1073-
...promptContextBoundaryOptions(entry.promptContextMinTimestampMs),
1155+
...promptContextBoundaryOptions(
1156+
entry.promptContextMinTimestampMs,
1157+
entry.promptContextAmbientWatermark,
1158+
),
10741159
...spooledReplayOptions(entry.spooledReplayParticipants),
10751160
},
10761161
dispatchDedupeKeys: entry.dispatchDedupeKeys,
@@ -1241,6 +1326,21 @@ export const registerTelegramHandlers = ({
12411326
...(options?.promptContextMinTimestampMs !== undefined
12421327
? { minTimestampMs: options.promptContextMinTimestampMs }
12431328
: {}),
1329+
...(isGroup && options?.promptContextAmbientWatermark !== undefined
1330+
? {
1331+
includeNode: (
1332+
node: TelegramCachedMessageNode,
1333+
flags?: { replyTarget?: boolean },
1334+
) =>
1335+
// Explicit reply targets stay visible so the current turn is not shown
1336+
// as a reply to invisible transcript-owned text.
1337+
flags?.replyTarget === true ||
1338+
isTelegramHistoryEntryAfterAmbientWatermark(
1339+
node,
1340+
options.promptContextAmbientWatermark,
1341+
),
1342+
}
1343+
: {}),
12441344
});
12451345
const conversationContextById = new Map(
12461346
conversationContext.flatMap((entry) =>
@@ -1968,6 +2068,7 @@ export const registerTelegramHandlers = ({
19682068
sendOversizeWarning: boolean;
19692069
oversizeLogMessage: string;
19702070
promptContextMinTimestampMs?: number;
2071+
promptContextAmbientWatermark?: TelegramAmbientTranscriptWatermark;
19712072
dispatchDedupeKeys: string[];
19722073
}) => {
19732074
const {
@@ -1988,6 +2089,7 @@ export const registerTelegramHandlers = ({
19882089
sendOversizeWarning,
19892090
oversizeLogMessage,
19902091
promptContextMinTimestampMs,
2092+
promptContextAmbientWatermark,
19912093
dispatchDedupeKeys,
19922094
} = params;
19932095

@@ -2064,6 +2166,10 @@ export const registerTelegramHandlers = ({
20642166
existing.promptContextMinTimestampMs,
20652167
promptContextMinTimestampMs,
20662168
);
2169+
existing.promptContextAmbientWatermark = latestPromptContextAmbientWatermark(
2170+
existing.promptContextAmbientWatermark,
2171+
promptContextAmbientWatermark,
2172+
);
20672173
existing.dispatchDedupeKeys = mergeDispatchDedupeKeys(
20682174
existing.dispatchDedupeKeys,
20692175
dispatchDedupeKeys,
@@ -2089,7 +2195,10 @@ export const registerTelegramHandlers = ({
20892195
messages: [{ msg, ctx, receivedAtMs: nowMs }],
20902196
dispatchDedupeKeys,
20912197
spooledReplayParticipants: spooledReplayParticipant ? [spooledReplayParticipant] : [],
2092-
...promptContextBoundaryOptions(promptContextMinTimestampMs),
2198+
...promptContextBoundaryOptions(
2199+
promptContextMinTimestampMs,
2200+
promptContextAmbientWatermark,
2201+
),
20932202
timer: setTimeout(() => {}, TELEGRAM_TEXT_FRAGMENT_MAX_GAP_MS),
20942203
};
20952204
textFragmentBuffer.set(key, entry);
@@ -2128,6 +2237,10 @@ export const registerTelegramHandlers = ({
21282237
existing.promptContextMinTimestampMs,
21292238
promptContextMinTimestampMs,
21302239
);
2240+
existing.promptContextAmbientWatermark = latestPromptContextAmbientWatermark(
2241+
existing.promptContextAmbientWatermark,
2242+
promptContextAmbientWatermark,
2243+
);
21312244
existing.dispatchDedupeKeys = mergeDispatchDedupeKeys(
21322245
existing.dispatchDedupeKeys,
21332246
dispatchDedupeKeys,
@@ -2156,7 +2269,10 @@ export const registerTelegramHandlers = ({
21562269
topicConfig,
21572270
dispatchDedupeKeys,
21582271
spooledReplayParticipants: spooledReplayParticipant ? [spooledReplayParticipant] : [],
2159-
...promptContextBoundaryOptions(promptContextMinTimestampMs),
2272+
...promptContextBoundaryOptions(
2273+
promptContextMinTimestampMs,
2274+
promptContextAmbientWatermark,
2275+
),
21602276
timer: setTimeout(() => {
21612277
mediaGroupBuffer.delete(mediaGroupKey);
21622278
void queueBufferedProcessing(mediaGroupProcessingByKey, mediaGroupKey, async () => {
@@ -2290,7 +2406,7 @@ export const registerTelegramHandlers = ({
22902406
debounceKey: isAbortControlMessage ? null : debounceKey,
22912407
debounceLane,
22922408
botUsername,
2293-
...promptContextBoundaryOptions(promptContextMinTimestampMs),
2409+
...promptContextBoundaryOptions(promptContextMinTimestampMs, promptContextAmbientWatermark),
22942410
dispatchDedupeKeys,
22952411
};
22962412
if (
@@ -3305,18 +3421,26 @@ export const registerTelegramHandlers = ({
33053421
effectiveGroupAllow,
33063422
} = gate.context;
33073423

3424+
const sessionState = resolveTelegramSessionState({
3425+
chatId: event.chatId,
3426+
isGroup: event.isGroup,
3427+
isForum: event.isForum,
3428+
messageThreadId: event.messageThreadId,
3429+
resolvedThreadId,
3430+
botHasTopicsEnabled: resolveTelegramBotHasTopicsEnabled(event.ctx.me),
3431+
senderId: event.senderId,
3432+
runtimeCfg: cfg,
3433+
});
33083434
const promptContextMinTimestampMs = normalizePromptContextMinTimestampMs(
3309-
resolveTelegramSessionState({
3310-
chatId: event.chatId,
3311-
isGroup: event.isGroup,
3312-
isForum: event.isForum,
3313-
messageThreadId: event.messageThreadId,
3314-
resolvedThreadId,
3315-
botHasTopicsEnabled: resolveTelegramBotHasTopicsEnabled(event.ctx.me),
3316-
senderId: event.senderId,
3317-
runtimeCfg: cfg,
3318-
}).sessionEntry?.sessionStartedAt,
3435+
sessionState.sessionEntry?.sessionStartedAt,
33193436
);
3437+
const promptContextAmbientWatermark = resolvePromptContextAmbientWatermark({
3438+
chatId: event.chatId,
3439+
isGroup: event.isGroup,
3440+
resolvedThreadId,
3441+
sessionKey: sessionState.sessionKey,
3442+
storePath: sessionState.storePath,
3443+
});
33203444

33213445
const dispatchDedupe = await claimMessageDispatchDedupe(event.msg);
33223446
if (!dispatchDedupe.process) {
@@ -3342,7 +3466,7 @@ export const registerTelegramHandlers = ({
33423466
sendOversizeWarning: event.sendOversizeWarning,
33433467
oversizeLogMessage: event.oversizeLogMessage,
33443468
dispatchDedupeKeys,
3345-
...promptContextBoundaryOptions(promptContextMinTimestampMs),
3469+
...promptContextBoundaryOptions(promptContextMinTimestampMs, promptContextAmbientWatermark),
33463470
});
33473471
} catch (err) {
33483472
releaseDispatchDedupeKeys(dispatchDedupeKeys, err);

0 commit comments

Comments
 (0)