| 1 | import test from "node:test"; |
| 2 | import assert from "node:assert/strict"; |
| 3 | import fs from "node:fs/promises"; |
| 4 | import path from "node:path"; |
| 5 | import { fileURLToPath } from "node:url"; |
| 6 | |
| 7 | const __dirname = path.dirname(fileURLToPath(import.meta.url)); |
| 8 | |
| 9 | async function readBridgeSource() { |
| 10 | return fs.readFile(path.join(__dirname, "../src/index.mjs"), "utf8"); |
| 11 | } |
| 12 | |
| 13 | function extractFunction(source, name) { |
| 14 | const asyncMarker = `async function ${name}`; |
| 15 | const marker = source.includes(asyncMarker) ? asyncMarker : `function ${name}`; |
| 16 | const start = source.indexOf(marker); |
| 17 | assert.notEqual(start, -1, `${name} should exist`); |
| 18 | |
| 19 | let depth = 0; |
| 20 | let opened = false; |
| 21 | const bodyStart = source.indexOf("{", source.indexOf(")", start)); |
| 22 | for (let index = bodyStart; index < source.length; index += 1) { |
| 23 | const char = source[index]; |
| 24 | if (char === "{") { |
| 25 | depth += 1; |
| 26 | opened = true; |
| 27 | } else if (char === "}") { |
| 28 | depth -= 1; |
| 29 | if (opened && depth === 0) { |
| 30 | return source.slice(start, index + 1); |
| 31 | } |
| 32 | } |
| 33 | } |
| 34 | assert.fail(`${name} body should close`); |
| 35 | } |
| 36 | |
| 37 | test("prompt command starts a tracked background turn instead of blocking update dispatch", async () => { |
| 38 | const source = await readBridgeSource(); |
| 39 | const handleCommand = extractFunction(source, "handleCommand"); |
| 40 | const promptCase = handleCommand.slice(handleCommand.indexOf('case "prompt":')); |
| 41 | |
| 42 | assert.match(source, /const activeTurnTasks = new Map\(\);/); |
| 43 | assert.match(promptCase, /startPromptTurn\(chatId, action\.prompt\);/); |
| 44 | assert.doesNotMatch(promptCase, /await\s+runPrompt\(/); |
| 45 | |
| 46 | const starter = extractFunction(source, "startPromptTurn"); |
| 47 | assert.ok( |
| 48 | starter.indexOf("activeTurnTasks.set(chatId") < starter.indexOf("void runPrompt"), |
| 49 | "turn registry entry must be installed before runPrompt can await" |
| 50 | ); |
| 51 | }); |
| 52 | |
| 53 | test("stale callback acknowledgements cannot skip modal actions", async () => { |
| 54 | const source = await readBridgeSource(); |
| 55 | const callbackHandler = extractFunction(source, "handleCallbackQuery"); |
| 56 | |
| 57 | assert.doesNotMatch(callbackHandler, /await\s+answerCallback\(query\.id,\s*"Working\.\.\."\)/); |
| 58 | assert.match(callbackHandler, /answerCallback\(query\.id,\s*"Working\.\.\."\)\.catch/); |
| 59 | assert.match(callbackHandler, /await handleModalAction\(identity\.chatId, action, query\);/); |
| 60 | }); |
| 61 | |
| 62 | test("polling persists offsets only after successful update handling", async () => { |
| 63 | const source = await readBridgeSource(); |
| 64 | const startup = source.slice( |
| 65 | source.indexOf("const threadStore = await ThreadStore.open"), |
| 66 | source.indexOf("function requestStop") |
| 67 | ); |
| 68 | const pollTelegram = extractFunction(source, "pollTelegram"); |
| 69 | const markUpdateHandled = extractFunction(source, "markUpdateHandled"); |
| 70 | |
| 71 | assert.match( |
| 72 | startup, |
| 73 | /let updateOffset = threadStore\.getCursor\(\s*"telegram\.update_offset",\s*Number\(process\.env\.TELEGRAM_UPDATE_OFFSET \|\| 0\)\s*\);/ |
| 74 | ); |
| 75 | assert.doesNotMatch(pollTelegram, /updateOffset = Math\.max\(updateOffset, update\.update_id \+ 1\)/); |
| 76 | assert.match(pollTelegram, /await handleIncomingUpdate\(update\);\s*await markUpdateHandled\(update\);/); |
| 77 | assert.match( |
| 78 | pollTelegram, |
| 79 | /catch \(error\) {\s*console\.error\("failed to handle incoming Telegram update", error\);\s*break;\s*}/ |
| 80 | ); |
| 81 | assert.match(markUpdateHandled, /const nextOffset = Math\.max\(updateOffset, Number\(update\.update_id\) \+ 1\);/); |
| 82 | assert.match(markUpdateHandled, /await threadStore\.setCursor\("telegram\.update_offset", updateOffset\);/); |
| 83 | }); |
| 84 | |
| 85 | test("polling conflicts use bounded escalation instead of a flat retry loop", async () => { |
| 86 | const source = await readBridgeSource(); |
| 87 | const pollTelegram = extractFunction(source, "pollTelegram"); |
| 88 | |
| 89 | assert.match(source, /telegramPollingConflictDelayMs/); |
| 90 | assert.match(pollTelegram, /let pollingConflictAttempts = 0;/); |
| 91 | assert.match(pollTelegram, /telegramPollingConflictDelayMs\(pollingConflictAttempts\)/); |
| 92 | assert.match(pollTelegram, /pollingConflictAttempts \+= 1;/); |
| 93 | assert.match(pollTelegram, /throw new Error\(/); |
| 94 | assert.doesNotMatch(pollTelegram, /Retrying in 10s/); |
| 95 | assert.doesNotMatch(pollTelegram, /await delay\(10000\)/); |
| 96 | }); |
| 97 | |
| 98 | test("callback replay is ignored before modal dispatch", async () => { |
| 99 | const source = await readBridgeSource(); |
| 100 | const incomingHandler = extractFunction(source, "handleIncomingUpdate"); |
| 101 | const replayHelper = extractFunction(source, "isReplayCallbackUpdate"); |
| 102 | const storedAction = extractFunction(source, "handleStoredAction"); |
| 103 | const resumeCase = storedAction.slice(storedAction.indexOf('if (stored.kind === "resume")')); |
| 104 | |
| 105 | assert.match(incomingHandler, /if \(await isReplayCallbackUpdate\(update\)\) return;\s*await handleCallbackQuery\(update\.callback_query\);/); |
| 106 | assert.match(replayHelper, /if \(update\.update_id == null\) return false;/); |
| 107 | assert.match(replayHelper, /return threadStore\.recordMessage\(`callback:\$\{update\.update_id\}`\);/); |
| 108 | assert.ok( |
| 109 | resumeCase.indexOf("await threadStore.takeAction(action.token);") < |
| 110 | resumeCase.indexOf("await resumeThread(chatId, stored.threadId);"), |
| 111 | "resume callback actions should be consumed before dispatch" |
| 112 | ); |
| 113 | }); |
| 114 | |
| 115 | test("reattached streams are detached and shutdown preserves active turn state", async () => { |
| 116 | const source = await readBridgeSource(); |
| 117 | const reattach = extractFunction(source, "reattachActiveTurns"); |
| 118 | const runPrompt = extractFunction(source, "runPrompt"); |
| 119 | |
| 120 | assert.match(reattach, /startTrackedTurnStream\(chatId, state\.threadId, turnId, sinceSeq\);/); |
| 121 | assert.doesNotMatch(reattach, /await\s+streamTurnEvents\(/); |
| 122 | assert.match(source, /async function clearActiveTurn\(chatId\)/); |
| 123 | assert.match(runPrompt, /if \(!stopping\) {\s*await clearActiveTurn\(chatId\);\s*}/); |
| 124 | |
| 125 | const trackedStream = extractFunction(source, "startTrackedTurnStream"); |
| 126 | assert.match(trackedStream, /if \(!stopping\) {\s*await clearActiveTurn\(chatId\);\s*}/); |
| 127 | }); |
| 128 | |
| 129 | test("turn update sends retry without ending the stream", async () => { |
| 130 | const source = await readBridgeSource(); |
| 131 | const streamTurnEvents = extractFunction(source, "streamTurnEvents"); |
| 132 | const sendTurnText = extractFunction(source, "sendTurnText"); |
| 133 | const telegramApi = extractFunction(source, "telegramApi"); |
| 134 | |
| 135 | assert.doesNotMatch(streamTurnEvents, /await\s+sendText\(/); |
| 136 | assert.match(streamTurnEvents, /await\s+sendTurnText\(/); |
| 137 | assert.match(sendTurnText, /catch \(error\) {\s*console\.error\("failed to send Telegram turn update"/); |
| 138 | assert.match(telegramApi, /method === "sendMessage" \? telegramSendRetryDelayMs\(error, attempt\) : null/); |
| 139 | }); |
| 140 | |
| 141 | test("turn streams keep Telegram typing visible and pause while waiting for approval", async () => { |
| 142 | const source = await readBridgeSource(); |
| 143 | const streamTurnEvents = extractFunction(source, "streamTurnEvents"); |
| 144 | const sendTypingAction = extractFunction(source, "sendTypingAction"); |
| 145 | const telegramApiOnce = extractFunction(source, "telegramApiOnce"); |
| 146 | |
| 147 | assert.match(source, /const TYPING_INTERVAL_MS = 2000;/); |
| 148 | assert.match(source, /const TYPING_TIMEOUT_MS = 1500;/); |
| 149 | assert.match(streamTurnEvents, /let typingPaused = false;/); |
| 150 | assert.match(streamTurnEvents, /let typingInFlight = false;/); |
| 151 | assert.match(streamTurnEvents, /const typingTimer = setInterval\(\(\) => {\s*void tickTyping\(\);/); |
| 152 | assert.match(streamTurnEvents, /void tickTyping\(\);/); |
| 153 | assert.match(streamTurnEvents, /const stopTypingEvent =/); |
| 154 | assert.match(streamTurnEvents, /if \(typingPaused && record\.event !== "approval\.required" && !stopTypingEvent\)/); |
| 155 | assert.match(streamTurnEvents, /typingPaused = true;/); |
| 156 | assert.match(streamTurnEvents, /clearInterval\(typingTimer\);/); |
| 157 | assert.match(sendTypingAction, /telegramApi\(\s*"sendChatAction"/); |
| 158 | assert.match(sendTypingAction, /action: "typing"/); |
| 159 | assert.match(sendTypingAction, /setTimeout\(\(\) => controller\.abort\(\), TYPING_TIMEOUT_MS\)/); |
| 160 | assert.match(telegramApiOnce, /signal: options\.signal/); |
| 161 | }); |
| 162 | |
| 163 | test("turn streams debounce last-seq writes and flush before exit", async () => { |
| 164 | const source = await readBridgeSource(); |
| 165 | const streamTurnEvents = extractFunction(source, "streamTurnEvents"); |
| 166 | const flushLastSeq = extractFunction(source, "flushLastSeq"); |
| 167 | const streamWithoutFlushHelper = streamTurnEvents.replace(flushLastSeq, ""); |
| 168 | |
| 169 | assert.match(source, /const LAST_SEQ_FLUSH_INTERVAL_MS = 2000;/); |
| 170 | assert.doesNotMatch( |
| 171 | streamWithoutFlushHelper, |
| 172 | /await threadStore\.patchChat\(chatId, \{ lastSeq: latestSeq \}\);/ |
| 173 | ); |
| 174 | assert.match(streamTurnEvents, /await flushLastSeq\(false\);/); |
| 175 | assert.match(streamTurnEvents, /await flushLastSeq\(true\);/); |
| 176 | assert.match(flushLastSeq, /if \(latestSeq <= flushedSeq\) return;/); |
| 177 | assert.match(flushLastSeq, /Date\.now\(\) - lastSeqFlushAt < LAST_SEQ_FLUSH_INTERVAL_MS/); |
| 178 | assert.match(flushLastSeq, /await threadStore\.patchChat\(chatId, \{ lastSeq: latestSeq \}\);/); |
| 179 | assert.match(flushLastSeq, /flushedSeq = latestSeq;/); |
| 180 | }); |
| 181 | |
| 182 | test("Telegram sends MarkdownV2 with plain-text fallback on parse errors", async () => { |
| 183 | const source = await readBridgeSource(); |
| 184 | const sendText = extractFunction(source, "sendText"); |
| 185 | |
| 186 | assert.match(sendText, /telegramMessageBody\(chunk, \{ markdown: true, maxChars: config\.maxReplyChars \}\)/); |
| 187 | assert.match(sendText, /isTelegramMarkdownParseError\(error\)/); |
| 188 | assert.match(sendText, /telegramMessageBody\(chunk, \{ markdown: false, maxChars: config\.maxReplyChars \}\)/); |
| 189 | assert.match(sendText, /await telegramApi\("sendMessage", fallbackBody\);/); |
| 190 | }); |
| 191 |