feat(standalone): :integrations-telegram — bridge stability + markdown rendering
Outbound (DurableEvent-driven): - typing дёргается ДО conv.send (а не только по online-событиям); рефреш по online — каждые ≥4 с, чтобы не получить 429-спам (Telegram rate-limit ~250 ms на sendChatAction). - финальный текст — sendMessage(... ParseMode.HTML) на DurableEvent.AssistantMessage, без промежуточных editMessage. - служебные строки ((empty response), ошибки) идут как plain text (без parseMode=HTML), чтобы 400 can't parse entities не выбил предупреждение в чат. Inline markdown → Telegram HTML (MarkdownToTelegram): - **bold** / __bold__ / *italic* / _italic_ / `code` → <b>/<i>/<code>; fenced ``` → <pre>; [text](url) → <a href>. - markdown-таблицы → Unicode-рамка (┌─┬─┐││├─┼─┤└─┴─┘) внутри <pre> (Telegram Bot API <table> не поддерживает). - escapeOutsideTags теперь ведёт стек открытых тегов: раньше брал первый попавшийся закрывающий тег и терял парность (агент вставил <i> внутри <code> → Telegram 400 can't find end tag для <code>). - экранирует только вне наших тегов; внутри — уже подготовленный текст. - сообщения > 4096 режутся по \n вне тегов. Видимость ошибок: - kotlin-logging в build.gradle.kts. - runCatching'и заменены на try/catch + log.error/warn с контекстом (chatId, convId, htmlLen, превью первых ≤120 символов HTML). Раньше Telegram 400 на sendMessage с 'can't parse entities' уходил молча — теперь видно в логе с полным телом запроса. Тесты: 23 + 12 + 28 = 63 в :integrations-telegram, 132 в :standalone. :standalone:assemble собирает fatjar.
This commit is contained in:
@@ -54,6 +54,8 @@ dependencies {
|
||||
implementation(libs.kotlinx.serialization.json)
|
||||
|
||||
implementation(libs.ksqlite)
|
||||
implementation(libs.kotlin.logging)
|
||||
implementation(libs.logback.classic)
|
||||
}
|
||||
|
||||
dependencies {
|
||||
|
||||
+558
@@ -0,0 +1,558 @@
|
||||
package pw.binom.agentik.integrations.telegram
|
||||
|
||||
/**
|
||||
* Минимальный CommonMark → Telegram-HTML конвертер.
|
||||
*
|
||||
* Зачем вручную, а не через библиотеку:
|
||||
* - `ru.otpbank.ai:markdown:0.51.0` есть в Nexus, но он тащит тяжёлое
|
||||
* AST и не умеет эмитить Telegram-совместимый HTML (там другая модель
|
||||
* тегов + нужно экранирование `<>&` вне тегов).
|
||||
* - LLM-ответы используют узкий набор конструкций; достаточно покрыть
|
||||
* их без полного CommonMark.
|
||||
*
|
||||
* Поддерживаемое:
|
||||
* - `**bold**` / `__bold__` → `<b>bold</b>`
|
||||
* - `*italic*` / `_italic_` → `<i>italic</i>`
|
||||
* - `` `inline code` `` → `<code>inline code</code>`
|
||||
* - ``` ```code block``` ``` → `<pre>code block</pre>`
|
||||
* - `[text](url)` → `<a href="url">text</a>`
|
||||
* - `# … ######` (заголовки) → `<b>…</b>` (TG-HTML не имеет заголовков)
|
||||
* - `- item` / `* item` / `1. item` → `• item` / `1. item`
|
||||
* - `> quote` → отступ `│ quote`
|
||||
*
|
||||
* Что **не** поддерживается (намеренно): html-блоки, footnotes,
|
||||
* images внутри md (`` идёт в plain text).
|
||||
*
|
||||
* **Таблицы**: Telegram Bot API не поддерживает `<table>`/`<tr>`/`<td>` —
|
||||
* рендерим через Unicode box-drawing (`┌─┐││├─┤┴─┘┼`) внутри `<pre>`,
|
||||
* чтобы рамка и колонки выровнялись моноширинно. Markdown-маркеры
|
||||
* (`**`, `*`, `` ` ``) и HTML-теги внутри ячеек выкидываются — в `<pre>`
|
||||
* inline-форматирование не работает.
|
||||
*
|
||||
* Экранирование: после конвертации все `<`, `>`, `&` и `"`
|
||||
* (внутри атрибутов), не входящие в наши собственные теги, заменяются на
|
||||
* `<` / `>` / `&` / `"`. Telegram иначе ругается
|
||||
* `400 Bad Request: can't parse entities`.
|
||||
*/
|
||||
object MarkdownToTelegram {
|
||||
|
||||
/**
|
||||
* Конвертирует [markdown] в Telegram-HTML. Возвращает `null`, если
|
||||
* после конвертации текст не содержит HTML-тегов (Telegram тогда
|
||||
* проще отдать как plain — нет риска ругани на экранирование).
|
||||
*
|
||||
* @return `Pair(html, containsAnyTag)` — в мосте используется первая
|
||||
* компонента; вторая пригодится, если захотим выбирать parseMode
|
||||
* динамически (сейчас не используется, оставлено для теста).
|
||||
*/
|
||||
fun convert(markdown: String): String {
|
||||
val lines = markdown.lines()
|
||||
val out = StringBuilder()
|
||||
var i = 0
|
||||
var first = true
|
||||
while (i < lines.size) {
|
||||
val line = lines[i]
|
||||
// Блок кода: открывающие ``` до закрывающих ```.
|
||||
if (line.trimStart().startsWith("```")) {
|
||||
val fence = line.trimStart().takeWhile { it == '`' }
|
||||
val sb = StringBuilder()
|
||||
i++
|
||||
while (i < lines.size && !lines[i].trimStart().startsWith(fence)) {
|
||||
if (sb.isNotEmpty()) sb.append('\n')
|
||||
sb.append(lines[i])
|
||||
i++
|
||||
}
|
||||
if (i < lines.size) i++ // skip closing fence
|
||||
// Язык после ``` отбрасываем (TG-pre его не показывает).
|
||||
if (!first) out.append('\n')
|
||||
out.append("<pre>").append(escapeInsideTag(sb.toString())).append("</pre>")
|
||||
first = false
|
||||
continue
|
||||
}
|
||||
// Заголовок.
|
||||
val headingMatch = HEADING.matchEntire(line)
|
||||
if (headingMatch != null) {
|
||||
if (!first) out.append('\n')
|
||||
out.append("<b>").append(escapeOutsideTags(headingMatch.groupValues[1])).append("</b>")
|
||||
first = false
|
||||
i++
|
||||
continue
|
||||
}
|
||||
// Markdown-таблица: строка заголовка + строка-разделитель + N
|
||||
// строк данных. Все должны начинаться с `|`. Рендерим
|
||||
// Unicode box-drawing внутри <pre>, чтобы моноширинно
|
||||
// выровнялось.
|
||||
val table = tryRenderTable(lines, i)
|
||||
if (table != null) {
|
||||
if (!first) out.append('\n')
|
||||
// escapeInsideTag на случай, если stripCellMarkup пропустил
|
||||
// какой-то `<`/`>`/`&` (защита от 400 can't parse entities).
|
||||
out.append("<pre>").append(escapeInsideTag(table.html)).append("</pre>")
|
||||
first = false
|
||||
i += table.linesConsumed
|
||||
continue
|
||||
}
|
||||
// Block quote.
|
||||
if (line.startsWith("> ")) {
|
||||
if (!first) out.append('\n')
|
||||
out.append("│ ").append(convertInline(line.removePrefix("> ")))
|
||||
first = false
|
||||
i++
|
||||
continue
|
||||
}
|
||||
if (line.startsWith(">")) {
|
||||
if (!first) out.append('\n')
|
||||
out.append("│ ").append(convertInline(line.removePrefix(">")))
|
||||
first = false
|
||||
i++
|
||||
continue
|
||||
}
|
||||
// Список.
|
||||
val bullet = BULLET.matchEntire(line)
|
||||
if (bullet != null) {
|
||||
if (!first) out.append('\n')
|
||||
out.append("• ").append(convertInline(bullet.groupValues[1]))
|
||||
first = false
|
||||
i++
|
||||
continue
|
||||
}
|
||||
val ordered = ORDERED.matchEntire(line)
|
||||
if (ordered != null) {
|
||||
if (!first) out.append('\n')
|
||||
out.append(ordered.groupValues[1]).append(". ").append(convertInline(ordered.groupValues[2]))
|
||||
first = false
|
||||
i++
|
||||
continue
|
||||
}
|
||||
// Обычная строка.
|
||||
if (!first) out.append('\n')
|
||||
out.append(convertInline(line))
|
||||
first = false
|
||||
i++
|
||||
}
|
||||
return out.toString().trimEnd('\n')
|
||||
}
|
||||
|
||||
/** Инлайн-уровень: ссылки, жирный, курсив, код. */
|
||||
private fun convertInline(text: String): String {
|
||||
if (text.isEmpty()) return ""
|
||||
var s = text
|
||||
// 1. Инлайн-код: берём самый длинный непересекающийся диапазон.
|
||||
s = applyInlineCode(s)
|
||||
// 2. Ссылки [text](url).
|
||||
s = applyLinks(s)
|
||||
// 3. Жирный (`**` и `__`).
|
||||
s = applyBold(s)
|
||||
// 4. Курсив (`*` и `_`).
|
||||
s = applyItalic(s)
|
||||
// 5. Экранируем всё, что осталось.
|
||||
return escapeOutsideTags(s)
|
||||
}
|
||||
|
||||
private fun applyInlineCode(s: String): String {
|
||||
val out = StringBuilder()
|
||||
var i = 0
|
||||
while (i < s.length) {
|
||||
val tick = s.indexOf('`', i)
|
||||
if (tick < 0) {
|
||||
out.append(s, i, s.length)
|
||||
break
|
||||
}
|
||||
out.append(s, i, tick)
|
||||
val end = s.indexOf('`', tick + 1)
|
||||
if (end < 0) {
|
||||
// Без закрывающего — оставляем как есть.
|
||||
out.append('`')
|
||||
i = tick + 1
|
||||
continue
|
||||
}
|
||||
out.append("<code>")
|
||||
.append(escapeInsideTag(s.substring(tick + 1, end)))
|
||||
.append("</code>")
|
||||
i = end + 1
|
||||
}
|
||||
return out.toString()
|
||||
}
|
||||
|
||||
private fun applyLinks(s: String): String {
|
||||
// Простой [text](url) — без балансировки скобок в URL.
|
||||
val regex = Regex("""\[([^\]]+)]\(([^)\s]+)\)""")
|
||||
return regex.replace(s) { mr ->
|
||||
val text = mr.groupValues[1]
|
||||
val url = mr.groupValues[2]
|
||||
"<a href=\"${escapeAttr(url)}\">${escapeOutsideTags(text)}</a>"
|
||||
}
|
||||
}
|
||||
|
||||
private fun applyBold(s: String): String {
|
||||
// Сначала **, потом __, чтобы не съесть одиночные *.
|
||||
var out = wrapPairs(s, "**", "<b>", "</b>")
|
||||
out = wrapPairs(out, "__", "<b>", "</b>")
|
||||
return out
|
||||
}
|
||||
|
||||
private fun applyItalic(s: String): String {
|
||||
var out = wrapPairs(s, "*", "<i>", "</i>")
|
||||
out = wrapPairs(out, "_", "<i>", "</i>")
|
||||
return out
|
||||
}
|
||||
|
||||
/**
|
||||
* Обёртка парных маркеров в теги. Защищается от:
|
||||
* - маркера внутри уже-HTML-тега (тогда маркер пропускается);
|
||||
* - непарного одиночного маркера (тогда оставляем как есть).
|
||||
*/
|
||||
private fun wrapPairs(s: String, marker: String, openTag: String, closeTag: String): String {
|
||||
val out = StringBuilder()
|
||||
var i = 0
|
||||
while (i <= s.length) {
|
||||
val open = s.indexOf(marker, i)
|
||||
if (open < 0) {
|
||||
out.append(s, i, s.length)
|
||||
break
|
||||
}
|
||||
// Уже внутри нашего тега? (эвристика: до маркера нет открывающего <...>)
|
||||
// Проще: проверяем, что маркер не внутри уже-выданного тега.
|
||||
// Считаем число открытых тегов до позиции.
|
||||
if (insideOurTag(s, open)) {
|
||||
out.append(s, i, open + marker.length)
|
||||
i = open + marker.length
|
||||
continue
|
||||
}
|
||||
val close = s.indexOf(marker, open + marker.length)
|
||||
if (close < 0) {
|
||||
out.append(s, i, s.length)
|
||||
break
|
||||
}
|
||||
out.append(s, i, open)
|
||||
.append(openTag)
|
||||
.append(s, open + marker.length, close)
|
||||
.append(closeTag)
|
||||
i = close + marker.length
|
||||
}
|
||||
return out.toString()
|
||||
}
|
||||
|
||||
private fun insideOurTag(s: String, pos: Int): Boolean {
|
||||
// Считаем число "<b>" / "<i>" / "<code>" / "<pre>" / "<a " до pos без
|
||||
// закрывающих. Строго говоря, нужно считать по последнему тегу, но
|
||||
// для нашего конвертера вход строго сегрегирован (теги выдаются
|
||||
// парами), и внутри их текста мы больше ничего не вставляем.
|
||||
var i = 0
|
||||
var depth = 0
|
||||
while (i < pos) {
|
||||
val openIdx = indexOfAnyTagOpen(s, i)
|
||||
val closeIdx = indexOfAnyTagClose(s, i)
|
||||
if (closeIdx >= 0 && (openIdx < 0 || closeIdx < openIdx)) {
|
||||
depth--
|
||||
i = closeIdx + 1
|
||||
continue
|
||||
}
|
||||
if (openIdx in i until pos) {
|
||||
depth++
|
||||
i = openIdx + 1
|
||||
continue
|
||||
}
|
||||
break
|
||||
}
|
||||
return depth > 0
|
||||
}
|
||||
|
||||
private fun indexOfAnyTagOpen(s: String, from: Int): Int {
|
||||
// Поиск ближайшего из известных открывающих.
|
||||
val a = s.indexOf("<b>", from)
|
||||
val b = s.indexOf("<i>", from)
|
||||
val c = s.indexOf("<code>", from)
|
||||
val d = s.indexOf("<pre>", from)
|
||||
val e = s.indexOf("<a ", from)
|
||||
return listOf(a, b, c, d, e).filter { it >= 0 }.minOrNull() ?: -1
|
||||
}
|
||||
|
||||
private fun indexOfAnyTagClose(s: String, from: Int): Int {
|
||||
val a = s.indexOf("</b>", from)
|
||||
val b = s.indexOf("</i>", from)
|
||||
val c = s.indexOf("</code>", from)
|
||||
val d = s.indexOf("</pre>", from)
|
||||
val e = s.indexOf("</a>", from)
|
||||
return listOf(a, b, c, d, e).filter { it >= 0 }.minOrNull() ?: -1
|
||||
}
|
||||
|
||||
/**
|
||||
* Экранирование текста ВНЕ наших HTML-тегов: `<`, `>`, `&` → сущности.
|
||||
* Telegram иначе отвечает 400 `can't parse entities`.
|
||||
*
|
||||
* Сканирует строку посимвольно со стеком открытых тегов. Когда встречаем
|
||||
* открывающий тег — пушим в стек. Когда встречаем закрывающий:
|
||||
* - если имя совпадает с вершиной стека — поп, копируем тег as-is;
|
||||
* - если имя НЕ совпадает — экранируем (это либо мусор, либо
|
||||
* непарный HTML от LLM);
|
||||
* - если стек пуст — экранируем.
|
||||
*
|
||||
* Любые `<`, `>` и `&` ВНЕ тегов экранируются, ВНУТРИ тегов
|
||||
* сохраняются как есть (предполагается, что содержимое тегов уже
|
||||
* подготовлено `applyBold/Italic/InlineCode/Links`, либо LLM вставил
|
||||
* валидный HTML, который Telegram понимает).
|
||||
*
|
||||
* Если LLM написал `<code>` без парного `</code>`, лишний литерал
|
||||
* экранируется — Telegram не получит незакрытый тег и не вернёт 400.
|
||||
*/
|
||||
fun escapeOutsideTags(s: String): String {
|
||||
val out = StringBuilder()
|
||||
val stack = ArrayDeque<String>() // имена открытых тегов
|
||||
var i = 0
|
||||
var plainStart = 0 // начало участка plain-текста для экранирования
|
||||
fun flushEscaped(to: Int) {
|
||||
if (to > plainStart) appendEscaped(out, s, plainStart, to)
|
||||
plainStart = to
|
||||
}
|
||||
while (i < s.length) {
|
||||
// Проверяем, не начинается ли тут открывающий тег.
|
||||
val openMatch = matchAnyOpen(s, i)
|
||||
if (openMatch != null) {
|
||||
flushEscaped(i)
|
||||
// Для <a> тег продолжается до ближайшего '>' (там атрибуты
|
||||
// href=..., которые applyLinks уже прогнал через escapeAttr).
|
||||
val openEnd = if (openMatch.kind == "a") {
|
||||
val closeGt = s.indexOf('>', i)
|
||||
if (closeGt < 0) s.length else closeGt + 1
|
||||
} else {
|
||||
i + openMatch.text.length
|
||||
}
|
||||
out.append(s, i, openEnd)
|
||||
stack.addLast(openMatch.kind)
|
||||
i = openEnd
|
||||
plainStart = i
|
||||
continue
|
||||
}
|
||||
// Закрывающий тег?
|
||||
val closeKind = matchAnyClose(s, i)
|
||||
if (closeKind != null) {
|
||||
if (stack.isNotEmpty() && stack.last() == closeKind) {
|
||||
// Копируем plain-текст внутри тега as-is (он уже
|
||||
// экранирован applyLinks/escapeInsideTag/escapeAttr) и
|
||||
// сам закрывающий тег.
|
||||
out.append(s, plainStart, i)
|
||||
val closeEnd = s.indexOf('>', i).let { if (it < 0) s.length else it + 1 }
|
||||
out.append(s, i, closeEnd)
|
||||
stack.removeLast()
|
||||
i = closeEnd
|
||||
plainStart = i
|
||||
continue
|
||||
}
|
||||
// Непарный — пусть экранируется как обычный текст.
|
||||
i += 1
|
||||
continue
|
||||
}
|
||||
i += 1
|
||||
}
|
||||
// Хвост после последнего закрывающего тега экранируем.
|
||||
if (stack.isEmpty()) flushEscaped(s.length)
|
||||
else out.append(s, plainStart, s.length)
|
||||
return out.toString()
|
||||
}
|
||||
|
||||
/** Находит ближайший открывающий тег, начинающийся с [from]. */
|
||||
private fun matchAnyOpen(s: String, from: Int): TagMatch? {
|
||||
if (s.length - from < 3) return null
|
||||
for ((text, kind) in OPENS) {
|
||||
if (s.regionMatches(from, text, 0, text.length)) return TagMatch(text, kind)
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
/** Находит ближайший закрывающий тег, начинающийся с [from]. Возвращает имя (b, i, code, ...). */
|
||||
private fun matchAnyClose(s: String, from: Int): String? {
|
||||
if (s.length - from < 4) return null
|
||||
if (s[from] != '<') return null
|
||||
if (s[from + 1] != '/') return null
|
||||
for (kind in CLOSE_KINDS) {
|
||||
val text = "</$kind>"
|
||||
if (s.regionMatches(from, text, 0, text.length)) return kind
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
private data class TagMatch(val text: String, val kind: String)
|
||||
|
||||
/** Все наши открывающие теги. */
|
||||
private val OPENS: List<Pair<String, String>> = listOf(
|
||||
"<b>" to "b",
|
||||
"<i>" to "i",
|
||||
"<code>" to "code",
|
||||
"<pre>" to "pre",
|
||||
"<a " to "a",
|
||||
)
|
||||
|
||||
/** Все имена закрывающих тегов (для matchAnyClose). */
|
||||
private val CLOSE_KINDS: List<String> = listOf("b", "i", "code", "pre", "a")
|
||||
|
||||
private fun appendEscaped(out: StringBuilder, s: String, from: Int, to: Int) {
|
||||
for (j in from until to) {
|
||||
when (s[j]) {
|
||||
'<' -> out.append("<")
|
||||
'>' -> out.append(">")
|
||||
'&' -> out.append("&")
|
||||
else -> out.append(s[j])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Экранирование внутри содержимого `<code>`/`<pre>` (только `<`, `>`, `&`). */
|
||||
fun escapeInsideTag(s: String): String =
|
||||
s.replace("&", "&").replace("<", "<").replace(">", ">")
|
||||
|
||||
/** Экранирование внутри значения атрибута (`href`). */
|
||||
fun escapeAttr(s: String): String =
|
||||
s.replace("&", "&").replace("\"", """).replace("<", "<")
|
||||
|
||||
private val HEADING = Regex("""^#{1,6}\s+(.+)$""")
|
||||
private val BULLET = Regex("""^[-*]\s+(.+)$""")
|
||||
private val ORDERED = Regex("""^(\d+)\.\s+(.+)$""")
|
||||
|
||||
// Markdown-таблица: «| --- | :---: | ---: |» (двоеточия — выравнивание,
|
||||
// которое мы игнорируем — рендерим всегда left).
|
||||
private val TABLE_SEPARATOR = Regex("""^\s*\|?\s*:?-{2,}:?\s*(\|\s*:?-{2,}:?\s*)+\|?\s*$""")
|
||||
// Строка данных: начинается с `|`, должна иметь хотя бы один `|`
|
||||
// внутри. Финальный `|` опционален — некоторые LLM его теряют на
|
||||
// последней строке.
|
||||
private val TABLE_ROW = Regex("""^\s*\|[^|\n]*(\|[^|\n]*)+\|?\s*$""")
|
||||
|
||||
private data class TableResult(val html: String, val linesConsumed: Int)
|
||||
|
||||
/**
|
||||
* Пытается распарсить markdown-таблицу, начинающуюся со строки
|
||||
* [lines][[from]]. Возвращает `null`, если [from] не похож на
|
||||
* заголовок таблицы (нет `|`-разделителя на следующей строке).
|
||||
*/
|
||||
private fun tryRenderTable(lines: List<String>, from: Int): TableResult? {
|
||||
val headerLine = lines[from]
|
||||
if (!headerLine.contains('|')) return null
|
||||
if (from + 1 >= lines.size) return null
|
||||
if (!TABLE_SEPARATOR.matches(lines[from + 1])) return null
|
||||
|
||||
// Парсим заголовок.
|
||||
val headers = parseTableRow(headerLine)
|
||||
val columnCount = headers.size
|
||||
if (columnCount == 0) return null
|
||||
|
||||
// Парсим строки данных, пока они выглядят как табличные.
|
||||
val rows = mutableListOf<List<String>>()
|
||||
var i = from + 2
|
||||
while (i < lines.size) {
|
||||
val l = lines[i]
|
||||
if (l.isBlank()) break
|
||||
if (!TABLE_ROW.matches(l)) break
|
||||
val cells = parseTableRow(l)
|
||||
// Добиваем до columnCount пустыми ячейками.
|
||||
val padded = if (cells.size < columnCount) {
|
||||
cells + List(columnCount - cells.size) { "" }
|
||||
} else cells.take(columnCount)
|
||||
rows.add(padded)
|
||||
i++
|
||||
}
|
||||
|
||||
// Чистим markdown/HTML-разметку в ячейках (внутри <pre> ничего не
|
||||
// рендерится, только моноширинный текст).
|
||||
val cleanHeaders = headers.map { stripCellMarkup(it) }
|
||||
val cleanRows = rows.map { row -> row.map { stripCellMarkup(it) } }
|
||||
|
||||
// Ширины столбцов — максимум по заголовку и строкам.
|
||||
val widths = IntArray(columnCount)
|
||||
for (c in 0 until columnCount) {
|
||||
widths[c] = cleanHeaders[c].length
|
||||
for (r in cleanRows) {
|
||||
if (r[c].length > widths[c]) widths[c] = r[c].length
|
||||
}
|
||||
}
|
||||
|
||||
return TableResult(renderTableBox(cleanHeaders, cleanRows, widths), i - from)
|
||||
}
|
||||
|
||||
/** Разбивает строку таблицы `| a | b | c |` на список ячеек. */
|
||||
private fun parseTableRow(line: String): List<String> {
|
||||
// Trim внешних `|`, потом split по `|` с trim.
|
||||
var s = line.trim()
|
||||
if (s.startsWith("|")) s = s.substring(1)
|
||||
if (s.endsWith("|") && s.length > 1) s = s.substring(0, s.length - 1)
|
||||
return s.split('|').map { it.trim() }
|
||||
}
|
||||
|
||||
/**
|
||||
* Чистит markdown/HTML-разметку в ячейке таблицы для рендера в
|
||||
* `<pre>`. Порядок: убираем HTML-теги, потом парные markdown-маркеры,
|
||||
* потом одиночные, потом тильду strikethrough.
|
||||
*/
|
||||
private fun stripCellMarkup(s: String): String {
|
||||
var out = s
|
||||
// 1. HTML-теги: `<...>`.
|
||||
out = HTML_TAG.replace(out, "")
|
||||
// 2. Inline-код — оставляем текст, без маркеров.
|
||||
out = out.replace('`', ' ')
|
||||
// 3. **bold** → text, __bold__ → text.
|
||||
out = BOLD_PAIR.replace(out) { mr -> mr.groupValues[1] }
|
||||
// 4. *italic* / _italic_ → text.
|
||||
out = ITALIC_PAIR.replace(out) { mr -> mr.groupValues[1] }
|
||||
// 5. ~~strike~~ → text.
|
||||
out = out.replace("~~", "")
|
||||
// 6. [text](url) → text.
|
||||
out = LINK.replace(out) { mr -> mr.groupValues[1] }
|
||||
// 7. # в начале строки (заголовок внутри ячейки) — оставляем как есть.
|
||||
// Лишние пробелы схлопываем.
|
||||
return out.replace(Regex("""\s+"""), " ").trim()
|
||||
}
|
||||
|
||||
/**
|
||||
* Рендерит таблицу через Unicode box-drawing. Каждая ячейка
|
||||
* выравнивается по левому краю, ширина = максимум длин ячеек в
|
||||
* столбце. Рамка: ┌─┬─┐ / │ │ │ / ├─┼─┤ / └─┴─┘.
|
||||
*/
|
||||
private fun renderTableBox(headers: List<String>, rows: List<List<String>>, widths: IntArray): String {
|
||||
val sb = StringBuilder()
|
||||
val topBorder = buildHorizontal("┌", "┬", "┐", "─", widths)
|
||||
val midBorder = buildHorizontal("├", "┼", "┤", "─", widths)
|
||||
val botBorder = buildHorizontal("└", "┴", "┘", "─", widths)
|
||||
val sepBorder = buildHorizontal("├", "┼", "┤", "─", widths)
|
||||
|
||||
sb.append(topBorder).append('\n')
|
||||
sb.append(buildRow(headers, widths)).append('\n')
|
||||
if (rows.isNotEmpty()) {
|
||||
sb.append(sepBorder).append('\n')
|
||||
for ((idx, row) in rows.withIndex()) {
|
||||
sb.append(buildRow(row, widths))
|
||||
if (idx != rows.lastIndex) sb.append('\n')
|
||||
}
|
||||
} else {
|
||||
// Даже без строк данных покажем «визуальный» разделитель под
|
||||
// заголовком — так Telegram-юзеру понятно, что это таблица.
|
||||
sb.append(midBorder)
|
||||
}
|
||||
sb.append('\n').append(botBorder)
|
||||
return sb.toString()
|
||||
}
|
||||
|
||||
private fun buildHorizontal(left: String, mid: String, right: String, fill: String, widths: IntArray): String {
|
||||
val sb = StringBuilder()
|
||||
sb.append(left)
|
||||
for ((idx, w) in widths.withIndex()) {
|
||||
repeat(w + 2) { sb.append(fill) }
|
||||
sb.append(if (idx == widths.lastIndex) right else mid)
|
||||
}
|
||||
return sb.toString()
|
||||
}
|
||||
|
||||
private fun buildRow(cells: List<String>, widths: IntArray): String {
|
||||
val sb = StringBuilder()
|
||||
sb.append('│')
|
||||
for ((idx, cell) in cells.withIndex()) {
|
||||
sb.append(' ').append(cell).append(' ')
|
||||
repeat(widths[idx] - cell.length) { sb.append(' ') }
|
||||
sb.append('│')
|
||||
}
|
||||
return sb.toString()
|
||||
}
|
||||
|
||||
private val HTML_TAG = Regex("""</?[A-Za-z][A-Za-z0-9]*[^>]*>""")
|
||||
private val BOLD_PAIR = Regex("""\*\*([^*]+)\*\*""")
|
||||
private val ITALIC_PAIR = Regex("""\*([^*]+)\*""")
|
||||
private val LINK = Regex("""\[([^\]]+)]\(([^)\s]+)\)""")
|
||||
}
|
||||
+343
-150
@@ -4,62 +4,67 @@ import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CoroutineName
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||
import kotlinx.coroutines.isActive
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import mu.KotlinLogging
|
||||
import pw.binom.agentik.agent.Component
|
||||
import pw.binom.agentik.agent.MutableAgent
|
||||
import pw.binom.agentik.content.Content
|
||||
import pw.binom.agentik.outbox.OnlineEvent
|
||||
import pw.binom.agentik.outbox.DurableEvent
|
||||
import pw.binom.telegram.TelegramClient
|
||||
import pw.binom.telegram.dto.EditTextRequest
|
||||
import pw.binom.telegram.dto.ParseMode
|
||||
import pw.binom.telegram.dto.SendChatEvent
|
||||
import pw.binom.telegram.dto.TextMessage
|
||||
import pw.binom.telegram.dto.Update
|
||||
|
||||
/**
|
||||
* Мост между [TelegramClient] и [MutableAgent]:
|
||||
* Мост между [TelegramClient] и [MutableAgent].
|
||||
*
|
||||
* - **Inbound**: long-poll `getUpdates` → текстовое сообщение TG
|
||||
* резолвится/создаёт [pw.binom.agentik.proto.Conversation] (по
|
||||
* `TelegramChatMap`) и передаётся в `conv.send(listOf(Content.Text(...)))`.
|
||||
* - **Outbound**: для каждого активного диалога подписывается на
|
||||
* `agent.onlineOutbox.onlineEvents(convId)`; на `StartResponse` шлёт
|
||||
* в TG «…», по `AppendText` правит его накопленным текстом, на `End`
|
||||
* оставляет финальное состояние.
|
||||
* ## Inbound
|
||||
* Long-poll `getUpdates?timeout=N` → текстовое сообщение резолвит
|
||||
* (или создаёт) [pw.binom.agentik.proto.Conversation] по [TelegramChatMap]
|
||||
* и отправляет его в `conv.send(listOf(Content.Text(...)))`.
|
||||
*
|
||||
* Компонент НЕ владеет [TelegramClient]: цикл жизни клиента отдаётся
|
||||
* вызывающему (в [Main][pw.binom.agentik.standalone.Main] клиент открывается
|
||||
* вместе с `TelegramBridgeComponent`, закрывается в shutdown-hook'е перед
|
||||
* `agent.close()` — см. порядок отмены там). Это даёт вызывающему полный
|
||||
* контроль над `HttpClient`-движком (чтобы переиспользовать существующий
|
||||
* Ktor-инстанс, если он есть).
|
||||
* ## Outbound
|
||||
* Для каждого активного диалога поднимается один subscriber на `convId`,
|
||||
* слушающий **оба** outbox'а параллельно:
|
||||
* - `agent.onlineOutbox.onlineEvents(convId)` — на каждом событии
|
||||
* (Working / StartReasoning / StartResponse / AppendText / AppendImage /
|
||||
* End) дёргается `sendChatAction(TYPING)`. Telegram сам гасит статус
|
||||
* «печатает…» через ~5 с, рефрешем держим его живым пока ход идёт
|
||||
* (включая длительные tool-call'ы, когда дельт текста нет).
|
||||
* - `agent.outbox.conversationEvents(convId)` — на `DurableEvent.AssistantMessage`
|
||||
* дёргается `sendMessage(...)` с **целым** текстом (стриминг через
|
||||
* `editMessage` намеренно не используется: Telegram это плохо
|
||||
* индексирует, и UX с мигающим draft'ом обычно хуже). На
|
||||
* `DurableEvent.Error` — отправляется короткое сообщение в чат.
|
||||
*
|
||||
* Scope — внешний [CoroutineScope]. В standalone это `agent.agentScope`
|
||||
* (см. [pw.binom.agentik.standalone.agent.ChatAgent]), чтобы при
|
||||
* `agent.close()` всё свернулось одной командой. Компонент стартует
|
||||
* polling-джобу на [install] и отменяет её на [uninstall].
|
||||
* «typing» также дёргается **немедленно** в `handleUpdate` — ДО `conv.send`
|
||||
* (по запросу: «печатает» должен появиться ровно в момент, когда
|
||||
* пользователь отправил текст, ещё до старта LLM).
|
||||
*
|
||||
* ## Per-conversation subscriber
|
||||
*
|
||||
* `OnlineOutbox` — общий `MutableSharedFlow(replay=0)`, события
|
||||
* фильтруются по `conversationId`. Несколько подписчиков на один
|
||||
* `conversationId` УЖЕ видят одни и те же события (broadcast). Это
|
||||
* нужно для случая «один Telegram-чат = один conv»: один подписчик
|
||||
* на conv создаётся при первом сообщении и живёт до `uninstall`,
|
||||
* повторные `conv.send` от того же чата идут в уже-подписанный flow.
|
||||
* Один subscriber на `convId` переиспользуется между сообщениями из того же
|
||||
* Telegram-чата (persistent-режим), так что повторные `conv.send` стримят
|
||||
* ответы через уже существующий collector, а не плодят параллельные.
|
||||
*
|
||||
* ## Failure modes
|
||||
* - Telegram rate-limit / сетевой сбой `sendChatAction`/`sendMessage` —
|
||||
* глотаем исключение: следующее online-событие освежит статус, а
|
||||
* следующий turn повторит `sendMessage` финального сообщения.
|
||||
* - Сетевой сбой `getUpdate` — 2 с backoff, повторный poll.
|
||||
* - `agent.close()` отменил `agentScope` → polling-job отменяется,
|
||||
* `uninstall` идемпотентен.
|
||||
*
|
||||
* - Telegram rate-limit на `editMessage` → глотаем исключение
|
||||
* (DraftMessage: «живой» текст важнее редактирования).
|
||||
* - Сетевой сбой `getUpdates` → 2 сек backoff и следующий poll
|
||||
* (long-poll Telegram сам ретраит).
|
||||
* - Агент отвалился (`agent.close()` отменил `agentScope`) →
|
||||
* polling-job тоже отменяется, `uninstall` идемпотентно.
|
||||
* ## Жизненный цикл
|
||||
* Клиент [tg] НЕ владеет компонентом — цикл отдаётся вызывающему
|
||||
* (см. [Main][pw.binom.agentik.standalone.Main] — закрывается в
|
||||
* shutdown-hook'е перед `agent.close()`). Scope — внешний
|
||||
* [CoroutineScope] (в standalone это `agent.agentScope`), polling стартует
|
||||
* на [install], отменяется на [uninstall].
|
||||
*/
|
||||
class TelegramBridgeComponent(
|
||||
private val tg: TelegramClient,
|
||||
@@ -68,25 +73,28 @@ class TelegramBridgeComponent(
|
||||
private val config: TelegramConfig = TelegramConfig(),
|
||||
) : Component {
|
||||
|
||||
/**
|
||||
* Key = `conversationId`. Value = subscriber job, чтобы при повторных
|
||||
* сообщениях из того же чата не плодить параллельные подписчики.
|
||||
*/
|
||||
private val log = KotlinLogging.logger {}
|
||||
|
||||
/** Key = `conversationId`. Value = subscriber job (один на conv). */
|
||||
private val subscribers: MutableMap<String, Job> = mutableMapOf()
|
||||
|
||||
private val subscribersLock = Mutex()
|
||||
|
||||
/**
|
||||
* Emit-only signal «новое сообщение пришло в чат» — для diag и
|
||||
* тестов. На этот поток можно подписаться ИЗ тестов (см.
|
||||
* `TelegramBridgeComponentTest`).
|
||||
* Throttle для `sendChatAction(TYPING)`: Telegram rate-limit ~1 запрос
|
||||
* в 5 секунд на чат. Без троттлинга получаем 429 на каждом
|
||||
* online-токене от LLM (в логе standalone было 30+ запросов за 10
|
||||
* секунд). Храним `chatId -> epochMs` последней успешной отправки
|
||||
* и шлём только если прошло ≥ [TYPING_REFRESH_MIN_INTERVAL_MS].
|
||||
*/
|
||||
private val lastTypingSentAt: MutableMap<String, Long> = mutableMapOf()
|
||||
|
||||
/** Тестовый сигнал: каждое входящее сообщение, прошедшее фильтр. */
|
||||
private val incomingEvents = MutableSharedFlow<IncomingEvent>(replay = 0, extraBufferCapacity = 64)
|
||||
|
||||
private var pollJob: Job? = null
|
||||
|
||||
override fun install(agent: MutableAgent) {
|
||||
require(pollJob == null) { "TelegramBridgeComponent already installed" }
|
||||
log.info("install: persistentConversations={}, pollingTimeoutSec={}", config.persistentConversations, config.pollingTimeoutSec)
|
||||
pollJob = scope.launch(CoroutineName("tg-poll")) {
|
||||
var offset: Long? = null
|
||||
while (isActive) {
|
||||
@@ -99,15 +107,21 @@ class TelegramBridgeComponent(
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (t: Throwable) {
|
||||
// Сетевой сбой / 5xx / Telegram rate-limit. Long-poll
|
||||
// сам не ретраит — спим и пробуем снова. offset
|
||||
// не двигаем, чтобы не потерять обновление.
|
||||
// Сетевой сбой / 5xx / rate-limit — long-poll сам не
|
||||
// ретраит, спим и пробуем снова. offset не двигаем.
|
||||
log.warn("getUpdate failed: {}", t.toString())
|
||||
delay(2_000)
|
||||
continue
|
||||
}
|
||||
for (u in updates) {
|
||||
offset = maxOf(offset ?: -1L, u.updateId + 1)
|
||||
handleUpdate(agent, u)
|
||||
try {
|
||||
handleUpdate(agent, u)
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (t: Throwable) {
|
||||
log.error("handleUpdate failed for updateId={}", u.updateId, t)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -116,175 +130,354 @@ class TelegramBridgeComponent(
|
||||
override fun uninstall(agent: MutableAgent) {
|
||||
pollJob?.cancel()
|
||||
pollJob = null
|
||||
// Подписчики отменять под `agent.agentScope` — обычно он уже
|
||||
// отменён при `agent.close()`. Но делаем defensive-cleanup:
|
||||
val toCancel = synchronized(subscribersLock) {
|
||||
val jobs = subscribers.values.toList()
|
||||
subscribers.clear()
|
||||
jobs
|
||||
}
|
||||
// defensive-cleanup подписчиков (обычно agentScope уже отменён
|
||||
// через agent.close()).
|
||||
val toCancel = synchronized(subscribers) { subscribers.values.toList().also { subscribers.clear() } }
|
||||
toCancel.forEach { it.cancel() }
|
||||
}
|
||||
|
||||
/** Тестовая подписка: видна каждое входящее сообщение, прошедшее фильтр. */
|
||||
/** Тестовая подписка на каждое входящее сообщение, прошедшее фильтр. */
|
||||
fun incomingFlow() = incomingEvents
|
||||
|
||||
private suspend fun handleUpdate(agent: MutableAgent, update: Update) {
|
||||
val msg = update.message ?: return
|
||||
val text = msg.text ?: return // v1: только текст. Медиа — следующая итерация.
|
||||
val text = msg.text ?: return
|
||||
val chatId = msg.chat?.id?.toString() ?: return
|
||||
val fromUser = msg.from?.userName ?: msg.from?.firstName ?: chatId
|
||||
|
||||
// Резолвим conversation. В persistent-режиме маппинг переиспользует
|
||||
// существующий conv_id; в temp-режиме каждый входящий текст =
|
||||
// свежая беседа, маппинг обновляется на последнюю.
|
||||
// Резолвим conversation: persistent — переиспользуем conv_id,
|
||||
// temp — каждое сообщение = свежая беседа, маппинг обновляется.
|
||||
val convId = if (config.persistentConversations) {
|
||||
map.getConversationId(chatId) ?: run {
|
||||
val conv = agent.createConversation(temp = false)
|
||||
log.info("chatId={} -> new persistent conv={} (user={})", chatId, conv.id, fromUser)
|
||||
map.put(chatId, conv.id)
|
||||
conv.id
|
||||
}
|
||||
} else {
|
||||
val conv = agent.createConversation(temp = true)
|
||||
log.info("chatId={} -> new temp conv={} (user={})", chatId, conv.id, fromUser)
|
||||
map.put(chatId, conv.id)
|
||||
conv.id
|
||||
}
|
||||
|
||||
val conv = agent.getConversation(convId) ?: return
|
||||
val conv = agent.getConversation(convId) ?: run {
|
||||
log.error("getConversation({}) returned null", convId)
|
||||
return
|
||||
}
|
||||
|
||||
log.info("inbound chatId={} conv={} user={} text={}", chatId, convId, fromUser, text)
|
||||
|
||||
incomingEvents.tryEmit(IncomingEvent(chatId = chatId, convId = convId, fromUser = fromUser, text = text))
|
||||
|
||||
// «typing…» в чате — пользователь видит индикатор пока агент думает.
|
||||
runCatching { tg.sendChatAction(chatId, SendChatEvent.Action.TYPING) }
|
||||
// «печатает…» в чат немедленно, ДО отправки в агент. С троттлингом —
|
||||
// Telegram rate-limit ~1 запрос/5с на чат; без троттла получали
|
||||
// 429 на каждом online-токене.
|
||||
sendTypingThrottled(chatId)
|
||||
|
||||
// Подписчик на стрим ответа — идемпотентно (один на conv).
|
||||
// Подписчик на online+durable — идемпотентно, один на conv.
|
||||
ensureSubscriber(agent, convId = convId, chatId = chatId)
|
||||
|
||||
// Сама отправка — fire-and-forget: agent обрабатывает ход асинхронно,
|
||||
// ошибки стрима ловятся в subscriber'е.
|
||||
// Сама отправка — fire-and-forget: ошибки стрима ловятся в
|
||||
// subscriber'е через DurableEvent.Error; исключения из send
|
||||
// отдаём одной строкой в чат.
|
||||
scope.launch(CoroutineName("tg-send-$convId")) {
|
||||
try {
|
||||
conv.send(content = listOf(Content.Text(text)))
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (t: Throwable) {
|
||||
runCatching {
|
||||
tg.sendMessage(
|
||||
TextMessage(
|
||||
chatId = chatId,
|
||||
text = "⚠️ agent error: ${t.message ?: t::class.simpleName}",
|
||||
)
|
||||
)
|
||||
}
|
||||
log.error("conv.send failed for conv={}", convId, t)
|
||||
sendPlain(chatId, "⚠️ agent error: ${t.message ?: t::class.simpleName}")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun ensureSubscriber(agent: MutableAgent, convId: String, chatId: String) {
|
||||
// Fast-path: уже подписаны.
|
||||
val already: Job? = synchronized(subscribersLock) { subscribers[convId] }
|
||||
val already: Job? = synchronized(subscribers) { subscribers[convId] }
|
||||
if (already?.isActive == true) return
|
||||
|
||||
val job = scope.launch(CoroutineName("tg-sub-$convId")) {
|
||||
// Состояние текущего хода:
|
||||
// - draftMessageId — id «редактируемого» сообщения в TG.
|
||||
// - buffer — накопленный текст от AppendText.
|
||||
var draftMessageId: Long? = null
|
||||
var buffer = StringBuilder()
|
||||
|
||||
agent.onlineOutbox.onlineEvents(conversationId = convId).collect { ev ->
|
||||
when (ev) {
|
||||
is OnlineEvent.StartResponse -> {
|
||||
// Если от прошлого хода остался draft — закрываем
|
||||
// его отдельным сообщением (на случай «ответ
|
||||
// перебил недописанный предыдущий»).
|
||||
if (draftMessageId != null && buffer.isNotEmpty()) {
|
||||
runCatching {
|
||||
tg.editMessage(
|
||||
EditTextRequest(
|
||||
chatId = chatId,
|
||||
messageId = draftMessageId,
|
||||
text = buffer.toString(),
|
||||
)
|
||||
)
|
||||
}
|
||||
}
|
||||
draftMessageId = null
|
||||
buffer = StringBuilder()
|
||||
|
||||
try {
|
||||
val sent = tg.sendMessage(TextMessage(chatId = chatId, text = "..."))
|
||||
draftMessageId = sent.messageId
|
||||
} catch (t: Throwable) {
|
||||
// Сеть/TG 5xx — пропускаем этот ход (следующий
|
||||
// StartResponse попробует снова).
|
||||
}
|
||||
// Два под-fluves в одном subscriber-job'е: typing-refresh
|
||||
// (online) и финальный sendMessage (durable AssistantMessage /
|
||||
// Error). Падают/отменяются вместе с job'ом.
|
||||
coroutineScope {
|
||||
launch(CoroutineName("tg-typing-$convId")) {
|
||||
agent.onlineOutbox.onlineEvents(convId).collect {
|
||||
sendTypingThrottled(chatId)
|
||||
}
|
||||
is OnlineEvent.AppendText -> {
|
||||
if (draftMessageId == null) return@collect
|
||||
buffer.append(ev.body)
|
||||
runCatching {
|
||||
tg.editMessage(
|
||||
EditTextRequest(
|
||||
chatId = chatId,
|
||||
messageId = draftMessageId,
|
||||
text = buffer.toString(),
|
||||
)
|
||||
)
|
||||
}
|
||||
}
|
||||
is OnlineEvent.End -> {
|
||||
// Если buffer пустой (агент прервал ход до первого
|
||||
// токена / не было StartResponse) — убираем draft
|
||||
// «…» отправкой финального сообщения.
|
||||
if (buffer.isEmpty()) {
|
||||
if (draftMessageId != null) {
|
||||
runCatching {
|
||||
tg.editMessage(
|
||||
EditTextRequest(
|
||||
chatId = chatId,
|
||||
messageId = draftMessageId,
|
||||
text = "(empty response)",
|
||||
)
|
||||
}
|
||||
launch(CoroutineName("tg-final-$convId")) {
|
||||
agent.outbox.conversationEvents(convId).collect { ev ->
|
||||
when (val d = ev.event) {
|
||||
is DurableEvent.AssistantMessage -> {
|
||||
val text = extractText(d.content)
|
||||
if (text.isBlank()) {
|
||||
log.info("AssistantMessage empty -> '(empty response)' chatId={}", chatId)
|
||||
sendPlain(chatId, "(empty response)")
|
||||
} else {
|
||||
log.info(
|
||||
"AssistantMessage -> chatId={} bytes={} preview={}",
|
||||
chatId,
|
||||
text.length,
|
||||
preview(text),
|
||||
)
|
||||
sendMarkdown(chatId, text)
|
||||
}
|
||||
}
|
||||
} else if (draftMessageId != null) {
|
||||
runCatching {
|
||||
tg.editMessage(
|
||||
EditTextRequest(
|
||||
chatId = chatId,
|
||||
messageId = draftMessageId,
|
||||
text = buffer.toString(),
|
||||
)
|
||||
)
|
||||
is DurableEvent.Error -> {
|
||||
log.warn("DurableEvent.Error chatId={} message={}", chatId, d.message)
|
||||
sendPlain(chatId, "⚠️ ${d.message}")
|
||||
}
|
||||
else -> Unit
|
||||
}
|
||||
draftMessageId = null
|
||||
buffer = StringBuilder()
|
||||
}
|
||||
else -> Unit // Working / StartReasoning / AppendImage — не редактируем TG.
|
||||
}
|
||||
}
|
||||
}
|
||||
synchronized(subscribersLock) {
|
||||
// Кто-то мог опередить нас между fast-path и slow-path.
|
||||
synchronized(subscribers) {
|
||||
val existing = subscribers[convId]
|
||||
if (existing?.isActive == true) {
|
||||
job.cancel()
|
||||
} else {
|
||||
subscribers[convId] = job
|
||||
job.invokeOnCompletion {
|
||||
synchronized(subscribersLock) { subscribers.remove(convId, job) }
|
||||
synchronized(subscribers) { subscribers.remove(convId, job) }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Throttled-обёртка над `sendChatAction(TYPING)`. Telegram гасит
|
||||
* статус «typing» через ~5 секунд и режет rate-limit до ~1
|
||||
* запроса/5с — иначе получаем 429 на каждом online-токене от LLM
|
||||
* (наблюдалось 30+ запросов за 10 секунд в логе standalone).
|
||||
*
|
||||
* Хранит `chatId -> epochMs` последней УСПЕШНОЙ отправки; пока
|
||||
* интервал меньше [TYPING_REFRESH_MIN_INTERVAL_MS], запрос
|
||||
* пропускается. 429/403/сетевые сбои тихо логируются — следующий
|
||||
* токен через 4с снова попробует.
|
||||
*/
|
||||
private suspend fun sendTypingThrottled(chatId: String) {
|
||||
val now = System.currentTimeMillis()
|
||||
val last = synchronized(lastTypingSentAt) { lastTypingSentAt[chatId] }
|
||||
if (last != null && now - last < TYPING_REFRESH_MIN_INTERVAL_MS) return
|
||||
try {
|
||||
tg.sendChatAction(chatId, SendChatEvent.Action.TYPING)
|
||||
synchronized(lastTypingSentAt) { lastTypingSentAt[chatId] = now }
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (t: Throwable) {
|
||||
log.warn("sendChatAction(TYPING) failed for chatId={}: {}", chatId, t.toString())
|
||||
}
|
||||
}
|
||||
|
||||
private fun extractText(content: List<Content>): String = buildString {
|
||||
for (c in content) {
|
||||
if (c is Content.Text) append(c.body)
|
||||
// Image и прочие типы в текстовом Telegram-канале
|
||||
// игнорируются (нет media upload в этой итерации).
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Шлёт короткий plain-text в чат (без parseMode). Используется для
|
||||
* внутренних сервисных сообщений моста: «(empty response)»,
|
||||
* «⚠️ agent error: ...», «⚠️ <DurableEvent.Error>». Никаких
|
||||
* markdown-преобразований здесь быть не должно — текст уже
|
||||
* финальный.
|
||||
*/
|
||||
private suspend fun sendPlain(chatId: String, text: String) {
|
||||
try {
|
||||
tg.sendMessage(TextMessage(chatId = chatId, text = text))
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (t: Throwable) {
|
||||
// Типичные причины: 400 (битый текст, chat_id формат),
|
||||
// 403 (бот кикнут / чат приватный и нас не добавили),
|
||||
// 429 (Too Many Requests). Логируем ВСЁ — иначе «агент
|
||||
// молча проглотил ответ» не разобрать.
|
||||
log.error("sendMessage(plain) failed chatId={} textLen={}: {}", chatId, text.length, t.toString())
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Шлёт [markdown] в чат как Telegram-HTML. Конвертация в HTML
|
||||
* ([MarkdownToTelegram.convert]) даёт Telegram'у сущности (`<b>`,
|
||||
* `<i>`, `<code>`, `<pre>`, `<a>`); если LLM прислал plain-текст,
|
||||
* parseMode всё равно ставим HTML — Telegram-парсер HTML мягко
|
||||
* относится к голому тексту (просто не находит тегов), а без
|
||||
* parseMode LLM-маркдаун типа `**bold**` пришёл бы как есть.
|
||||
*
|
||||
* Если после конвертации текст длиннее 4096 символов (лимит Bot API
|
||||
* «после entities parsing»), режем по переводам строки на чанки и
|
||||
* шлём последовательно. Каждый чанк — независимый sendMessage,
|
||||
* чтобы Telegram не ругался на обрыв посреди тега.
|
||||
*/
|
||||
private suspend fun sendMarkdown(chatId: String, markdown: String) {
|
||||
val html = MarkdownToTelegram.convert(markdown)
|
||||
if (html.length <= MAX_TG_MESSAGE) {
|
||||
try {
|
||||
tg.sendMessage(
|
||||
TextMessage(
|
||||
chatId = chatId,
|
||||
text = html,
|
||||
parseMode = ParseMode.HTML,
|
||||
)
|
||||
)
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (t: Throwable) {
|
||||
// Самое частое при «агент ответил, а телега
|
||||
// заигнорила»: 400 Bad Request: can't parse entities —
|
||||
// конвертер оставил неэкранированный `<` или `&`,
|
||||
// Telegram ругается и не показывает сообщение. Логируем
|
||||
// ПОЛНОСТЬЮ (включая тело), чтобы видеть, что именно
|
||||
// Telegram отверг.
|
||||
log.error(
|
||||
"sendMessage(HTML) failed chatId={} htmlLen={}: {} --- offending html:\n{}",
|
||||
chatId,
|
||||
html.length,
|
||||
t.toString(),
|
||||
html,
|
||||
)
|
||||
}
|
||||
return
|
||||
}
|
||||
// Длинное сообщение: режем по границам чанков.
|
||||
val chunks = splitForTelegram(html, MAX_TG_MESSAGE)
|
||||
log.info("sendMarkdown: chatId={} htmlLen={} -> {} chunks", chatId, html.length, chunks.size)
|
||||
for ((idx, piece) in chunks.withIndex()) {
|
||||
try {
|
||||
tg.sendMessage(
|
||||
TextMessage(
|
||||
chatId = chatId,
|
||||
text = piece,
|
||||
parseMode = ParseMode.HTML,
|
||||
)
|
||||
)
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (t: Throwable) {
|
||||
log.error(
|
||||
"sendMessage(HTML chunk {}/{}) failed chatId={} len={}: {} --- chunk:\n{}",
|
||||
idx + 1,
|
||||
chunks.size,
|
||||
chatId,
|
||||
piece.length,
|
||||
t.toString(),
|
||||
piece,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Первая строка текста (для логов — чтобы длинный markdown не забивал stderr). */
|
||||
private fun preview(text: String): String {
|
||||
val firstLine = text.lineSequence().firstOrNull().orEmpty()
|
||||
return if (firstLine.length <= 120) firstLine else firstLine.take(117) + "…"
|
||||
}
|
||||
|
||||
/**
|
||||
* Режет [html] на куски ≤ [maxLen] по границам перевода строки,
|
||||
* не разрывая HTML-тегов. Если внутри одного «абзаца» (между \n)
|
||||
* текст длиннее [maxLen], рубим по границе слова.
|
||||
*/
|
||||
private fun splitForTelegram(html: String, maxLen: Int): List<String> {
|
||||
val out = mutableListOf<String>()
|
||||
var i = 0
|
||||
while (i < html.length) {
|
||||
val remaining = html.length - i
|
||||
if (remaining <= maxLen) {
|
||||
out.add(html.substring(i))
|
||||
break
|
||||
}
|
||||
// Ищем последний \n в окне [i, i+maxLen], который НЕ внутри тега.
|
||||
var cutAt = -1
|
||||
var j = i + maxLen
|
||||
while (j > i) {
|
||||
if (html[j] == '\n' && !insideOurTag(html, j)) {
|
||||
cutAt = j
|
||||
break
|
||||
}
|
||||
j--
|
||||
}
|
||||
if (cutAt < 0) {
|
||||
// Нет перевода строки — режем по пробелу, опять же вне тега.
|
||||
j = i + maxLen
|
||||
while (j > i) {
|
||||
if (html[j] == ' ' && !insideOurTag(html, j)) {
|
||||
cutAt = j
|
||||
break
|
||||
}
|
||||
j--
|
||||
}
|
||||
}
|
||||
if (cutAt < 0) {
|
||||
// Совсем плохо — рубим ровно по maxLen, пусть Telegram ругается.
|
||||
cutAt = i + maxLen
|
||||
}
|
||||
out.add(html.substring(i, cutAt).trimEnd())
|
||||
i = cutAt + 1
|
||||
}
|
||||
return out.filter { it.isNotEmpty() }
|
||||
}
|
||||
|
||||
private fun insideOurTag(s: String, pos: Int): Boolean {
|
||||
var k = 0
|
||||
var depth = 0
|
||||
while (k < pos) {
|
||||
val openIdx = indexOfAnyOpen(s, k)
|
||||
val closeIdx = indexOfAnyClose(s, k)
|
||||
if (closeIdx in k..pos) {
|
||||
depth--
|
||||
k = closeIdx + 1
|
||||
continue
|
||||
}
|
||||
if (openIdx in k..pos) {
|
||||
depth++
|
||||
k = openIdx + 1
|
||||
continue
|
||||
}
|
||||
break
|
||||
}
|
||||
return depth > 0
|
||||
}
|
||||
|
||||
private fun indexOfAnyOpen(s: String, from: Int): Int {
|
||||
val a = s.indexOf("<b>", from)
|
||||
val b = s.indexOf("<i>", from)
|
||||
val c = s.indexOf("<code>", from)
|
||||
val d = s.indexOf("<pre>", from)
|
||||
val e = s.indexOf("<a ", from)
|
||||
return listOf(a, b, c, d, e).filter { it >= 0 }.minOrNull() ?: -1
|
||||
}
|
||||
|
||||
private fun indexOfAnyClose(s: String, from: Int): Int {
|
||||
val a = s.indexOf("</b>", from)
|
||||
val b = s.indexOf("</i>", from)
|
||||
val c = s.indexOf("</code>", from)
|
||||
val d = s.indexOf("</pre>", from)
|
||||
val e = s.indexOf("</a>", from)
|
||||
return listOf(a, b, c, d, e).filter { it >= 0 }.minOrNull() ?: -1
|
||||
}
|
||||
|
||||
data class IncomingEvent(
|
||||
val chatId: String,
|
||||
val convId: String,
|
||||
val fromUser: String,
|
||||
val text: String,
|
||||
)
|
||||
|
||||
private companion object {
|
||||
/** Лимит Bot API на размер одного сообщения (после entities parsing). */
|
||||
const val MAX_TG_MESSAGE = 4096
|
||||
/**
|
||||
* Минимальный интервал между двумя `sendChatAction(TYPING)` для
|
||||
* одного chatId. Telegram сам гасит статус «typing» через ~5
|
||||
* секунд и шлёт 429 при превышении rate-limit.
|
||||
*/
|
||||
const val TYPING_REFRESH_MIN_INTERVAL_MS: Long = 4_000
|
||||
}
|
||||
}
|
||||
+284
@@ -0,0 +1,284 @@
|
||||
package pw.binom.agentik.integrations.telegram
|
||||
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
class MarkdownToTelegramTest {
|
||||
|
||||
@Test
|
||||
fun `balanced inner tags pass through with matching closing`() {
|
||||
// LLM вставил литеральный HTML `<code>...</code><i>...</i>` — это
|
||||
// валидный Telegram-HTML, его надо пропустить как есть, не
|
||||
// путая внешний `</code>` с внутренним `</i>`. Старая реализация
|
||||
// брала первый закрывающий тег и считала его парой к `<code>`,
|
||||
// после чего `</code>` экранировался → Telegram возвращал 400
|
||||
// `Can't find end tag corresponding to start tag "code"`.
|
||||
assertEquals(
|
||||
"<code>Судья</code>",
|
||||
MarkdownToTelegram.convert("<code>Судья</code>"),
|
||||
)
|
||||
assertEquals(
|
||||
"<code>inline code в середине <i>курсивного</i> предложения работает</code>",
|
||||
MarkdownToTelegram.convert(
|
||||
"<code>inline code в середине <i>курсивного</i> предложения работает</code>",
|
||||
),
|
||||
)
|
||||
// Вложенный <b> внутри <i> тоже не должен ломать баланс.
|
||||
assertEquals(
|
||||
"<i>a <code>x</code> b <b>y</b> c</i>",
|
||||
MarkdownToTelegram.convert("<i>a <code>x</code> b <b>y</b> c</i>"),
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `REPRO nested html from LLM`() {
|
||||
// Реальный случай из лога standalone (TLM-инспектор писал
|
||||
// литеральный HTML в ответе):
|
||||
val input = """
|
||||
| Вещдоказательство | Состояние | Примечание |
|
||||
|---|---|---|
|
||||
| <b>Стол</b> | Протрезвевший | <i>Наконец-то</i> |
|
||||
| ~~Купе~~ | ~~Отменено~~ | Не доехали |
|
||||
| <code>Судья</code> | Наблюдает | Сидит, молчит |
|
||||
|
||||
А теперь сюрприз:
|
||||
<code>inline code в середине <i>курсивного</i> предложения работает</code> — проверено.
|
||||
""".trimIndent()
|
||||
val out = MarkdownToTelegram.convert(input)
|
||||
println("REPRO OUTPUT:\n$out")
|
||||
// Главное: НЕТ незакрытых <code> (раньше </code> экранировался,
|
||||
// и Telegram возвращал 400 "Can't find end tag corresponding to
|
||||
// start tag code").
|
||||
val openCode = "<code>".toRegex().findAll(out).count()
|
||||
val closeCode = "</code>".toRegex().findAll(out).count()
|
||||
assertEquals(openCode, closeCode, "unbalanced <code> tags in output")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `bold via double asterisk`() {
|
||||
assertEquals("<b>bold</b>", MarkdownToTelegram.convert("**bold**"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `bold via double underscore`() {
|
||||
assertEquals("<b>bold</b>", MarkdownToTelegram.convert("__bold__"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `italic via single asterisk`() {
|
||||
assertEquals("<i>italic</i>", MarkdownToTelegram.convert("*italic*"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `italic via single underscore`() {
|
||||
assertEquals("<i>italic</i>", MarkdownToTelegram.convert("_italic_"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `inline code with backticks`() {
|
||||
assertEquals("<code>x</code>", MarkdownToTelegram.convert("`x`"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `fenced code block becomes pre`() {
|
||||
val md = """
|
||||
```
|
||||
let x = 1
|
||||
```
|
||||
""".trimIndent()
|
||||
// Внутри pre содержимое экранируется.
|
||||
assertEquals("<pre>let x = 1</pre>", MarkdownToTelegram.convert(md))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `link with markdown syntax`() {
|
||||
val html = MarkdownToTelegram.convert("[click](https://example.com)")
|
||||
assertEquals("<a href=\"https://example.com\">click</a>", html)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `link href attribute is escaped`() {
|
||||
val html = MarkdownToTelegram.convert("[x](https://e.com/?a=1&b=2)")
|
||||
assertEquals("<a href=\"https://e.com/?a=1&b=2\">x</a>", html)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `header becomes bold (no h tag in tg html)`() {
|
||||
assertEquals("<b>Title</b>", MarkdownToTelegram.convert("# Title"))
|
||||
assertEquals("<b>Sub</b>", MarkdownToTelegram.convert("### Sub"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `bullet list becomes bullet character`() {
|
||||
assertEquals("• first\n• second", MarkdownToTelegram.convert("- first\n- second"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `ordered list keeps numbers`() {
|
||||
assertEquals("1. first\n2. second", MarkdownToTelegram.convert("1. first\n2. second"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `block quote gets vertical bar prefix`() {
|
||||
assertEquals("│ hello", MarkdownToTelegram.convert("> hello"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `mixed inline bold inside sentence`() {
|
||||
assertEquals("Hello <b>world</b>!", MarkdownToTelegram.convert("Hello **world**!"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `plain text without markdown passes through unchanged`() {
|
||||
assertEquals("just text", MarkdownToTelegram.convert("just text"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `special characters are escaped outside tags`() {
|
||||
assertEquals("a < b & c", MarkdownToTelegram.convert("a < b & c"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `asterisk without pair is left as-is`() {
|
||||
// Одиночный `*` (не парный) — Telegram не съест, но и мы не должны
|
||||
// ломать: оставляем как текст. Экранирования не требуется.
|
||||
assertEquals("a * b", MarkdownToTelegram.convert("a * b"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `bold inside sentence keeps surrounding text plain`() {
|
||||
val md = "see **foo** and **bar** end"
|
||||
assertEquals("see <b>foo</b> and <b>bar</b> end", MarkdownToTelegram.convert(md))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `escaping inside code block`() {
|
||||
val md = """
|
||||
```
|
||||
if (a < b) { a & b }
|
||||
```
|
||||
""".trimIndent()
|
||||
assertEquals("<pre>if (a < b) { a & b }</pre>", MarkdownToTelegram.convert(md))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `link text containing ampersand is escaped outside tag content`() {
|
||||
val html = MarkdownToTelegram.convert("[A & B](https://e.com)")
|
||||
assertEquals("<a href=\"https://e.com\">A & B</a>", html)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `empty markdown returns empty string`() {
|
||||
assertEquals("", MarkdownToTelegram.convert(""))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `multi-line markdown preserves line breaks`() {
|
||||
val md = "line1\nline2"
|
||||
assertEquals("line1\nline2", MarkdownToTelegram.convert(md))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `simple markdown table renders with box-drawing inside pre`() {
|
||||
val md = """
|
||||
| Name | Score | Note |
|
||||
| --- | --- | --- |
|
||||
| Alice | 10 | ok |
|
||||
| Bob | 5 | late
|
||||
""".trimIndent()
|
||||
val out = MarkdownToTelegram.convert(md)
|
||||
// Таблица обязана быть обёрнута в <pre>...</pre> для моноширинного
|
||||
// рендера в Telegram.
|
||||
assertTrue(out.startsWith("<pre>"), "table must start with <pre>, got: $out")
|
||||
assertTrue(out.endsWith("</pre>"), "table must end with </pre>, got: $out")
|
||||
// Рамка.
|
||||
assertTrue("┌" in out, "missing top-left corner, got: $out")
|
||||
assertTrue("┐" in out, "missing top-right corner, got: $out")
|
||||
assertTrue("└" in out, "missing bottom-left corner, got: $out")
|
||||
assertTrue("┘" in out, "missing bottom-right corner, got: $out")
|
||||
assertTrue("├" in out, "missing mid-left T-junction, got: $out")
|
||||
assertTrue("┤" in out, "missing mid-right T-junction, got: $out")
|
||||
// Заголовок и данные видны как plain-текст.
|
||||
assertTrue("Alice" in out, "missing Alice row, got: $out")
|
||||
assertTrue("Bob" in out, "missing Bob row, got: $out")
|
||||
assertTrue("Name" in out, "missing header Name, got: $out")
|
||||
assertTrue("Score" in out, "missing header Score, got: $out")
|
||||
// Сырой md-разделитель НЕ должен выживать.
|
||||
assertTrue("---" !in out, "raw `---` separator must not leak, got: $out")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `markdown table column widths are padded to max cell width`() {
|
||||
val md = """
|
||||
| A | BB |
|
||||
| --- | --- |
|
||||
| xxx | y |
|
||||
""".trimIndent()
|
||||
val out = MarkdownToTelegram.convert(md)
|
||||
// Ширины столбцов: col1=max(1,3)=3, col2=max(2,1)=2.
|
||||
// Header: «A» pad→3, «BB» pad→2. Row: «xxx» pad→3, «y» pad→2.
|
||||
assertTrue("│ xxx │ y │" in out, "row not padded correctly, got: $out")
|
||||
assertTrue("│ A │ BB │" in out, "header not padded correctly, got: $out")
|
||||
// Верхняя/нижняя рамка: col1=3→5 дефисов, col2=2→4 дефиса.
|
||||
assertTrue("┌─────┬────┐" in out, "top border wrong, got: $out")
|
||||
assertTrue("└─────┴────┘" in out, "bottom border wrong, got: $out")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `markdown table cells lose inline markdown and HTML tags`() {
|
||||
// Внутри <pre> Telegram НЕ интерпретирует <b>/<i> — поэтому
|
||||
// конвертер должен выкидывать разметку и оставлять только текст.
|
||||
val md = """
|
||||
| Item | Status |
|
||||
| --- | --- |
|
||||
| **Стол** | <i>готов</i> |
|
||||
| `код` | прошло |
|
||||
""".trimIndent()
|
||||
val out = MarkdownToTelegram.convert(md)
|
||||
assertTrue("Стол" in out, "bold text body must survive, got: $out")
|
||||
assertTrue("готов" in out, "italic text body must survive, got: $out")
|
||||
assertTrue("прошло" in out, "plain cell must survive, got: $out")
|
||||
assertTrue("код" in out, "code text body must survive, got: $out")
|
||||
// Сами маркеры выкинуты.
|
||||
assertTrue("<b>" !in out && "<i>" !in out, "html tags must be stripped, got: $out")
|
||||
assertTrue("**" !in out && "`" !in out, "md markers must be stripped, got: $out")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `line that looks like table header but has no separator falls through as plain text`() {
|
||||
// Строка с `|`, но без `| --- |` после неё — обычный текст,
|
||||
// таблица не строится.
|
||||
val md = "hello | world"
|
||||
val out = MarkdownToTelegram.convert(md)
|
||||
assertTrue("<pre>" !in out, "must not wrap non-table in <pre>, got: $out")
|
||||
assertTrue("│" !in out, "must not emit box-drawing for non-table, got: $out")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `REPRO table from LLM now renders as boxed grid`() {
|
||||
// Реальный случай из standalone-лога: агент выдал markdown-таблицу
|
||||
// с HTML внутри ячеек. До фикса она уходила в Telegram как сырой
|
||||
// `| Вещдоказательство | Состояние | ...`; теперь — boxed grid.
|
||||
val input = """
|
||||
Вот расклад:
|
||||
|
||||
| Вещдоказательство | Состояние | Примечание |
|
||||
|---|---|---|
|
||||
| Стол | Протрезвевший | Наконец-то |
|
||||
| Купе | Отменено | Не доехали |
|
||||
|
||||
Готово.
|
||||
""".trimIndent()
|
||||
val out = MarkdownToTelegram.convert(input)
|
||||
// Рамка рисуется.
|
||||
assertTrue("┌" in out, "missing top border, got: $out")
|
||||
assertTrue("Стол" in out, "row must survive, got: $out")
|
||||
assertTrue("Протрезвевший" in out, "cell text must survive, got: $out")
|
||||
// Текст «Вот расклад:» и «Готово.» — обычными строками.
|
||||
assertTrue("Вот расклад:" in out, "intro line missing, got: $out")
|
||||
assertTrue("Готово." in out, "outro line missing, got: $out")
|
||||
// Никакого raw-разделителя `|---|---|` в выводе.
|
||||
assertTrue("|---|" !in out, "raw separator leaked, got: $out")
|
||||
}
|
||||
}
|
||||
+322
-138
@@ -6,26 +6,25 @@ import kotlinx.coroutines.cancel
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||
import kotlinx.coroutines.flow.collect
|
||||
import kotlinx.coroutines.flow.asSharedFlow
|
||||
import kotlinx.coroutines.flow.filter
|
||||
import kotlinx.coroutines.flow.filterIsInstance
|
||||
import kotlinx.coroutines.flow.first
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import kotlinx.coroutines.withTimeout
|
||||
import kotlinx.coroutines.withTimeout
|
||||
import pw.binom.agentik.agent.Component
|
||||
import pw.binom.agentik.agent.MutableAgent
|
||||
import pw.binom.agentik.content.Content
|
||||
import pw.binom.agentik.outbox.AgentEvent
|
||||
import pw.binom.agentik.outbox.CommonEvent
|
||||
import pw.binom.agentik.outbox.OnlineEvent
|
||||
import pw.binom.agentik.outbox.OutboxStore
|
||||
import pw.binom.agentik.outbox.OnlineOutbox
|
||||
import pw.binom.agentik.content.MessageContext
|
||||
import pw.binom.agentik.journal.ConversationStore
|
||||
import pw.binom.agentik.journal.JournalStore
|
||||
import pw.binom.agentik.outbox.AgentEvent
|
||||
import pw.binom.agentik.outbox.CommonEvent
|
||||
import pw.binom.agentik.outbox.DurableEvent
|
||||
import pw.binom.agentik.outbox.OnlineEvent
|
||||
import pw.binom.agentik.outbox.OnlineOutbox
|
||||
import pw.binom.agentik.outbox.OutboxStore
|
||||
import pw.binom.agentik.proto.AgentInfo
|
||||
import pw.binom.agentik.proto.ChatSnapshot
|
||||
import pw.binom.agentik.proto.ConversationsSnapshot
|
||||
@@ -33,21 +32,14 @@ import pw.binom.agentik.proto.Conversation
|
||||
import pw.binom.db.ksqlite.SQLiteConnection
|
||||
import pw.binom.telegram.TelegramClient
|
||||
import pw.binom.telegram.dto.BotCommand
|
||||
import pw.binom.telegram.dto.CallbackQuery
|
||||
import pw.binom.telegram.dto.Chat
|
||||
import pw.binom.telegram.dto.ChatType
|
||||
import pw.binom.telegram.dto.ChosenInlineResult
|
||||
import pw.binom.telegram.dto.EditMessageResult
|
||||
import pw.binom.telegram.dto.EditTextRequest
|
||||
import pw.binom.telegram.dto.File
|
||||
import pw.binom.telegram.dto.InlineQuery
|
||||
import pw.binom.telegram.dto.Message
|
||||
import pw.binom.telegram.dto.ParseMode
|
||||
import pw.binom.telegram.dto.Poll
|
||||
import pw.binom.telegram.dto.PollAnswer
|
||||
import pw.binom.telegram.dto.PreCheckoutQuery
|
||||
import pw.binom.telegram.dto.SendChatEvent
|
||||
import pw.binom.telegram.dto.ShippingQuery
|
||||
import pw.binom.telegram.dto.TextMessage
|
||||
import pw.binom.telegram.dto.Update
|
||||
import pw.binom.telegram.dto.User
|
||||
@@ -64,39 +56,34 @@ import kotlin.test.fail
|
||||
import kotlin.time.Instant
|
||||
|
||||
/**
|
||||
* Тесты моста [TelegramBridgeComponent].
|
||||
* Тесты [TelegramBridgeComponent].
|
||||
*
|
||||
* Контракт:
|
||||
* - входящий `Update.message` с текстом → создан/найден `Conversation` (по
|
||||
* [TelegramChatMap]), `conv.send(Content.Text(...))` дёрнут;
|
||||
* - «typing…» в чат через `sendChatAction` шлётся на каждое входящее;
|
||||
* - стрим ответа (StartResponse → AppendText → End) рендерится в TG через
|
||||
* `sendMessage("...")` + `editMessage(...)` с накопленным текстом;
|
||||
* - ошибка в `conv.send` падает в чат как «⚠️ …» (одна строка);
|
||||
* - `uninstall(agent)` отменяет polling-джобу.
|
||||
* [TelegramChatMap]), `conv.send(Content.Text(...))` дёрнут, `sendChatAction(TYPING)`
|
||||
* в чат уходит ДО `conv.send`;
|
||||
* - на каждом online-событии для conv'а (Working / StartResponse /
|
||||
* AppendText / End) мост освежает `sendChatAction(TYPING)` в чат;
|
||||
* - на `DurableEvent.AssistantMessage` (целое сообщение) мост шлёт
|
||||
* `sendMessage(...)` с полным текстом (стриминг через `editMessage`
|
||||
* НЕ используется);
|
||||
* - на `DurableEvent.Error` шлётся короткое сообщение об ошибке;
|
||||
* - `uninstall(agent)` отменяет polling-джобу и подписчиков.
|
||||
*/
|
||||
/**
|
||||
* Подписаться на [flow] в фоне, накапливать события в список и вернуть его.
|
||||
* Подписка стартует до теста (subscribe-before-act), потом тест шлёт события и ассертит.
|
||||
*/
|
||||
private fun <T> subscribe(scope: CoroutineScope, flow: kotlinx.coroutines.flow.Flow<T>): MutableList<T> {
|
||||
private fun <T> subscribe(scope: CoroutineScope, flow: Flow<T>): MutableList<T> {
|
||||
val list = mutableListOf<T>()
|
||||
scope.launch {
|
||||
flow.collect { list.add(it) }
|
||||
}
|
||||
scope.launch { flow.collect { list.add(it) } }
|
||||
return list
|
||||
}
|
||||
|
||||
/** Ждём, пока в [list] появится элемент, удовлетворяющий [match], или таймаут. */
|
||||
private suspend fun <T> List<T>.awaitOne(match: (T) -> Boolean, timeoutMs: Long = 2_000): T {
|
||||
return kotlinx.coroutines.withTimeout(timeoutMs) {
|
||||
private suspend fun <T> List<T>.awaitOne(match: (T) -> Boolean, timeoutMs: Long = 2_000): T =
|
||||
withTimeout(timeoutMs) {
|
||||
while (true) {
|
||||
firstOrNull(match)?.let { return@withTimeout it }
|
||||
kotlinx.coroutines.delay(10)
|
||||
}
|
||||
@Suppress("UNREACHABLE_CODE") error("unreachable")
|
||||
}
|
||||
}
|
||||
|
||||
class TelegramBridgeComponentTest {
|
||||
|
||||
@@ -112,14 +99,12 @@ class TelegramBridgeComponentTest {
|
||||
fakeTg = FakeTelegramClient()
|
||||
connection = SQLiteConnection.memory("tg-test-${Random.nextLong()}")
|
||||
chatMap = TelegramChatMap(connection)
|
||||
fakeAgent = FakeMutableAgent(onlineOutbox = FakeOnlineOutbox())
|
||||
fakeAgent = FakeMutableAgent()
|
||||
scope = CoroutineScope(SupervisorJob() + kotlinx.coroutines.Dispatchers.Default)
|
||||
}
|
||||
|
||||
@AfterTest
|
||||
fun tearDown() {
|
||||
// Defensive: инициализация могла не дойти (например, SQLiteException
|
||||
// в @BeforeTest — connection создан, chatMap — нет).
|
||||
runCatching { if (::bridge.isInitialized) bridge.uninstall(fakeAgent) }
|
||||
runCatching { if (::scope.isInitialized) scope.cancel() }
|
||||
runCatching { if (::chatMap.isInitialized) chatMap.close() }
|
||||
@@ -152,22 +137,22 @@ class TelegramBridgeComponentTest {
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
// ============================================================================
|
||||
// Inbound: маппинг chat ↔ conv, conv.send, typing ДО send
|
||||
// ============================================================================
|
||||
|
||||
@Test
|
||||
fun `first incoming text creates a persistent conversation and forwards the text`() = runBlocking {
|
||||
// persistent=true (default): первый Update.message → createConversation(temp=false),
|
||||
// chat_map сохраняет chat_id → conv_id, conv.send(Content.Text(text)) дёрнут.
|
||||
bridge = newBridge(scope)
|
||||
bridge.install(fakeAgent)
|
||||
val incoming = subscribe(scope, bridge.incomingFlow())
|
||||
try {
|
||||
val typingBefore = fakeTg.chatActions.size
|
||||
fakeTg.push(makeUpdate(updateId = 1, chatId = 100, text = "hi"))
|
||||
|
||||
val ev = incoming.awaitOne({ e: TelegramBridgeComponent.IncomingEvent -> e.text == "hi" })
|
||||
val ev = incoming.awaitOne({ it.text == "hi" })
|
||||
assertEquals("hi", ev.text)
|
||||
assertEquals("100", ev.chatId)
|
||||
// Та же convId выдаётся для одного чата.
|
||||
val convId = ev.convId
|
||||
|
||||
// Маппинг зафиксирован.
|
||||
@@ -179,6 +164,13 @@ class TelegramBridgeComponentTest {
|
||||
val sent = fakeAgent.conversations.getValue(convId).sent
|
||||
assertEquals(1, sent.size)
|
||||
assertEquals("hi", (sent[0][0] as Content.Text).body)
|
||||
|
||||
// Typing в чат ушёл ДО conv.send (точнее, ДО того, как send
|
||||
// завершится — но в тесте мост его шлёт сразу в handleUpdate
|
||||
// синхронно, а send — fire-and-forget launch'ом).
|
||||
val newTypings = fakeTg.chatActions.drop(typingBefore)
|
||||
.filter { it.first == "100" && it.second == SendChatEvent.Action.TYPING }
|
||||
assertEquals(1, newTypings.size, "expected exactly one TYPING for chat 100, got $newTypings")
|
||||
} finally {
|
||||
bridge.uninstall(fakeAgent)
|
||||
}
|
||||
@@ -186,18 +178,17 @@ class TelegramBridgeComponentTest {
|
||||
|
||||
@Test
|
||||
fun `incoming text from same chat reuses the same conversation`() = runBlocking {
|
||||
// Два сообщения от одного chatId — маппинг один и тот же, новых createConversation нет.
|
||||
bridge = newBridge(scope)
|
||||
bridge.install(fakeAgent)
|
||||
val incoming = subscribe(scope, bridge.incomingFlow())
|
||||
try {
|
||||
fakeTg.push(makeUpdate(updateId = 1, chatId = 200, text = "first"))
|
||||
val first = incoming.awaitOne({ e: TelegramBridgeComponent.IncomingEvent -> e.text == "first" })
|
||||
val first = incoming.awaitOne({ it.text == "first" })
|
||||
val firstConv = first.convId
|
||||
// Ждём, пока bridge обработает (poll вернётся пустым после Update).
|
||||
fakeTg.drain()
|
||||
fakeTg.push(makeUpdate(updateId = 2, chatId = 200, text = "second"))
|
||||
incoming.awaitOne({ e: TelegramBridgeComponent.IncomingEvent -> e.text == "second" })
|
||||
incoming.awaitOne({ it.text == "second" })
|
||||
|
||||
// createConversation вызвался ровно один раз.
|
||||
assertEquals(1, fakeAgent.created.size)
|
||||
// Оба сообщения ушли в один conv.
|
||||
@@ -212,8 +203,6 @@ class TelegramBridgeComponentTest {
|
||||
|
||||
@Test
|
||||
fun `non-persistent config creates a fresh conversation per message`() = runBlocking {
|
||||
// persistentConversations=false: каждое сообщение = новая (temp) беседа,
|
||||
// маппинг всё равно обновляется на последнюю.
|
||||
bridge = newBridge(
|
||||
scope = scope,
|
||||
config = TelegramConfig(token = "test", pollingTimeoutSec = 0, persistentConversations = false),
|
||||
@@ -222,96 +211,207 @@ class TelegramBridgeComponentTest {
|
||||
val incoming = subscribe(scope, bridge.incomingFlow())
|
||||
try {
|
||||
fakeTg.push(makeUpdate(updateId = 1, chatId = 300, text = "a"))
|
||||
val first = incoming.awaitOne({ e: TelegramBridgeComponent.IncomingEvent -> e.text == "a" })
|
||||
val first = incoming.awaitOne({ it.text == "a" })
|
||||
fakeTg.drain()
|
||||
fakeTg.push(makeUpdate(updateId = 2, chatId = 300, text = "b"))
|
||||
val second = incoming.awaitOne({ e: TelegramBridgeComponent.IncomingEvent -> e.text == "b" })
|
||||
val second = incoming.awaitOne({ it.text == "b" })
|
||||
|
||||
// Два разных conv (temp).
|
||||
assertEquals(2, fakeAgent.created.size)
|
||||
assertEquals(true, fakeAgent.created.all { it.temp })
|
||||
// Маппинг указывает на последний.
|
||||
assertEquals(second.convId, chatMap.getConversationId("300"))
|
||||
// Первая convId больше не текущая.
|
||||
assertTrue(first.convId != second.convId)
|
||||
} finally {
|
||||
bridge.uninstall(fakeAgent)
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Typing: refresh на каждом online-событии
|
||||
// ============================================================================
|
||||
|
||||
@Test
|
||||
fun `typing chat action is sent on every incoming text`() = runBlocking {
|
||||
fun `typing chat action is throttled and only one TYPING per chat in burst`() = runBlocking {
|
||||
// Telegram режет rate-limit на sendChatAction — мост троттлит
|
||||
// до ≤1 запроса в 4 секунды на chatId. Здесь шлём 2 входящих
|
||||
// и 5 online-событий подряд — должны получить ровно 1 TYPING
|
||||
// (сразу на первое входящее), всё остальное подавлено.
|
||||
bridge = newBridge(scope)
|
||||
bridge.install(fakeAgent)
|
||||
val incoming: MutableList<TelegramBridgeComponent.IncomingEvent> = subscribe(scope, bridge.incomingFlow())
|
||||
val incoming = subscribe(scope, bridge.incomingFlow())
|
||||
try {
|
||||
fakeTg.push(makeUpdate(updateId = 1, chatId = 400, text = "ping"))
|
||||
val ev = incoming.awaitOne({ it.text == "ping" })
|
||||
val convId = ev.convId
|
||||
fakeTg.drain()
|
||||
fakeTg.push(makeUpdate(updateId = 2, chatId = 400, text = "pong"))
|
||||
incoming.awaitOne({ e: TelegramBridgeComponent.IncomingEvent -> e.text == "pong" })
|
||||
incoming.awaitOne({ it.text == "pong" })
|
||||
fakeTg.drain()
|
||||
|
||||
val typing = fakeTg.chatActions.filter { it.second == SendChatEvent.Action.TYPING }
|
||||
assertTrue(typing.size >= 2, "expected ≥2 TYPING actions, got ${fakeTg.chatActions}")
|
||||
assertTrue(typing.all { it.first == "400" }, "expected all TYPING to chat 400, got $typing")
|
||||
val typings = fakeTg.chatActions.filter { it.first == "400" && it.second == SendChatEvent.Action.TYPING }
|
||||
assertEquals(1, typings.size, "burst of two messages must collapse to one TYPING, got $typings")
|
||||
|
||||
// Online-события подряд — все подавлены троттлом.
|
||||
val baseline = typings.size
|
||||
val now = Instant.fromEpochMilliseconds(1_700_000_000_000)
|
||||
val out = fakeAgent.onlineOutbox
|
||||
out.emit(OnlineEvent.Working(now, convId))
|
||||
out.emit(OnlineEvent.StartResponse(now, convId, OnlineEvent.ResponseType.TEXT))
|
||||
out.emit(OnlineEvent.AppendText(now, convId, "a"))
|
||||
out.emit(OnlineEvent.AppendText(now, convId, "b"))
|
||||
out.emit(OnlineEvent.End(now, convId))
|
||||
// Даём корутинам время отработать burst, но без ожидания
|
||||
// конкретного счётчика — ожидаем ровно 0 новых TYPING.
|
||||
kotlinx.coroutines.delay(100)
|
||||
val refreshed = fakeTg.chatActions
|
||||
.drop(baseline)
|
||||
.filter { it.first == "400" && it.second == SendChatEvent.Action.TYPING }
|
||||
assertEquals(0, refreshed.size, "online burst within throttle window must not send more TYPING, got $refreshed")
|
||||
} finally {
|
||||
bridge.uninstall(fakeAgent)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `outbound streaming renders answer via editMessage edits`() = runBlocking {
|
||||
fun `typing chat action is refreshed after throttle interval expires`() = runBlocking {
|
||||
// Мост шлёт TYPING сразу на первое входящее, потом пропускает
|
||||
// burst online-событий; следующее входящее ПОСЛЕ 4s троттла
|
||||
// должно освежить TYPING.
|
||||
bridge = newBridge(scope)
|
||||
bridge.install(fakeAgent)
|
||||
val incoming = subscribe(scope, bridge.incomingFlow())
|
||||
try {
|
||||
fakeTg.push(makeUpdate(updateId = 1, chatId = 500, text = "hi"))
|
||||
incoming.awaitOne({ it.text == "hi" })
|
||||
fakeTg.drain()
|
||||
val first = fakeTg.chatActions.count { it.first == "500" && it.second == SendChatEvent.Action.TYPING }
|
||||
assertEquals(1, first)
|
||||
|
||||
// Провалидируем мост: между двумя входящими должен пройти
|
||||
// TYPING_REFRESH_MIN_INTERVAL_MS — иначе второе входящее тоже
|
||||
// схлопнется в первый TYPING. Здесь «время» управляется через
|
||||
// ручное продвижение в `sendTypingThrottled` — мы проверяем,
|
||||
// что МОСТ после интервала снова шлёт TYPING (эмулируем,
|
||||
// подменив `lastTypingSentAt` через отражение/private API не
|
||||
// делаем — это покрывается end-to-end сценарием ниже).
|
||||
// Реальный прогон: меняем internal-флаг через рефлексию и
|
||||
// шлём второе входящее.
|
||||
val field = bridge.javaClass.getDeclaredField("lastTypingSentAt").apply { isAccessible = true }
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
val map = field.get(bridge) as MutableMap<String, Long>
|
||||
map["500"] = 0L // сбрасываем «последний раз» — троттл сразу разрешает
|
||||
|
||||
fakeTg.push(makeUpdate(updateId = 2, chatId = 500, text = "again"))
|
||||
incoming.awaitOne({ it.text == "again" })
|
||||
fakeTg.drain()
|
||||
val second = fakeTg.chatActions.count { it.first == "500" && it.second == SendChatEvent.Action.TYPING }
|
||||
assertEquals(2, second, "after throttle resets, next incoming must send fresh TYPING")
|
||||
} finally {
|
||||
bridge.uninstall(fakeAgent)
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Outbound: DurableEvent.AssistantMessage → sendMessage(полный текст)
|
||||
// ============================================================================
|
||||
|
||||
@Test
|
||||
fun `DurableEvent AssistantMessage sends full text as a single message`() = runBlocking {
|
||||
bridge = newBridge(scope)
|
||||
bridge.install(fakeAgent)
|
||||
val incoming = subscribe(scope, bridge.incomingFlow())
|
||||
try {
|
||||
fakeTg.push(makeUpdate(updateId = 1, chatId = 500, text = "ask"))
|
||||
val ev = incoming.awaitOne({ e: TelegramBridgeComponent.IncomingEvent -> e.text == "ask" })
|
||||
val ev = incoming.awaitOne({ it.text == "ask" })
|
||||
val convId = ev.convId
|
||||
|
||||
val now = Instant.fromEpochMilliseconds(1_700_000_000_000)
|
||||
val out = fakeAgent.onlineOutbox
|
||||
out.emit(OnlineEvent.StartResponse(now, convId, OnlineEvent.ResponseType.TEXT))
|
||||
out.emit(OnlineEvent.AppendText(now, convId, "Hello"))
|
||||
out.emit(OnlineEvent.AppendText(now, conversationId = convId, body = ", "))
|
||||
out.emit(OnlineEvent.AppendText(now, conversationId = convId, body = "world!"))
|
||||
out.emit(OnlineEvent.End(now, convId))
|
||||
fakeTg.waitForEdits(3)
|
||||
|
||||
val drafts = fakeTg.sentMessages.filter { it.chatId == "500" && it.text == "..." }
|
||||
assertEquals(1, drafts.size, "expected exactly one draft '...' sent, got ${fakeTg.sentMessages}")
|
||||
val edits = fakeTg.edited.filter { it.chatId == "500" }
|
||||
assertTrue(edits.size >= 3, "expected ≥3 edits (one per AppendText), got ${edits.size}")
|
||||
val lastText = edits.last().text
|
||||
assertTrue("Hello" in lastText && "world!" in lastText, "final edit must contain full text: $lastText")
|
||||
// Полный текст приходит в DurableEvent.AssistantMessage.
|
||||
fakeAgent.outbox.emitDurable(
|
||||
convId = convId,
|
||||
event = DurableEvent.AssistantMessage(
|
||||
date = now,
|
||||
id = "msg-1",
|
||||
content = listOf(Content.Text("Hello, world!")),
|
||||
),
|
||||
)
|
||||
fakeTg.waitForMessageContaining("Hello, world!")
|
||||
|
||||
// Ровно ОДНО сообщение, никаких drafts / edits.
|
||||
val final = fakeTg.sentMessages.filter { it.chatId == "500" && it.text == "Hello, world!" }
|
||||
assertEquals(1, final.size, "expected one sendMessage with full text, got ${fakeTg.sentMessages}")
|
||||
// Никаких editMessage'й не было.
|
||||
assertEquals(0, fakeTg.edited.size, "editMessage must not be used, got ${fakeTg.edited}")
|
||||
} finally {
|
||||
bridge.uninstall(fakeAgent)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `End with empty buffer replaces draft with (empty response)`() = runBlocking {
|
||||
fun `AssistantMessage with markdown is sent as Telegram HTML with parseMode`() = runBlocking {
|
||||
bridge = newBridge(scope)
|
||||
bridge.install(fakeAgent)
|
||||
val incoming = subscribe(scope, bridge.incomingFlow())
|
||||
try {
|
||||
fakeTg.push(makeUpdate(updateId = 1, chatId = 550, text = "ask"))
|
||||
val ev = incoming.awaitOne({ it.text == "ask" })
|
||||
val convId = ev.convId
|
||||
val now = Instant.fromEpochMilliseconds(1_700_000_000_000)
|
||||
fakeAgent.outbox.emitDurable(
|
||||
convId = convId,
|
||||
event = DurableEvent.AssistantMessage(
|
||||
date = now,
|
||||
id = "msg-md",
|
||||
content = listOf(Content.Text("Hello **world** — see [docs](https://e.com)")),
|
||||
),
|
||||
)
|
||||
// Ждём конкретный HTML, а не plain.
|
||||
fakeTg.waitForHtmlContaining("550", "<b>world</b>")
|
||||
|
||||
val sent = fakeTg.sentMessages.filter { it.chatId == "550" }
|
||||
assertEquals(1, sent.size, "expected one sendMessage, got ${fakeTg.sentMessages}")
|
||||
assertEquals("HTML", sent[0].parseMode?.code, "parseMode must be HTML, got ${sent[0].parseMode}")
|
||||
assertEquals(
|
||||
"Hello <b>world</b> — see <a href=\"https://e.com\">docs</a>",
|
||||
sent[0].text,
|
||||
)
|
||||
} finally {
|
||||
bridge.uninstall(fakeAgent)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `AssistantMessage with empty content falls back to (empty response)`() = runBlocking {
|
||||
bridge = newBridge(scope)
|
||||
bridge.install(fakeAgent)
|
||||
val incoming = subscribe(scope, bridge.incomingFlow())
|
||||
try {
|
||||
fakeTg.push(makeUpdate(updateId = 1, chatId = 600, text = "ask"))
|
||||
val ev = incoming.awaitOne({ e: TelegramBridgeComponent.IncomingEvent -> e.text == "ask" })
|
||||
val ev = incoming.awaitOne({ it.text == "ask" })
|
||||
val convId = ev.convId
|
||||
val now = Instant.fromEpochMilliseconds(1_700_000_000_000)
|
||||
val out = fakeAgent.onlineOutbox
|
||||
// StartResponse → draft "..." создан. Сразу End без AppendText → buffer пустой.
|
||||
out.emit(OnlineEvent.StartResponse(now, convId, OnlineEvent.ResponseType.TEXT))
|
||||
out.emit(OnlineEvent.End(now, convId))
|
||||
fakeTg.waitForEditContaining("(empty response)")
|
||||
fakeAgent.outbox.emitDurable(
|
||||
convId = convId,
|
||||
event = DurableEvent.AssistantMessage(
|
||||
date = now,
|
||||
id = "msg-empty",
|
||||
content = emptyList(),
|
||||
),
|
||||
)
|
||||
fakeTg.waitForMessageContaining("(empty response)")
|
||||
|
||||
val emptyEdits = fakeTg.edited.filter { it.chatId == "600" && it.text == "(empty response)" }
|
||||
assertEquals(1, emptyEdits.size, "expected one (empty response) edit, got ${fakeTg.edited}")
|
||||
val fallback = fakeTg.sentMessages.filter {
|
||||
it.chatId == "600" && it.text == "(empty response)"
|
||||
}
|
||||
assertEquals(1, fallback.size, "expected one (empty response), got ${fakeTg.sentMessages}")
|
||||
} finally {
|
||||
bridge.uninstall(fakeAgent)
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Errors / lifecycle
|
||||
// ============================================================================
|
||||
|
||||
@Test
|
||||
fun `conv send error is reported to the chat as a single line`() = runBlocking {
|
||||
bridge = newBridge(scope)
|
||||
@@ -319,10 +419,10 @@ class TelegramBridgeComponentTest {
|
||||
val incoming = subscribe(scope, bridge.incomingFlow())
|
||||
try {
|
||||
fakeTg.push(makeUpdate(updateId = 1, chatId = 700, text = "warmup"))
|
||||
val warmup = incoming.awaitOne({ e: TelegramBridgeComponent.IncomingEvent -> e.text == "warmup" })
|
||||
val warmup = incoming.awaitOne({ it.text == "warmup" })
|
||||
fakeAgent.conversations.getValue(warmup.convId).sendError = IllegalStateException("model offline")
|
||||
fakeTg.push(makeUpdate(updateId = 2, chatId = 700, text = "boom"))
|
||||
incoming.awaitOne({ e: TelegramBridgeComponent.IncomingEvent -> e.text == "boom" })
|
||||
incoming.awaitOne({ it.text == "boom" })
|
||||
fakeTg.waitForMessageContaining("⚠️ agent error: model offline")
|
||||
|
||||
val errors = fakeTg.sentMessages.filter {
|
||||
@@ -334,6 +434,31 @@ class TelegramBridgeComponentTest {
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `DurableEvent Error is forwarded to the chat as a warning line`() = runBlocking {
|
||||
bridge = newBridge(scope)
|
||||
bridge.install(fakeAgent)
|
||||
val incoming = subscribe(scope, bridge.incomingFlow())
|
||||
try {
|
||||
fakeTg.push(makeUpdate(updateId = 1, chatId = 750, text = "ask"))
|
||||
val ev = incoming.awaitOne({ it.text == "ask" })
|
||||
val convId = ev.convId
|
||||
val now = Instant.fromEpochMilliseconds(1_700_000_000_000)
|
||||
fakeAgent.outbox.emitDurable(
|
||||
convId = convId,
|
||||
event = DurableEvent.Error(date = now, message = "context overflow", code = "ctx"),
|
||||
)
|
||||
fakeTg.waitForMessageContaining("⚠️ context overflow")
|
||||
|
||||
val errs = fakeTg.sentMessages.filter {
|
||||
it.chatId == "750" && it.text == "⚠️ context overflow"
|
||||
}
|
||||
assertEquals(1, errs.size, "expected one ⚠️ error line, got ${fakeTg.sentMessages}")
|
||||
} finally {
|
||||
bridge.uninstall(fakeAgent)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `uninstall cancels polling and no further updates are processed`() = runBlocking {
|
||||
bridge = newBridge(scope)
|
||||
@@ -341,7 +466,7 @@ class TelegramBridgeComponentTest {
|
||||
val incoming = subscribe(scope, bridge.incomingFlow())
|
||||
try {
|
||||
fakeTg.push(makeUpdate(updateId = 1, chatId = 800, text = "alive"))
|
||||
incoming.awaitOne({ e: TelegramBridgeComponent.IncomingEvent -> e.text == "alive" })
|
||||
incoming.awaitOne({ it.text == "alive" })
|
||||
bridge.uninstall(fakeAgent)
|
||||
|
||||
val sentBefore = fakeAgent.conversations.values.sumOf { it.sent.size }
|
||||
@@ -354,9 +479,61 @@ class TelegramBridgeComponentTest {
|
||||
}
|
||||
}
|
||||
|
||||
// Fakes
|
||||
// ============================================================================
|
||||
@Test
|
||||
fun `Telegram rejection of AssistantMessage is swallowed (polling survives)`() = runBlocking {
|
||||
bridge = newBridge(scope)
|
||||
bridge.install(fakeAgent)
|
||||
val incoming = subscribe(scope, bridge.incomingFlow())
|
||||
try {
|
||||
fakeTg.push(makeUpdate(updateId = 1, chatId = 700, text = "ask"))
|
||||
val ev = incoming.awaitOne({ it.text == "ask" })
|
||||
val convId = ev.convId
|
||||
val now = Instant.fromEpochMilliseconds(1_700_000_000_000)
|
||||
|
||||
// Имитируем TelegramException (400 Bad Request: can't parse entities).
|
||||
// Мост должен:
|
||||
// - не упасть (polling выживает)
|
||||
// - залогировать ошибку через slf4j (log.error("sendMessage(HTML) failed ..."))
|
||||
// - не отправить сообщение в чат
|
||||
fakeTg.failNextSend = pw.binom.telegram.TelegramException(
|
||||
code = 400,
|
||||
description = "Bad Request: can't parse entities",
|
||||
)
|
||||
fakeAgent.outbox.emitDurable(
|
||||
convId = convId,
|
||||
event = DurableEvent.AssistantMessage(
|
||||
date = now,
|
||||
id = "msg-bad",
|
||||
content = listOf(Content.Text("**boom**")),
|
||||
),
|
||||
)
|
||||
// Дать корутине шанс обработать исключение.
|
||||
kotlinx.coroutines.delay(150)
|
||||
|
||||
// Сообщение не дошло (Telegram отверг).
|
||||
assertTrue(
|
||||
fakeTg.sentMessages.none { it.chatId == "700" && "boom" in it.text },
|
||||
"Telegram отверг сообщение — в sentMessages его быть не должно",
|
||||
)
|
||||
|
||||
// Polling выжил: следующий AssistantMessage уходит нормально.
|
||||
fakeAgent.outbox.emitDurable(
|
||||
convId = convId,
|
||||
event = DurableEvent.AssistantMessage(
|
||||
date = now,
|
||||
id = "msg-ok",
|
||||
content = listOf(Content.Text("hi again")),
|
||||
),
|
||||
)
|
||||
fakeTg.waitForMessageContaining("hi again")
|
||||
assertTrue(
|
||||
fakeTg.sentMessages.any { it.chatId == "700" && "hi again" in it.text },
|
||||
"после ошибки мост продолжил работать",
|
||||
)
|
||||
} finally {
|
||||
bridge.uninstall(fakeAgent)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -364,15 +541,10 @@ class TelegramBridgeComponentTest {
|
||||
* методы пишут в лог-структуры.
|
||||
*/
|
||||
class FakeTelegramClient : TelegramClient {
|
||||
/** Запросы `getUpdate`, ожидающие обработки polling'ом. */
|
||||
private val queue = Channel<List<Update>>(capacity = Channel.UNLIMITED)
|
||||
|
||||
/** «Опубликовать» пачку апдейтов — следующий getUpdate её заберёт. */
|
||||
fun push(vararg updates: Update) {
|
||||
queue.trySend(updates.toList())
|
||||
}
|
||||
fun push(vararg updates: Update) { queue.trySend(updates.toList()) }
|
||||
|
||||
/** Слить все текущие «опубликованные» апдейты (после теста). */
|
||||
suspend fun drain() {
|
||||
while (true) {
|
||||
val r = queue.tryReceive()
|
||||
@@ -385,6 +557,9 @@ class FakeTelegramClient : TelegramClient {
|
||||
val chatActions: MutableList<Pair<String, SendChatEvent.Action>> = mutableListOf()
|
||||
val deleted: MutableList<Pair<String, Long>> = mutableListOf()
|
||||
|
||||
/** Если задано — следующий `sendMessage` бросит [RuntimeException] (имитация 400/403/429). */
|
||||
var failNextSend: RuntimeException? = null
|
||||
|
||||
private val messageIdSeq = java.util.concurrent.atomic.AtomicLong(1000L)
|
||||
|
||||
override suspend fun getUpdate(
|
||||
@@ -395,6 +570,11 @@ class FakeTelegramClient : TelegramClient {
|
||||
): List<Update> = queue.receive()
|
||||
|
||||
override suspend fun sendMessage(message: TextMessage): Message {
|
||||
failNextSend?.let {
|
||||
val e = it
|
||||
failNextSend = null
|
||||
throw e
|
||||
}
|
||||
sentMessages.add(message)
|
||||
return Message(
|
||||
messageId = messageIdSeq.getAndIncrement(),
|
||||
@@ -422,27 +602,15 @@ class FakeTelegramClient : TelegramClient {
|
||||
return true
|
||||
}
|
||||
|
||||
suspend fun waitForEdits(minCount: Int) {
|
||||
withTimeout(2_000) {
|
||||
while (edited.size < minCount) {
|
||||
kotlinx.coroutines.delay(10)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
suspend fun waitForEditContaining(text: String) {
|
||||
withTimeout(2_000) {
|
||||
while (edited.none { text in it.text }) {
|
||||
kotlinx.coroutines.delay(10)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
suspend fun waitForMessageContaining(text: String) {
|
||||
withTimeout(2_000) {
|
||||
while (sentMessages.none { text in it.text }) {
|
||||
kotlinx.coroutines.delay(10)
|
||||
}
|
||||
while (sentMessages.none { text in it.text }) kotlinx.coroutines.delay(10)
|
||||
}
|
||||
}
|
||||
|
||||
suspend fun waitForHtmlContaining(chatId: String, html: String) {
|
||||
withTimeout(2_000) {
|
||||
while (sentMessages.none { it.chatId == chatId && html in it.text }) kotlinx.coroutines.delay(10)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -467,10 +635,18 @@ class FakeTelegramClient : TelegramClient {
|
||||
}
|
||||
|
||||
/**
|
||||
* Минимальный `MutableAgent`: создаёт `FakeConversation`, помнит их по id.
|
||||
* `onlineOutbox` управляется из теста (`FakeOnlineOutbox`).
|
||||
* Минимальный `MutableAgent` для тестов моста.
|
||||
*
|
||||
* - `onlineOutbox` / `outbox` управляются из теста (тест пушит события
|
||||
* через `emit(...)` / `emitDurable(...)`, мост слушает `onlineEvents(convId)`
|
||||
* и `conversationEvents(convId)`).
|
||||
* - `createConversation` возвращает `FakeConversation` с управляемой
|
||||
* отправкой (`sendError`).
|
||||
*/
|
||||
class FakeMutableAgent(override val onlineOutbox: FakeOnlineOutbox) : MutableAgent {
|
||||
class FakeMutableAgent : MutableAgent {
|
||||
override val onlineOutbox: FakeOnlineOutbox = FakeOnlineOutbox()
|
||||
override val outbox: FakeOutboxStore = FakeOutboxStore()
|
||||
|
||||
override val systemProviders: MutableList<pw.binom.agentik.agent.SystemPromptProvider> = mutableListOf()
|
||||
override val toolProviders: MutableList<pw.binom.agentik.agent.ToolProvider> = mutableListOf()
|
||||
val conversations: MutableMap<String, FakeConversation> = mutableMapOf()
|
||||
@@ -483,7 +659,6 @@ class FakeMutableAgent(override val onlineOutbox: FakeOnlineOutbox) : MutableAge
|
||||
override val id: String = "test-agent"
|
||||
override val info: AgentInfo = AgentInfo(name = "test", description = "", usefulness = "")
|
||||
override val journal: JournalStore by lazy { error("journal not used in bridge test") }
|
||||
override val outbox: OutboxStore by lazy { error("outbox not used in bridge test") }
|
||||
override val conversationStore: ConversationStore by lazy { error("conversationStore not used in bridge test") }
|
||||
|
||||
override fun createConversation(temp: Boolean): Conversation {
|
||||
@@ -510,7 +685,8 @@ class FakeMutableAgent(override val onlineOutbox: FakeOnlineOutbox) : MutableAge
|
||||
}
|
||||
|
||||
/**
|
||||
* Управляемый `OnlineOutbox` — тест пушит OnlineEvent'ы, мост подписан на `onlineEvents(convId)`.
|
||||
* Управляемый `OnlineOutbox`: тест эмитит `OnlineEvent` через [emit],
|
||||
* мост слушает `onlineEvents(convId)`.
|
||||
*/
|
||||
class FakeOnlineOutbox : OnlineOutbox {
|
||||
private val bus = MutableSharedFlow<OnlineEvent>(replay = 0, extraBufferCapacity = 64)
|
||||
@@ -524,13 +700,37 @@ class FakeOnlineOutbox : OnlineOutbox {
|
||||
override fun close() {}
|
||||
}
|
||||
|
||||
/**
|
||||
* Управляемый `OutboxStore`: тест эмитит durable-события для конкретного
|
||||
* conv'а через [emitDurable], мост слушает `conversationEvents(convId)`.
|
||||
*/
|
||||
class FakeOutboxStore : OutboxStore {
|
||||
private val bus = MutableSharedFlow<CommonEvent>(replay = 0, extraBufferCapacity = 64)
|
||||
|
||||
suspend fun emitDurable(convId: String, event: DurableEvent) {
|
||||
bus.emit(CommonEvent.Conversation(date = event.date, conversationId = convId, event = event))
|
||||
}
|
||||
|
||||
override fun events(): Flow<CommonEvent> = bus.asSharedFlow()
|
||||
override fun conversationEvents(conversationId: String?): Flow<CommonEvent.Conversation> =
|
||||
bus.asSharedFlow()
|
||||
.filterIsInstance<CommonEvent.Conversation>()
|
||||
.let { filtered ->
|
||||
if (conversationId == null) filtered
|
||||
else filtered.filter { it.conversationId == conversationId }
|
||||
}
|
||||
override fun agentEvents(): Flow<CommonEvent.Agent> =
|
||||
bus.asSharedFlow().filterIsInstance<CommonEvent.Agent>()
|
||||
|
||||
override fun close() {}
|
||||
}
|
||||
|
||||
class FakeConversation(
|
||||
override val id: String,
|
||||
override val isTemporal: Boolean = false,
|
||||
) : Conversation {
|
||||
val sent: MutableList<List<Content>> = mutableListOf()
|
||||
var interrupted = false
|
||||
/** Если задано, следующий `send` кинет это исключение. */
|
||||
var sendError: Throwable? = null
|
||||
|
||||
override val isSupportImageInput: Boolean = false
|
||||
@@ -549,23 +749,7 @@ class FakeConversation(
|
||||
interrupted = true
|
||||
}
|
||||
|
||||
override suspend fun getMessages(after: Instant, offset: Int, limit: Int) = emptyList<pw.binom.agentik.proto.Message>()
|
||||
override suspend fun getMessages(after: Instant, offset: Int, limit: Int) =
|
||||
emptyList<pw.binom.agentik.proto.Message>()
|
||||
override fun close() {}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Flow helpers
|
||||
// ============================================================================
|
||||
|
||||
private suspend fun <T> Flow<T>.firstAsync(predicate: (T) -> Boolean = { true }): T =
|
||||
kotlinx.coroutines.withTimeout(2_000) {
|
||||
this@firstAsync.first { predicate(it) }
|
||||
}
|
||||
|
||||
private suspend fun <T> Flow<T>.firstAsyncOrNull(predicate: (T) -> Boolean): T? = try {
|
||||
kotlinx.coroutines.withTimeout(1_000) {
|
||||
this@firstAsyncOrNull.first { predicate(it) }
|
||||
}
|
||||
} catch (_: kotlinx.coroutines.TimeoutCancellationException) {
|
||||
null
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user