From 7901e72325868d942a726966a2d9e33cda0d685d Mon Sep 17 00:00:00 2001 From: q Date: Sun, 18 Oct 2026 08:31:47 +0000 Subject: [PATCH 1/2] Settle QA Telegram probes on a 3 s quiet window under the wait bound --- docs/CHANGELOG.md | 5 + docs/contracts/kit-template-and-qa.md | 10 + services/langgraph/src/prompts/qa/__init__.py | 7 +- .../tests/unit/test_qa_async_replies.py | 10 +- shared/telegram_access_probe.py | 20 +- shared/telegram_bot_probe.py | 66 +- .../tests/unit/test_telegram_access_probe.py | 152 ++++ shared/tests/unit/test_telegram_bot_probe.py | 660 ++++++------------ .../unit/test_telegram_probe_injection.py | 2 +- 9 files changed, 434 insertions(+), 498 deletions(-) create mode 100644 shared/tests/unit/test_telegram_access_probe.py diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 3abbcce00..bd7caac5f 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -3,6 +3,11 @@ One `## YYYY-MM-DD` heading per merge day, newest first; one bullet of at most two lines per entry. See the CHANGELOG rule in [AGENTS.md](../AGENTS.md). +## 2026-10-18 + +- QA Telegram probes, button presses and the `/start` preflight return once the bot's response stays + unchanged for 3 s; the wait stays the upper bound and silence still runs to it. + ## 2026-10-17 - Kit 14fa729c (core 2.6.0) is the scaffold, tooling and catalog pin, so a new product's bot registry diff --git a/docs/contracts/kit-template-and-qa.md b/docs/contracts/kit-template-and-qa.md index b682fb3a8..42a686fae 100644 --- a/docs/contracts/kit-template-and-qa.md +++ b/docs/contracts/kit-template-and-qa.md @@ -688,6 +688,16 @@ reports which path settled the Run and claims nothing about the executor. Telegram message and callback tools accept an integer collection wait of 1–60 seconds, default 15; their child process timeout is that wait plus 30 seconds. Invalid waits return a typed `invalid_wait` refusal without delivery or a run blocker. +The wait is an upper bound. The tool polls every 2 seconds and returns once a bot +response has stayed unchanged for a 3-second quiet window, or at the wait, with a +final read before returning. A message probe's response is any bot message after the +sent one; a button press's response is a message after the press or a change to the +pressed message. A new message or a change to the serialized text, caption, media type +or button markup restarts the window; a callback answer alone, even a non-empty one +that is kept as evidence, does not. Silence runs to the wait. Quiet ends observation; +it does not prove a final answer, and a reply sent after the tool returned is absent +from its evidence. The `/start` access preflight settles the same way on its replies +with a 10-second bound, and a refusal seen in any read still blocks the run. At the single runner settlement point, an executor-declared `product` cause is always preserved. Settlement moves only failed `qa_capability` / `qa_tooling` rows to unverified; brief text and optional `telegram_step` evidence never diff --git a/services/langgraph/src/prompts/qa/__init__.py b/services/langgraph/src/prompts/qa/__init__.py index 73105bd62..7f44989d0 100644 --- a/services/langgraph/src/prompts/qa/__init__.py +++ b/services/langgraph/src/prompts/qa/__init__.py @@ -112,8 +112,11 @@ def _bot_section(bot_username: str) -> str: every resulting bot reply, plus post-press evidence for the clicked message so an edit-in-place is observable. - The collection wait defaults to 15 seconds; choose an integer from 1 to 60. - Use a longer wait when a reply says work continues (pending/progress) or a - callback starts work. Judge the final answer after the chosen collection wait. + It is an upper bound: a call returns once the bot's replies (and, for a + button, the pressed message) stay unchanged for 3 seconds, and silence runs + to the wait. Use a longer wait when the bot needs longer for its first reply. + Quiet is not a final answer: a reply sent after the call returned is not in + its evidence, so a pending/progress reply is not the bot's final answer. An invalid wait returns a typed refusal and sends nothing; correct the value. - For a Telegram check, include `telegram_step` from the tool's result in its verdict row. This ties the check to the recorded input and collected replies. diff --git a/services/langgraph/tests/unit/test_qa_async_replies.py b/services/langgraph/tests/unit/test_qa_async_replies.py index 81c3df053..2a0aecc03 100644 --- a/services/langgraph/tests/unit/test_qa_async_replies.py +++ b/services/langgraph/tests/unit/test_qa_async_replies.py @@ -164,7 +164,7 @@ def get_messages(self, bot, *, ids=None, min_id=None, limit=None): return original if limit == 1: return [original] - return [final, photo, progress] if clock[0] >= 20 else [progress] + return [final, photo, progress][2 - (clock[0] >= 2) - (clock[0] >= 4) :] modules = { "telethon.sync": {"TelegramClient": Client}, @@ -185,12 +185,14 @@ def get_messages(self, bot, *, ids=None, min_id=None, limit=None): output = StringIO() with redirect_stdout(output): exec(script, {}) # noqa: S102 - exercise the generated child script - return QATelegramProbeEvidence.model_validate(parse_bot_probe_result(output.getvalue())) + evidence = parse_bot_probe_result(output.getvalue()) + return QATelegramProbeEvidence.model_validate(evidence), clock[0] @pytest.mark.parametrize("action", ["message", "callback"]) -def test_long_wait_collects_progress_delayed_photo_and_messages(monkeypatch, action): - evidence = _fake_bot(monkeypatch, action, 60) +def test_progress_photo_and_answer_settle_well_inside_a_long_wait(monkeypatch, action): + evidence, elapsed = _fake_bot(monkeypatch, action, 60) + assert elapsed == 7 assert [reply.id for reply in evidence.replies] == [11, 12, 13] assert evidence.replies[1].caption == "The cards" assert evidence.replies[2].text == "Your answer" diff --git a/shared/telegram_access_probe.py b/shared/telegram_access_probe.py index 728df6229..edbf14d07 100644 --- a/shared/telegram_access_probe.py +++ b/shared/telegram_access_probe.py @@ -19,9 +19,11 @@ from shared.contracts.bot_access import QA_TEST_TELEGRAM_ID from shared.contracts.dto.run_result import QABlocker, QABlockerCategory +from shared.telegram_bot_probe import settling_collector_source # How long the probe waits for the bot to answer /start before calling it silence. TELEGRAM_ACCESS_PROBE_TIMEOUT = 10 +TELEGRAM_ACCESS_PROBE_POLL_INTERVAL = 1 DENIAL_MARKER = "telegram_access_denied:" IDENTITY_MISMATCH_MARKER = "telegram_identity_mismatch:" @@ -64,7 +66,8 @@ def build_access_probe_script(bot_username: str) -> str: Exit 0 means the bot answered something other than a refusal, exit 2 a refusal, exit 3 the wrong Telegram account. The account is checked first: a - probe run from another identity would answer a question nobody asked. + probe run from another identity would answer a question nobody asked. A + refusal seen in any observation exits 2 before the answers settle. """ return ( "import os\n" @@ -72,7 +75,8 @@ def build_access_probe_script(bot_username: str) -> str: "from telethon.sessions import StringSession\n" "import sys\n" "import time\n" - "client = TelegramClient(StringSession(os.environ['TELETHON_SESSION']), " + + settling_collector_source() + + "client = TelegramClient(StringSession(os.environ['TELETHON_SESSION']), " "int(os.environ['TELETHON_API_ID']), os.environ['TELETHON_API_HASH'])\n" "client.start()\n" "try:\n" @@ -83,10 +87,9 @@ def build_access_probe_script(bot_username: str) -> str: " sys.exit(3)\n" f" bot = client.get_entity('@{bot_username}')\n" " sent = client.send_message(bot, '/start')\n" - f" deadline = time.monotonic() + {TELEGRAM_ACCESS_PROBE_TIMEOUT}\n" - " while time.monotonic() < deadline:\n" - " replies = client.get_messages(bot, min_id=sent.id, limit=5)\n" - " for reply in replies:\n" + " def observe():\n" + " answers = []\n" + " for reply in client.get_messages(bot, min_id=sent.id, limit=5):\n" " if reply.out or reply.id <= sent.id:\n" " continue\n" " text = (reply.raw_text or reply.message or '').strip()\n" @@ -97,7 +100,10 @@ def build_access_probe_script(bot_username: str) -> str: " if denied:\n" f" print('{DENIAL_MARKER}' + text[:500])\n" " sys.exit(2)\n" - " time.sleep(1)\n" + " answers.append((reply.id, text))\n" + " return answers\n" + f" collect_until_settled(observe, bool, {TELEGRAM_ACCESS_PROBE_TIMEOUT}, " + f"{TELEGRAM_ACCESS_PROBE_POLL_INTERVAL})\n" f" print('{PROBE_PASSED_MARKER}')\n" "finally:\n" " client.disconnect()" diff --git a/shared/telegram_bot_probe.py b/shared/telegram_bot_probe.py index f6820bb26..8011fd036 100644 --- a/shared/telegram_bot_probe.py +++ b/shared/telegram_bot_probe.py @@ -18,12 +18,15 @@ import json -# Overall deadline for collecting the bot's final visible state after a probe. +# Upper bound for collecting the bot's visible state after a probe. TELEGRAM_REPLY_TIMEOUT = 15 TELEGRAM_MAX_REPLY_TIMEOUT = 60 # Bound polling against the shared QA account while still reading exactly at the # collection deadline. TELEGRAM_POLL_INTERVAL = 2 +# Once the bot has responded, collection ends after its visible response stays +# unchanged this long. Quiet is an observation boundary, not a final answer. +TELEGRAM_QUIET_WINDOW = 3 MAX_REPLIES = 10 # The child gets connection and teardown room beyond its answer-collection # deadline. Keep this larger than TELEGRAM_REPLY_TIMEOUT. @@ -39,7 +42,7 @@ def telegram_probe_process_timeout(wait_seconds: int) -> int: def _script_helpers() -> str: - """The self-contained serializer shared by the message and callback scripts.""" + """The self-contained serializer and collector shared by the message and callback scripts.""" return """\ def encode_callback_data(value): if value is None: @@ -92,6 +95,35 @@ def received_replies(client, bot, minimum_id): def newest_message_id(client, bot): messages = client.get_messages(bot, limit=1) return messages[0].id if messages else 0 +""" + settling_collector_source() + + +def settling_collector_source() -> str: + """Script source of ``collect_until_settled(observe, responded, wait, poll)``. + + It polls ``observe()`` until ``responded`` holds and the observation has stayed + equal for the quiet window, or until ``wait`` seconds pass, and always returns + right after an observation. + """ + return f"""\ +def collect_until_settled(observe, responded, wait, poll): + deadline = time.monotonic() + wait + last = None + while True: + observed = observe() + now = time.monotonic() + if observed != last: + last = observed + changed_at = now + if now >= deadline: + return + wake = min(now + poll, deadline) + if responded(observed): + settled_at = changed_at + {TELEGRAM_QUIET_WINDOW} + if now >= settled_at: + return + wake = min(wake, settled_at) + time.sleep(wake - now) """ @@ -137,16 +169,10 @@ def build_bot_message_script( f" sent = client.send_message(bot, {json.dumps(message)})\n" " result['delivered'] = True\n" " result['message_id'] = sent.id\n" - f" deadline = time.monotonic() + {wait_seconds}\n" - " last_replies = []\n" - " while True:\n" - " now = time.monotonic()\n" + " def observe():\n" " result['replies'] = received_replies(client, bot, sent.id)\n" - " if result['replies'] != last_replies:\n" - " last_replies = result['replies']\n" - " if now >= deadline:\n" - " break\n" - " time.sleep(min(POLL_INTERVAL, deadline - now))\n" + " return result['replies']\n" + f" collect_until_settled(observe, bool, {wait_seconds}, POLL_INTERVAL)\n" "except Exception as exc:\n" " result['error'] = type(exc).__name__ + ': ' + str(exc)\n" "finally:\n" @@ -225,24 +251,18 @@ def build_bot_callback_script( " 'alert': bool(getattr(answer, 'alert', False)),\n" " 'url': getattr(answer, 'url', None),\n" " }\n" - f" deadline = time.monotonic() + {wait_seconds}\n" - " last_replies = []\n" - " last_post_press_message = pre_press_message\n" - " while True:\n" - " now = time.monotonic()\n" + " def observe():\n" " result['replies'] = received_replies(client, bot, reply_baseline)\n" " post_press = client.get_messages(bot, ids=message.id)\n" " result['post_press_message'] = (\n" " message_evidence(post_press)\n" " if post_press is not None and not post_press.out else None\n" " )\n" - " if (result['replies'] != last_replies\n" - " or result['post_press_message'] != last_post_press_message):\n" - " last_replies = result['replies']\n" - " last_post_press_message = result['post_press_message']\n" - " if now >= deadline:\n" - " break\n" - " time.sleep(min(POLL_INTERVAL, deadline - now))\n" + " return result['replies'], result['post_press_message']\n" + " def responded(observed):\n" + " replies, post_press_message = observed\n" + " return bool(replies) or post_press_message != pre_press_message\n" + f" collect_until_settled(observe, responded, {wait_seconds}, POLL_INTERVAL)\n" "except Exception as exc:\n" " result['error'] = type(exc).__name__ + ': ' + str(exc)\n" "finally:\n" diff --git a/shared/tests/unit/test_telegram_access_probe.py b/shared/tests/unit/test_telegram_access_probe.py new file mode 100644 index 000000000..cd21102ef --- /dev/null +++ b/shared/tests/unit/test_telegram_access_probe.py @@ -0,0 +1,152 @@ +"""The generated /start access probe, run on a virtual clock against a fake Telethon.""" + +from __future__ import annotations + +from contextlib import redirect_stdout +from io import StringIO +import sys +import time +from types import ModuleType, SimpleNamespace + +from shared.contracts.bot_access import QA_TEST_TELEGRAM_ID +from shared.contracts.dto.run_result import QABlockerCategory +from shared.telegram_access_probe import ( + PROBE_PASSED_MARKER, + build_access_probe_script, + classify_access_probe, +) + + +class _Bot: + """A bot whose answers after /start (message 10) are ``chat(now)``, oldest first.""" + + def __init__(self, chat, *, account_id=QA_TEST_TELEGRAM_ID): + self.now = 0 + self.chat = chat + self.account_id = account_id + self.reads: list[float] = [] + self.sent: list[str] = [] + self.disconnected = False + + def sleep(self, seconds): + assert seconds > 0 + self.now += seconds + + def client(self, *_args): + bot = self + + class Client: + def start(self): + return None + + def get_me(self): + return SimpleNamespace(id=bot.account_id) + + def get_entity(self, username): + return username + + def send_message(self, _entity, text): + bot.sent.append(text) + return SimpleNamespace(id=10) + + def get_messages(self, _entity, *, min_id, limit): + assert (min_id, limit) == (10, 5) + bot.reads.append(bot.now) + return list(reversed(bot.chat(bot.now))) + + def disconnect(self): + bot.disconnected = True + + return Client() + + +def _answer(identifier, text): + return SimpleNamespace(id=identifier, out=False, raw_text=text, message=text) + + +def _run(monkeypatch, bot: _Bot) -> tuple[int, str]: + monkeypatch.setattr(time, "monotonic", lambda: bot.now) + monkeypatch.setattr(time, "sleep", bot.sleep) + for name in ("TELETHON_SESSION", "TELETHON_API_HASH"): + monkeypatch.setenv(name, "fixture") + monkeypatch.setenv("TELETHON_API_ID", "123") + sync = ModuleType("telethon.sync") + sync.TelegramClient = bot.client + sessions = ModuleType("telethon.sessions") + sessions.StringSession = lambda value: value + monkeypatch.setitem(sys.modules, "telethon", ModuleType("telethon")) + monkeypatch.setitem(sys.modules, "telethon.sync", sync) + monkeypatch.setitem(sys.modules, "telethon.sessions", sessions) + stdout = StringIO() + exit_status = 0 + with redirect_stdout(stdout): + try: + exec(build_access_probe_script("fortune_bot"), {}) # noqa: S102 - generated probe + except SystemExit as exc: + exit_status = exc.code + return exit_status, stdout.getvalue() + + +def _verdict(exit_status, stdout): + return classify_access_probe( + exit_status=exit_status, stdout=stdout, stderr="", bot_username="fortune_bot" + ) + + +def test_a_stable_welcome_passes_once_the_quiet_window_ends(monkeypatch): + bot = _Bot(lambda _now: [_answer(11, "Welcome!")]) + + exit_status, stdout = _run(monkeypatch, bot) + + assert (exit_status, stdout.strip()) == (0, PROBE_PASSED_MARKER) + assert _verdict(exit_status, stdout) is None + assert bot.sent == ["/start"] + assert bot.reads == [0, 1, 2, 3] + assert bot.disconnected + + +def test_successive_answers_reset_the_quiet_window(monkeypatch): + bot = _Bot(lambda now: [_answer(11, "Welcome!"), _answer(12, "Menu")][: 1 + (now >= 2)]) + + exit_status, _stdout = _run(monkeypatch, bot) + + assert exit_status == 0 + assert bot.reads == [0, 1, 2, 3, 4, 5] + + +def test_silence_runs_to_the_deadline_and_is_not_a_denial(monkeypatch): + bot = _Bot(lambda _now: []) + + exit_status, stdout = _run(monkeypatch, bot) + + assert _verdict(exit_status, stdout) is None + assert bot.reads == list(range(11)) + assert bot.disconnected + + +def test_a_denial_after_a_first_answer_blocks_before_the_answers_settle(monkeypatch): + bot = _Bot( + lambda now: [_answer(11, "Hello"), _answer(12, "Access denied for you")][: 1 + (now >= 2)] + ) + + exit_status, stdout = _run(monkeypatch, bot) + + blocker = _verdict(exit_status, stdout) + assert exit_status == 2 + assert blocker.category == QABlockerCategory.TELEGRAM_ACCESS_DENIED + assert blocker.received == "Access denied for you" + assert bot.reads == [0, 1, 2] + assert bot.disconnected + + +def test_another_account_is_refused_before_anything_is_sent(monkeypatch): + bot = _Bot(lambda _now: [], account_id=QA_TEST_TELEGRAM_ID + 1) + + exit_status, stdout = _run(monkeypatch, bot) + + blocker = _verdict(exit_status, stdout) + assert exit_status == 3 + assert blocker.category == QABlockerCategory.UNKNOWN + assert blocker.received == f"expected={QA_TEST_TELEGRAM_ID};actual={QA_TEST_TELEGRAM_ID + 1}" + assert bot.sent == [] + assert bot.disconnected diff --git a/shared/tests/unit/test_telegram_bot_probe.py b/shared/tests/unit/test_telegram_bot_probe.py index fee50d166..c628361e8 100644 --- a/shared/tests/unit/test_telegram_bot_probe.py +++ b/shared/tests/unit/test_telegram_bot_probe.py @@ -7,7 +7,9 @@ import json import sys import time -from types import ModuleType +from types import ModuleType, SimpleNamespace + +import pytest from shared.telegram_bot_probe import ( PROBE_RESULT_MARKER, @@ -214,12 +216,7 @@ def disconnect(self): return None _install_telethon(monkeypatch, FakeClient) - monkeypatch.setenv("TELETHON_SESSION", "session") - monkeypatch.setenv("TELETHON_API_ID", "123") - monkeypatch.setenv("TELETHON_API_HASH", "hash") - ticks = iter((0, 0, 2)) - monkeypatch.setattr(time, "monotonic", lambda: next(ticks)) - monkeypatch.setattr(time, "sleep", lambda _seconds: None) + _virtual_clock(monkeypatch) stdout = StringIO() with redirect_stdout(stdout): @@ -347,12 +344,7 @@ def disconnect(self): return None _install_telethon(monkeypatch, FakeClient, callback_request=Request) - monkeypatch.setenv("TELETHON_SESSION", "session") - monkeypatch.setenv("TELETHON_API_ID", "123") - monkeypatch.setenv("TELETHON_API_HASH", "hash") - ticks = iter((0, 0, 2)) - monkeypatch.setattr(time, "monotonic", lambda: next(ticks)) - monkeypatch.setattr(time, "sleep", lambda _seconds: None) + _virtual_clock(monkeypatch) stdout = StringIO() with redirect_stdout(stdout): @@ -376,223 +368,68 @@ def disconnect(self): assert evidence["post_press_message"]["text"] == "Details opened" -def test_generated_callback_collects_an_interim_and_later_answer(monkeypatch): - class KeyboardButtonCallback: - def __init__(self, text, data): - self.text = text - self.data = data - - class Row: - def __init__(self, *buttons): - self.buttons = buttons - - class ReplyInlineMarkup: - def __init__(self, *rows): - self.rows = rows +class MessageMediaPhoto: + pass - class Request: - def __init__(self, **kwargs): - self.kwargs = kwargs - - def bot_message(identifier, text, *, markup=None): - return type( - "Message", - (), - { - "id": identifier, - "out": False, - "raw_text": text, - "message": text, - "media": None, - "reply_markup": markup, - }, - )() - - original = bot_message( - 7, - "Choose", - markup=ReplyInlineMarkup(Row(KeyboardButtonCallback("Career", b"career"))), - ) - interim = bot_message(10, "Working on it") - answer = bot_message(11, "Your career reading") - - class FakeClient: - def __init__(self, *_args): - self.callback_sent = False - self.reply_reads = 0 - - def start(self): - return None - - def get_entity(self, _username): - return "fortune-bot" - def get_messages(self, _bot, *, ids=None, min_id=None, limit=None): - if ids == 7: - return original - if limit == 1: - return [original] - assert min_id == 7 - self.reply_reads += 1 - replies = ([interim], [answer, interim], [answer, interim], [answer, interim]) - return replies[self.reply_reads - 1] +class KeyboardButtonCallback: + def __init__(self, text, data): + self.text = text + self.data = data - def __call__(self, _request): - self.callback_sent = True - return type("Answer", (), {"message": None, "alert": False, "url": None})() - def disconnect(self): - return None +class Row: + def __init__(self, *buttons): + self.buttons = buttons - _install_telethon(monkeypatch, FakeClient, callback_request=Request) - monkeypatch.setenv("TELETHON_SESSION", "session") - monkeypatch.setenv("TELETHON_API_ID", "123") - monkeypatch.setenv("TELETHON_API_HASH", "hash") - ticks = iter((0, 0, 1, 3, 10)) - monkeypatch.setattr(time, "monotonic", lambda: next(ticks)) - monkeypatch.setattr(time, "sleep", lambda _seconds: None) - stdout = StringIO() - with redirect_stdout(stdout): - exec( # noqa: S102 - generated probe source - build_bot_callback_script( - "fortune_bot", - 7, - "Y2FyZWVy", - button_text="Career", - wait_seconds=10, - ), - {"__name__": "telegram_callback_script"}, - ) +class ReplyInlineMarkup: + def __init__(self, *rows): + self.rows = rows - evidence = parse_bot_probe_result(stdout.getvalue()) - assert evidence["error"] is None - assert [reply["text"] for reply in evidence["replies"]] == [ - "Working on it", - "Your career reading", - ] +class CallbackRequest: + def __init__(self, **kwargs): + self.kwargs = kwargs -def test_generated_callback_keeps_polling_until_an_interim_edit_is_final(monkeypatch): - class KeyboardButtonCallback: - def __init__(self, text, data): - self.text = text - self.data = data +class _Clock: + """Virtual monotonic time that the generated script reads and advances by sleeping.""" - class Row: - def __init__(self, *buttons): - self.buttons = buttons + def __init__(self): + self.now = 0 + self.sleeps: list[float] = [] - class ReplyInlineMarkup: - def __init__(self, *rows): - self.rows = rows + def sleep(self, seconds): + assert seconds > 0 + self.sleeps.append(seconds) + self.now += seconds - class Request: - def __init__(self, **kwargs): - self.kwargs = kwargs - def bot_message(text, *, markup=None): - return type( - "Message", - (), - { - "id": 7, - "out": False, - "raw_text": text, - "message": text, - "media": None, - "reply_markup": markup, - }, - )() - - original = bot_message( - "Choose", - markup=ReplyInlineMarkup(Row(KeyboardButtonCallback("Career", b"career"))), - ) - interim = bot_message("Working on it") - answer = bot_message("Your career reading") - - class FakeClient: - def __init__(self, *_args): - self.callback_sent = False - self.post_press_reads = 0 - - def start(self): - return None - - def get_entity(self, _username): - return "fortune-bot" - - def get_messages(self, _bot, *, ids=None, min_id=None, limit=None): - if ids == 7: - if not self.callback_sent: - return original - self.post_press_reads += 1 - return (interim, answer, answer, answer)[self.post_press_reads - 1] - if limit == 1: - return [original] - assert min_id == 7 - return [] - - def __call__(self, _request): - self.callback_sent = True - return type("Answer", (), {"message": None, "alert": False, "url": None})() - - def disconnect(self): - return None - - _install_telethon(monkeypatch, FakeClient, callback_request=Request) +def _virtual_clock(monkeypatch) -> _Clock: + clock = _Clock() + monkeypatch.setattr(time, "monotonic", lambda: clock.now) + monkeypatch.setattr(time, "sleep", clock.sleep) monkeypatch.setenv("TELETHON_SESSION", "session") monkeypatch.setenv("TELETHON_API_ID", "123") monkeypatch.setenv("TELETHON_API_HASH", "hash") - ticks = iter((0, 0, 1, 3, 10)) - monkeypatch.setattr(time, "monotonic", lambda: next(ticks)) - monkeypatch.setattr(time, "sleep", lambda _seconds: None) - - stdout = StringIO() - with redirect_stdout(stdout): - exec( # noqa: S102 - generated probe source - build_bot_callback_script( - "fortune_bot", - 7, - "Y2FyZWVy", - button_text="Career", - wait_seconds=10, - ), - {"__name__": "telegram_callback_script"}, - ) + return clock - evidence = parse_bot_probe_result(stdout.getvalue()) - assert evidence["error"] is None - assert evidence["post_press_message"]["text"] == "Your career reading" +def _bot_message(identifier, text, *, media=None, markup=None): + return SimpleNamespace( + id=identifier, out=False, raw_text=text, message=text, media=media, reply_markup=markup + ) -def test_generated_message_collects_a_later_photo_reply(monkeypatch): - class MessageMediaPhoto: - pass +def _message_bot(clock, chat): + """A bot whose replies after the probe's message 10 are ``chat(now)``, oldest first.""" - def bot_message(identifier, text, *, media=None): - return type( - "Message", - (), - { - "id": identifier, - "out": False, - "raw_text": text, - "message": text, - "media": media, - "reply_markup": None, - }, - )() + class Client: + reads: list[float] = [] - interim = bot_message(11, "Working on it") - photo = bot_message(12, "The cards", media=MessageMediaPhoto()) - - class FakeClient: def __init__(self, *_args): - self.reply_reads = 0 + pass def start(self): return None @@ -601,60 +438,44 @@ def get_entity(self, _username): return "fortune-bot" def send_message(self, _bot, _message): - return type("Sent", (), {"id": 10})() + return SimpleNamespace(id=10) def get_messages(self, _bot, *, min_id, limit): - assert (min_id, limit) == (10, 10) - self.reply_reads += 1 - replies = ([interim], [photo, interim], [photo, interim], [photo, interim]) - return replies[self.reply_reads - 1] + assert min_id == 10 + Client.reads.append(clock.now) + return list(reversed(chat(clock.now)))[:limit] def disconnect(self): return None - _install_telethon(monkeypatch, FakeClient) - monkeypatch.setenv("TELETHON_SESSION", "session") - monkeypatch.setenv("TELETHON_API_ID", "123") - monkeypatch.setenv("TELETHON_API_HASH", "hash") - ticks = iter((0, 0, 1, 3, 10)) - monkeypatch.setattr(time, "monotonic", lambda: next(ticks)) - monkeypatch.setattr(time, "sleep", lambda _seconds: None) + return Client + +def _probe(monkeypatch, client, *, wait_seconds) -> dict: + _install_telethon(monkeypatch, client) stdout = StringIO() with redirect_stdout(stdout): exec( # noqa: S102 - generated probe source - build_bot_message_script("fortune_bot", "/reading", wait_seconds=10), + build_bot_message_script("fortune_bot", "/reading", wait_seconds=wait_seconds), {"__name__": "telegram_probe_script"}, ) + return parse_bot_probe_result(stdout.getvalue()) - evidence = parse_bot_probe_result(stdout.getvalue()) - - assert evidence["error"] is None - assert [reply["id"] for reply in evidence["replies"]] == [11, 12] - assert evidence["replies"][1]["caption"] == "The cards" - assert evidence["replies"][1]["media_type"] == "MessageMediaPhoto" +def _callback_bot(clock, *, pressed, chat, answer=None): + """Message 7 offers one button; after the press it reads ``pressed(now)``. -def test_generated_message_keeps_polling_to_the_deadline_after_one_reply(monkeypatch): - reply = type( - "Message", - (), - { - "id": 11, - "out": False, - "raw_text": "One answer", - "message": "One answer", - "media": None, - "reply_markup": None, - }, - )() + The newest message before the press is 9; ``chat(now)`` is what follows it. + """ + original = _bot_message( + 7, "Choose", markup=ReplyInlineMarkup(Row(KeyboardButtonCallback("Career", b"career"))) + ) - class FakeClient: - instance = None + class Client: + reads: list[float] = [] def __init__(self, *_args): - self.reply_reads = 0 - FakeClient.instance = self + self.pressed = False def start(self): return None @@ -662,285 +483,202 @@ def start(self): def get_entity(self, _username): return "fortune-bot" - def send_message(self, _bot, _message): - return type("Sent", (), {"id": 10})() + def get_messages(self, _bot, *, ids=None, min_id=None, limit=None): + if ids == 7: + return pressed(clock.now) if self.pressed else original + if limit == 1: + return [SimpleNamespace(id=9)] + assert min_id == 9 + Client.reads.append(clock.now) + return list(reversed(chat(clock.now))) - def get_messages(self, _bot, *, min_id, limit): - assert (min_id, limit) == (10, 10) - self.reply_reads += 1 - return [reply] + def __call__(self, request): + assert request.kwargs == {"peer": "fortune-bot", "msg_id": 7, "data": b"career"} + self.pressed = True + return SimpleNamespace(message=answer, alert=False, url=None) def disconnect(self): return None - _install_telethon(monkeypatch, FakeClient) - monkeypatch.setenv("TELETHON_SESSION", "session") - monkeypatch.setenv("TELETHON_API_ID", "123") - monkeypatch.setenv("TELETHON_API_HASH", "hash") - ticks = iter((0, 0, 1, 3, 10)) - monkeypatch.setattr(time, "monotonic", lambda: next(ticks)) - sleeps: list[int] = [] - monkeypatch.setattr(time, "sleep", sleeps.append) + return Client, original + +def _press(monkeypatch, client, *, wait_seconds) -> dict: + _install_telethon(monkeypatch, client, callback_request=CallbackRequest) stdout = StringIO() with redirect_stdout(stdout): exec( # noqa: S102 - generated probe source - build_bot_message_script("fortune_bot", "/reading", wait_seconds=10), - {"__name__": "telegram_probe_script"}, + build_bot_callback_script( + "fortune_bot", 7, "Y2FyZWVy", button_text="Career", wait_seconds=wait_seconds + ), + {"__name__": "telegram_callback_script"}, ) + return parse_bot_probe_result(stdout.getvalue()) - evidence = parse_bot_probe_result(stdout.getvalue()) - assert evidence["replies"][0]["text"] == "One answer" - assert FakeClient.instance.reply_reads == 4 - assert sleeps == [2, 2, 2] +def test_generated_message_returns_once_one_reply_stays_unchanged_for_the_quiet_window( + monkeypatch, +): + clock = _virtual_clock(monkeypatch) + client = _message_bot(clock, lambda _now: [_bot_message(11, "One answer")]) + evidence = _probe(monkeypatch, client, wait_seconds=15) -def test_generated_message_collects_a_reply_well_after_the_old_quiet_window(monkeypatch): - def bot_message(identifier, text): - return type( - "Message", - (), - { - "id": identifier, - "out": False, - "raw_text": text, - "message": text, - "media": None, - "reply_markup": None, - }, - )() + assert [reply["text"] for reply in evidence["replies"]] == ["One answer"] + assert client.reads == [0, 2, 3] + assert clock.sleeps == [2, 1] - interim = bot_message(11, "Working on it") - answer = bot_message(12, "Your career reading") - class FakeClient: - def __init__(self, *_args): - self.reply_reads = 0 +def test_generated_message_silence_runs_to_the_deadline_and_reads_it(monkeypatch): + clock = _virtual_clock(monkeypatch) + client = _message_bot(clock, lambda _now: []) - def start(self): - return None + evidence = _probe(monkeypatch, client, wait_seconds=15) - def get_entity(self, _username): - return "fortune-bot" + assert evidence["delivered"] is True + assert evidence["replies"] == [] + assert client.reads == [0, 2, 4, 6, 8, 10, 12, 14, 15] + assert clock.now == 15 - def send_message(self, _bot, _message): - return type("Sent", (), {"id": 10})() - def get_messages(self, _bot, *, min_id, limit): - assert (min_id, limit) == (10, 10) - self.reply_reads += 1 - replies = ([interim], [interim], [interim], [answer, interim], [answer, interim]) - return replies[self.reply_reads - 1] +def test_generated_message_successive_replies_reset_the_quiet_window(monkeypatch): + clock = _virtual_clock(monkeypatch) + interim = _bot_message(11, "Working on it") + photo = _bot_message(12, "The cards", media=MessageMediaPhoto()) + answer = _bot_message(13, "Your reading") + client = _message_bot( + clock, lambda now: [interim, photo, answer][: 1 + (now >= 2) + (now >= 4)] + ) - def disconnect(self): - return None + evidence = _probe(monkeypatch, client, wait_seconds=60) - _install_telethon(monkeypatch, FakeClient) - monkeypatch.setenv("TELETHON_SESSION", "session") - monkeypatch.setenv("TELETHON_API_ID", "123") - monkeypatch.setenv("TELETHON_API_HASH", "hash") - ticks = iter((0, 0, 1, 3, 10, 15)) - monkeypatch.setattr(time, "monotonic", lambda: next(ticks)) - monkeypatch.setattr(time, "sleep", lambda _seconds: None) + assert [reply["id"] for reply in evidence["replies"]] == [11, 12, 13] + assert evidence["replies"][1]["caption"] == "The cards" + assert evidence["replies"][1]["media_type"] == "MessageMediaPhoto" + assert client.reads == [0, 2, 4, 6, 7] - stdout = StringIO() - with redirect_stdout(stdout): - exec( # noqa: S102 - generated probe source - build_bot_message_script("fortune_bot", "/reading", wait_seconds=15), - {"__name__": "telegram_probe_script"}, - ) - evidence = parse_bot_probe_result(stdout.getvalue()) +def test_generated_message_edits_of_text_media_and_markup_reset_the_quiet_window(monkeypatch): + clock = _virtual_clock(monkeypatch) - assert evidence["error"] is None - assert [reply["text"] for reply in evidence["replies"]] == [ - "Working on it", - "Your career reading", - ] + def chat(now): + if now < 2: + return [_bot_message(11, "Loading")] + markup = ReplyInlineMarkup( + Row(KeyboardButtonCallback("Again" if now >= 6 else "More", b"more")) + ) + if now < 4: + return [_bot_message(11, "The cards", media=MessageMediaPhoto())] + return [_bot_message(11, "The cards", media=MessageMediaPhoto(), markup=markup)] + client = _message_bot(clock, chat) -def test_generated_message_honors_an_extended_deadline_without_early_exit(monkeypatch): - def bot_message(identifier, text): - return type( - "Message", - (), - { - "id": identifier, - "out": False, - "raw_text": text, - "message": text, - "media": None, - "reply_markup": None, - }, - )() + evidence = _probe(monkeypatch, client, wait_seconds=60) - interim = bot_message(11, "Working on it") - answer = bot_message(12, "Your career reading") + (reply,) = evidence["replies"] + assert reply["caption"] == "The cards" + assert reply["reply_markup"]["buttons"][0]["text"] == "Again" + assert client.reads == [0, 2, 4, 6, 8, 9] - class FakeClient: - def __init__(self, *_args): - self.reply_reads = 0 - def start(self): - return None +def test_generated_message_does_not_collect_a_reply_after_a_completed_quiet_window( + monkeypatch, +): + clock = _virtual_clock(monkeypatch) + interim = _bot_message(11, "Working on it") + answer = _bot_message(12, "Your reading") + client = _message_bot(clock, lambda now: [interim, answer] if now >= 10 else [interim]) - def get_entity(self, _username): - return "fortune-bot" + evidence = _probe(monkeypatch, client, wait_seconds=15) - def send_message(self, _bot, _message): - return type("Sent", (), {"id": 10})() + assert [reply["text"] for reply in evidence["replies"]] == ["Working on it"] + assert clock.now == 3 - def get_messages(self, _bot, *, min_id, limit): - assert (min_id, limit) == (10, 10) - self.reply_reads += 1 - return [answer, interim] if self.reply_reads == 4 else [interim] - def disconnect(self): - return None +def test_generated_message_short_wait_ends_before_the_quiet_window(monkeypatch): + clock = _virtual_clock(monkeypatch) + client = _message_bot(clock, lambda _now: [_bot_message(11, "One answer")]) - _install_telethon(monkeypatch, FakeClient) - monkeypatch.setenv("TELETHON_SESSION", "session") - monkeypatch.setenv("TELETHON_API_ID", "123") - monkeypatch.setenv("TELETHON_API_HASH", "hash") - ticks = iter((0, 0, 14, 15, 20)) - monkeypatch.setattr(time, "monotonic", lambda: next(ticks)) - monkeypatch.setattr(time, "sleep", lambda _seconds: None) + evidence = _probe(monkeypatch, client, wait_seconds=1) - stdout = StringIO() - with redirect_stdout(stdout): - exec( # noqa: S102 - generated probe source - build_bot_message_script("fortune_bot", "/reading", wait_seconds=20), - {"__name__": "telegram_probe_script"}, - ) + assert [reply["id"] for reply in evidence["replies"]] == [11] + assert client.reads == [0, 1] + assert clock.sleeps == [1] - evidence = parse_bot_probe_result(stdout.getvalue()) - - assert evidence["error"] is None - assert [reply["text"] for reply in evidence["replies"]] == [ - "Working on it", - "Your career reading", - ] +def test_generated_message_replies_changing_through_the_deadline_are_read_at_it(monkeypatch): + clock = _virtual_clock(monkeypatch) + client = _message_bot( + clock, + lambda now: [_bot_message(identifier, "tick") for identifier in range(11, 12 + int(now))], + ) -def test_generated_message_reads_the_final_state_at_the_deadline(monkeypatch): - def bot_message(identifier, text): - return type( - "Message", - (), - { - "id": identifier, - "out": False, - "raw_text": text, - "message": text, - "media": None, - "reply_markup": None, - }, - )() + evidence = _probe(monkeypatch, client, wait_seconds=5) - interim = bot_message(11, "Working on it") - answer = bot_message(12, "Your career reading") + assert [reply["id"] for reply in evidence["replies"]] == list(range(11, 17)) + assert client.reads == [0, 2, 4, 5] - class FakeClient: - def __init__(self, *_args): - self.reply_reads = 0 - def start(self): - return None - - def get_entity(self, _username): - return "fortune-bot" +def test_generated_message_keeps_the_newest_replies_at_the_evidence_bound(monkeypatch): + clock = _virtual_clock(monkeypatch) + replies = [_bot_message(identifier, f"Reply {identifier}") for identifier in range(11, 23)] + client = _message_bot(clock, lambda _now: replies) - def send_message(self, _bot, _message): - return type("Sent", (), {"id": 10})() + evidence = _probe(monkeypatch, client, wait_seconds=10) - def get_messages(self, _bot, *, min_id, limit): - assert (min_id, limit) == (10, 10) - self.reply_reads += 1 - return [answer, interim] if self.reply_reads == 5 else [interim] - - def disconnect(self): - return None + assert [reply["id"] for reply in evidence["replies"]] == list(range(13, 23)) - _install_telethon(monkeypatch, FakeClient) - monkeypatch.setenv("TELETHON_SESSION", "session") - monkeypatch.setenv("TELETHON_API_ID", "123") - monkeypatch.setenv("TELETHON_API_HASH", "hash") - ticks = iter((0, 0, 1, 3, 14, 15)) - monkeypatch.setattr(time, "monotonic", lambda: next(ticks)) - monkeypatch.setattr(time, "sleep", lambda _seconds: None) - stdout = StringIO() - with redirect_stdout(stdout): - exec( # noqa: S102 - generated probe source - build_bot_message_script("fortune_bot", "/reading", wait_seconds=15), - {"__name__": "telegram_probe_script"}, - ) +def test_generated_callback_replies_after_the_press_settle_before_the_deadline(monkeypatch): + clock = _virtual_clock(monkeypatch) + interim = _bot_message(10, "Working on it") + answer = _bot_message(11, "Your career reading") + client, original = _callback_bot( + clock, + pressed=lambda _now: original, + chat=lambda now: [interim, answer] if now >= 2 else [interim], + ) - evidence = parse_bot_probe_result(stdout.getvalue()) + evidence = _press(monkeypatch, client, wait_seconds=10) + assert evidence["error"] is None assert [reply["text"] for reply in evidence["replies"]] == [ "Working on it", "Your career reading", ] + assert client.reads == [0, 2, 4, 5] -def test_generated_message_keeps_the_newest_replies_at_the_evidence_bound(monkeypatch): - replies = [ - type( - "Message", - (), - { - "id": identifier, - "out": False, - "raw_text": f"Reply {identifier}", - "message": f"Reply {identifier}", - "media": None, - "reply_markup": None, - }, - )() - for identifier in range(11, 23) - ] - - class FakeClient: - def __init__(self, *_args): - pass - - def start(self): - return None - - def get_entity(self, _username): - return "fortune-bot" - - def send_message(self, _bot, _message): - return type("Sent", (), {"id": 10})() +def test_generated_callback_edits_of_the_pressed_message_reset_the_quiet_window(monkeypatch): + clock = _virtual_clock(monkeypatch) + client, _original = _callback_bot( + clock, + pressed=lambda now: _bot_message(7, "Your career reading" if now >= 2 else "Working on it"), + chat=lambda _now: [], + ) - def get_messages(self, _bot, *, min_id, limit): - assert min_id == 10 - # Telethon returns newest-first and applies the requested limit. - return list(reversed(replies))[:limit] + evidence = _press(monkeypatch, client, wait_seconds=10) - def disconnect(self): - return None + assert evidence["error"] is None + assert evidence["replies"] == [] + assert evidence["post_press_message"]["text"] == "Your career reading" + assert client.reads == [0, 2, 4, 5] - _install_telethon(monkeypatch, FakeClient) - monkeypatch.setenv("TELETHON_SESSION", "session") - monkeypatch.setenv("TELETHON_API_ID", "123") - monkeypatch.setenv("TELETHON_API_HASH", "hash") - ticks = iter((0, 0, 2, 10)) - monkeypatch.setattr(time, "monotonic", lambda: next(ticks)) - monkeypatch.setattr(time, "sleep", lambda _seconds: None) - stdout = StringIO() - with redirect_stdout(stdout): - exec( # noqa: S102 - generated probe source - build_bot_message_script("fortune_bot", "/reading", wait_seconds=10), - {"__name__": "telegram_probe_script"}, - ) +@pytest.mark.parametrize("answer", [None, "Loading your reading"]) +def test_generated_callback_acknowledgement_alone_runs_to_the_deadline(monkeypatch, answer): + """An unchanged pressed message and a callback answer are not a bot response.""" + clock = _virtual_clock(monkeypatch) + client, original = _callback_bot( + clock, pressed=lambda _now: original, chat=lambda _now: [], answer=answer + ) - evidence = parse_bot_probe_result(stdout.getvalue()) + evidence = _press(monkeypatch, client, wait_seconds=10) - assert [reply["id"] for reply in evidence["replies"]] == list(range(13, 23)) + assert evidence["delivered"] is True + assert evidence["callback"] == {"text": answer, "alert": False, "url": None} + assert evidence["post_press_message"]["text"] == "Choose" + assert client.reads == [0, 2, 4, 6, 8, 10] def _install_telethon(monkeypatch, client, *, callback_request=None): diff --git a/shared/tests/unit/test_telegram_probe_injection.py b/shared/tests/unit/test_telegram_probe_injection.py index b8d8bec84..8a9aefdef 100644 --- a/shared/tests/unit/test_telegram_probe_injection.py +++ b/shared/tests/unit/test_telegram_probe_injection.py @@ -62,7 +62,7 @@ def __init__(self, value): import base64, os, time from telethon import record -# Advance the child's clock to the validated one-second deadline before its +# Put the child's clock at the validated one-second deadline right after its # first collection read. The security checks need one poll, not real sleep. _ticks = iter((0, 1)) time.monotonic = lambda: next(_ticks) From 071c05ad0159866339f4b18005319c98e4e8fcb3 Mon Sep 17 00:00:00 2001 From: q Date: Sun, 18 Oct 2026 08:51:38 +0000 Subject: [PATCH 2/2] Keep an observed probe response qualifying after the pressed message reverts --- shared/telegram_bot_probe.py | 10 ++++++---- shared/tests/unit/test_telegram_bot_probe.py | 18 ++++++++++++++++++ 2 files changed, 24 insertions(+), 4 deletions(-) diff --git a/shared/telegram_bot_probe.py b/shared/telegram_bot_probe.py index 8011fd036..ca37cfb01 100644 --- a/shared/telegram_bot_probe.py +++ b/shared/telegram_bot_probe.py @@ -101,24 +101,26 @@ def newest_message_id(client, bot): def settling_collector_source() -> str: """Script source of ``collect_until_settled(observe, responded, wait, poll)``. - It polls ``observe()`` until ``responded`` holds and the observation has stayed - equal for the quiet window, or until ``wait`` seconds pass, and always returns - right after an observation. + It polls ``observe()`` until ``responded`` has held for any observation and the + observation has stayed equal for the quiet window, or until ``wait`` seconds + pass, and always returns right after an observation. """ return f"""\ def collect_until_settled(observe, responded, wait, poll): deadline = time.monotonic() + wait last = None + has_responded = False while True: observed = observe() now = time.monotonic() if observed != last: last = observed changed_at = now + has_responded = has_responded or responded(observed) if now >= deadline: return wake = min(now + poll, deadline) - if responded(observed): + if has_responded: settled_at = changed_at + {TELEGRAM_QUIET_WINDOW} if now >= settled_at: return diff --git a/shared/tests/unit/test_telegram_bot_probe.py b/shared/tests/unit/test_telegram_bot_probe.py index c628361e8..53b87dd83 100644 --- a/shared/tests/unit/test_telegram_bot_probe.py +++ b/shared/tests/unit/test_telegram_bot_probe.py @@ -665,6 +665,24 @@ def test_generated_callback_edits_of_the_pressed_message_reset_the_quiet_window( assert client.reads == [0, 2, 4, 5] +def test_generated_callback_settles_after_an_edit_restores_the_original_message(monkeypatch): + """A response once observed still qualifies after the pressed message reverts.""" + clock = _virtual_clock(monkeypatch) + client, original = _callback_bot( + clock, + pressed=lambda now: original if now >= 2 else _bot_message(7, "Loading"), + chat=lambda _now: [], + ) + + evidence = _press(monkeypatch, client, wait_seconds=60) + + assert evidence["error"] is None + assert evidence["post_press_message"]["text"] == "Choose" + assert evidence["post_press_message"]["reply_markup"]["buttons"][0]["text"] == "Career" + assert client.reads == [0, 2, 4, 5] + assert clock.now == 5 + + @pytest.mark.parametrize("answer", [None, "Loading your reading"]) def test_generated_callback_acknowledgement_alone_runs_to_the_deadline(monkeypatch, answer): """An unchanged pressed message and a callback answer are not a bot response."""