codegen-orchestrator-1695: Return QA Telegram probes after replies settle

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

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

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.

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"

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,15 +66,17 @@ 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"
         "from telethon.sync import TelegramClient\n"
         "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()"

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,37 @@ 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`` 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 has_responded:
+            settled_at = changed_at + {TELEGRAM_QUIET_WINDOW}
+            if now >= settled_at:
+                return
+            wake = min(wake, settled_at)
+        time.sleep(wake - now)
 """
 
 
@@ -137,16 +171,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 +253,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"

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