Skip to content

Commit fc16e15

Browse files
lidge-junlidge-jun
committed
fix(telegram): mark coalesced direct messages session-bound
1 parent aa1bc67 commit fc16e15

3 files changed

Lines changed: 144 additions & 7 deletions

File tree

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

Lines changed: 42 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -234,6 +234,30 @@ export const registerTelegramHandlers = ({
234234
const promptContextMinTimestampMs = normalizePromptContextMinTimestampMs(timestampMs);
235235
return promptContextMinTimestampMs === undefined ? {} : { promptContextMinTimestampMs };
236236
};
237+
const sourceMessageIdOptions = (
238+
messages: readonly Message[],
239+
): Pick<TelegramMessageContextOptions, "sourceMessageIds"> => {
240+
const sourceMessageIds = messages
241+
.map((msg) => (typeof msg.message_id === "number" ? String(msg.message_id) : undefined))
242+
.filter((messageId): messageId is string => Boolean(messageId));
243+
return sourceMessageIds.length > 0 ? { sourceMessageIds } : {};
244+
};
245+
const sessionBoundMessageIds = (
246+
messageId: string,
247+
sourceMessageIds?: readonly string[],
248+
): string[] => {
249+
const seen = new Set<string>();
250+
const ids: string[] = [];
251+
for (const candidate of [...(sourceMessageIds ?? []), messageId]) {
252+
const normalized = candidate.trim();
253+
if (!normalized || seen.has(normalized)) {
254+
continue;
255+
}
256+
seen.add(normalized);
257+
ids.push(normalized);
258+
}
259+
return ids;
260+
};
237261
const latestPromptContextMinTimestampMs = (
238262
...timestamps: Array<number | undefined>
239263
): number | undefined => {
@@ -445,6 +469,7 @@ export const registerTelegramHandlers = ({
445469
}
446470
if (entries.length === 1) {
447471
await processMessageWithReplyChain(last.ctx, last.msg, last.allMedia, last.storeAllowFrom, {
472+
...sourceMessageIdOptions([last.msg]),
448473
receivedAtMs: last.receivedAtMs,
449474
ingressBuffer: "inbound-debounce",
450475
...promptContextBoundaryOptions(last.promptContextMinTimestampMs),
@@ -478,6 +503,7 @@ export const registerTelegramHandlers = ({
478503
first.storeAllowFrom,
479504
{
480505
...(messageIdOverride ? { messageIdOverride } : {}),
506+
...sourceMessageIdOptions(entries.map(({ msg }) => msg)),
481507
receivedAtMs: first.receivedAtMs,
482508
ingressBuffer: "inbound-debounce",
483509
...promptContextBoundaryOptions(promptContextMinTimestampMs),
@@ -795,7 +821,10 @@ export const registerTelegramHandlers = ({
795821
primaryEntry.msg,
796822
allMedia,
797823
entry.storeAllowFrom,
798-
promptContextBoundaryOptions(entry.promptContextMinTimestampMs),
824+
{
825+
...sourceMessageIdOptions(messages.map(({ msg }) => msg)),
826+
...promptContextBoundaryOptions(entry.promptContextMinTimestampMs),
827+
},
799828
);
800829
} catch (err) {
801830
runtime.error?.(danger(`media group handler failed: ${String(err)}`));
@@ -829,6 +858,7 @@ export const registerTelegramHandlers = ({
829858
const syntheticCtx = buildSyntheticContext(baseCtx, syntheticMessage);
830859
await processMessageWithReplyChain(syntheticCtx, syntheticMessage, [], storeAllowFrom, {
831860
messageIdOverride: String(last.msg.message_id),
861+
...sourceMessageIdOptions(messages.map(({ msg }) => msg)),
832862
receivedAtMs: first.receivedAtMs,
833863
ingressBuffer: "text-fragment",
834864
...promptContextBoundaryOptions(entry.promptContextMinTimestampMs),
@@ -1010,12 +1040,17 @@ export const registerTelegramHandlers = ({
10101040
options?.resolvePromptContext?.(params) ??
10111041
buildPromptContextForMessage(msg, replyChainNodes, options, params.sessionKey),
10121042
markSessionBoundMessage: (params) => {
1013-
messageCache.markSessionBound({
1014-
accountId,
1015-
chatId: msg.chat.id,
1016-
messageId: params.messageId,
1017-
sessionKey: params.sessionKey,
1018-
});
1043+
for (const messageId of sessionBoundMessageIds(
1044+
params.messageId,
1045+
options?.sourceMessageIds,
1046+
)) {
1047+
messageCache.markSessionBound({
1048+
accountId,
1049+
chatId: msg.chat.id,
1050+
messageId,
1051+
sessionKey: params.sessionKey,
1052+
});
1053+
}
10191054
options?.markSessionBoundMessage?.(params);
10201055
},
10211056
};

‎extensions/telegram/src/bot-message-context.types.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ export type TelegramMessageContextOptions = {
2121
commandSource?: "text" | "native";
2222
forceWasMentioned?: boolean;
2323
messageIdOverride?: string;
24+
sourceMessageIds?: string[];
2425
receivedAtMs?: number;
2526
ingressBuffer?: "inbound-debounce" | "text-fragment";
2627
promptContextMinTimestampMs?: number;

‎extensions/telegram/src/bot.test.ts‎

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1815,6 +1815,107 @@ describe("createTelegramBot", () => {
18151815
expect(payload.UntrustedStructuredContext).toBeUndefined();
18161816
});
18171817

1818+
it("marks every debounced direct source message as session-bound", async () => {
1819+
const DEBOUNCE_MS = 4321;
1820+
onSpy.mockClear();
1821+
replySpy.mockClear();
1822+
loadConfig.mockReturnValue({
1823+
agents: {
1824+
defaults: {
1825+
envelopeTimezone: "utc",
1826+
},
1827+
},
1828+
messages: {
1829+
inbound: {
1830+
debounceMs: DEBOUNCE_MS,
1831+
},
1832+
},
1833+
channels: {
1834+
telegram: { dmPolicy: "open", allowFrom: ["*"] },
1835+
},
1836+
});
1837+
1838+
const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
1839+
try {
1840+
const flushLastDebounceTimer = async () => {
1841+
const flushTimerCallIndex = setTimeoutSpy.mock.calls.findLastIndex(
1842+
(call) => call[1] === DEBOUNCE_MS,
1843+
);
1844+
const flushTimer =
1845+
flushTimerCallIndex >= 0
1846+
? (setTimeoutSpy.mock.calls[flushTimerCallIndex]?.[0] as (() => unknown) | undefined)
1847+
: undefined;
1848+
if (flushTimerCallIndex >= 0) {
1849+
clearTimeout(
1850+
setTimeoutSpy.mock.results[flushTimerCallIndex]?.value as ReturnType<typeof setTimeout>,
1851+
);
1852+
}
1853+
expect(flushTimer).toBeTypeOf("function");
1854+
await flushTimer?.();
1855+
};
1856+
1857+
const firstReply = waitForReplyCalls(1);
1858+
createTelegramBot({ token: "tok" });
1859+
const handler = getOnHandler("message") as (ctx: Record<string, unknown>) => Promise<void>;
1860+
const baseCtx = {
1861+
me: { id: 999, username: "openclaw_bot" },
1862+
getFile: async () => ({ download: async () => new Uint8Array() }),
1863+
};
1864+
const chat = { id: 42, type: "private", first_name: "Ada" };
1865+
const from = { id: 42, is_bot: false, first_name: "Ada" };
1866+
1867+
await handler({
1868+
...baseCtx,
1869+
message: {
1870+
chat,
1871+
text: "first buffered direct message",
1872+
date: 1736380200,
1873+
message_id: 400,
1874+
from,
1875+
},
1876+
});
1877+
await handler({
1878+
...baseCtx,
1879+
message: {
1880+
chat,
1881+
text: "second buffered direct message",
1882+
date: 1736380201,
1883+
message_id: 401,
1884+
from,
1885+
},
1886+
});
1887+
expect(replySpy).not.toHaveBeenCalled();
1888+
await flushLastDebounceTimer();
1889+
await firstReply;
1890+
1891+
replySpy.mockClear();
1892+
const secondReply = waitForReplyCalls(1);
1893+
await handler({
1894+
...baseCtx,
1895+
message: {
1896+
chat,
1897+
text: "what did I just send?",
1898+
date: 1736380300,
1899+
message_id: 402,
1900+
from,
1901+
},
1902+
});
1903+
await flushLastDebounceTimer();
1904+
await secondReply;
1905+
1906+
expect(replySpy).toHaveBeenCalledTimes(1);
1907+
const payload = mockMsgContextArg(
1908+
replySpy as unknown as MockCallSource,
1909+
0,
1910+
0,
1911+
"replySpy call",
1912+
);
1913+
expect(payload.UntrustedStructuredContext).toBeUndefined();
1914+
} finally {
1915+
setTimeoutSpy.mockRestore();
1916+
}
1917+
});
1918+
18181919
it("omits stale Telegram topic context before the persisted session start", async () => {
18191920
onSpy.mockClear();
18201921
replySpy.mockClear();

0 commit comments

Comments
 (0)