fix: TG bot heartbeat race — generation counter kills resurrected heartbeat threads (shared stop Event cleared by next start let a >5s-stuck thread overwrite delivered replies with Processing...), delivered re-check before every edit in batch+streaming loops, superseded pending placeholders finalized in message+file paths (rapid-fire single-slot strand). Root-caused live from Patrick's frozen bubble; 6 new heartbeat tests, full TG suite 797 green devpulse-verified.

2026-07-14 11:00:48 -07:00
parent 62d047cac1
commit af12158cbc
4 changed files with 195 additions and 30 deletions
@@ -33,6 +33,23 @@ PyPI version — not the changelog header.

 ### Fixed

+- **TG bot heartbeat race: delivered replies no longer flip back to
+  "Processing…".** Patrick watched his answered bubble get overwritten live: a
+  heartbeat thread stuck >5s in a slow Telegram edit call survived its stop
+  (the join timed out), woke to a *shared* stop Event the next message had
+  already cleared, and re-edited the old placeholder with "Processing…
+  (elapsed)" over the delivered reply. Fixed structurally (@skills, devpulse
+  root-cause brief): a generation counter captured per heartbeat thread —
+  any stale thread breaks before every edit — plus a delivered re-check
+  immediately before each edit call in both batch and streaming loops.
+  Second bug in the same window: rapid-fire messages (photo + text in one
+  turn) overwrite the bot's single pending slot, stranding the earlier
+  placeholder frozen; superseded placeholders are now finalized to
+  "⏭ Superseded by newer message" in both message and file paths. 6 new
+  heartbeat tests; full TG suite 797 green (devpulse-verified). Deployment
+  lesson from the same morning: bot fixes aren't live until the systemd
+  units restart — commit ≠ deploy.
+
 - **TG mirror live-test fixes: main-chat messages mirror, TG messages don't
  echo.** Patrick's first morning test caught what 47 green tests missed: the
  relay's sub-agent skip blocked ALL daemon-backed main chats (they run with
@@ -229,6 +229,7 @@ class BaseBot:
        self._rate_limit_tracker: dict[int, list] = {}
        self._heartbeat_thread: threading.Thread | None = None
        self._heartbeat_stop = threading.Event()
+        self._heartbeat_gen: int = 0

        # Conversation state for /create flow (keyed by chat_id)
        self._create_state: dict[int, dict] = {}
@@ -696,20 +697,8 @@ class BaseBot:
            )
            return

-        # Inbound reliability: clean stale pending + warn on in-flight overwrite
-        self.clean_stale_pending()
-        if self.pending_file.exists():
-            try:
-                prev = json.loads(self.pending_file.read_text(encoding="utf-8"))
-                if not prev.get("delivered"):
-                    prev_id = prev.get("message_id", "?")
-                    logger.warning(
-                        "Overwriting undelivered pending (msg_id=%s) with new message %d",
-                        prev_id,
-                        message_id,
-                    )
-            except (json.JSONDecodeError, OSError):
-                pass
+        # Inbound reliability: clean stale pending + finalize stranded placeholder
+        self._finalize_superseded_pending(message_id)

        # Send processing indicator
        processing_result = self.send_message(chat_id, PROCESSING_MSG)
@@ -809,6 +798,9 @@ class BaseBot:
                file_path.unlink(missing_ok=True)
            return

+        # Inbound reliability: clean stale pending + finalize stranded placeholder
+        self._finalize_superseded_pending(message_id)
+
        # Send processing indicator
        processing_result = self.send_message(chat_id, f"Processing {file_type} file...")
        processing_msg_id = processing_result.get("message_id") if processing_result else None
@@ -1503,6 +1495,31 @@ class BaseBot:
        except OSError as e:
            logger.warning("Failed to clean stale pending file: %s", e)

+    def _finalize_superseded_pending(self, new_message_id: int) -> None:
+        """Clean stale pending and finalize stranded placeholder on overwrite."""
+        self.clean_stale_pending()
+        if not self.pending_file.exists():
+            return
+        try:
+            prev = json.loads(self.pending_file.read_text(encoding="utf-8"))
+            if not prev.get("delivered"):
+                prev_id = prev.get("message_id", "?")
+                logger.warning(
+                    "Overwriting undelivered pending (msg_id=%s) with new message %d",
+                    prev_id,
+                    new_message_id,
+                )
+                prev_proc_id = prev.get("processing_message_id")
+                prev_chat_id = prev.get("chat_id")
+                if prev_proc_id and prev_chat_id:
+                    self.edit_message(
+                        prev_chat_id,
+                        prev_proc_id,
+                        "⏭ Superseded by newer message",
+                    )
+        except (json.JSONDecodeError, OSError):
+            pass
+
    def _resolve_active_transcript(self) -> tuple[str | None, int]:
        """Identify the ACTIVE Claude JSONL transcript and return its path and line count.

@@ -1931,13 +1948,15 @@ class BaseBot:
        """
        self._stop_heartbeat()  # Ensure no stale thread
        self._heartbeat_stop.clear()
+        self._heartbeat_gen += 1
+        gen = self._heartbeat_gen

        def _heartbeat_loop():
            start = time.time()

            # Streaming mode (FPLAN-0297): live transcript tail
            if self._stream and self._active_transcript_path:
-                self._streaming_loop(chat_id, processing_msg_id, start)
+                self._streaming_loop(chat_id, processing_msg_id, start, gen)
                return

            # Batch mode (default): elapsed-time updates
@@ -1951,9 +1970,13 @@ class BaseBot:
                    break
                if not self._tmux_session_exists():
                    break
+                if self._heartbeat_gen != gen:
+                    break

                elapsed = time.time() - start
                elapsed_str = self._format_elapsed(elapsed)
+                if self._is_pending_delivered() or self._heartbeat_gen != gen:
+                    break
                self.edit_message(chat_id, processing_msg_id, f"Processing... ({elapsed_str})")

        self._heartbeat_thread = threading.Thread(target=_heartbeat_loop, daemon=True, name=f"heartbeat-{self.bot_id}")
@@ -1998,7 +2021,7 @@ class BaseBot:
    # STREAMING EDIT-IN-PLACE (FPLAN-0297)
    # =============================================

-    def _streaming_loop(self, chat_id: int, msg_id: int, start_time: float) -> None:
+    def _streaming_loop(self, chat_id: int, msg_id: int, start_time: float, gen: int) -> None:
        """Stream transcript content into the processing message via edit-in-place."""
        path = self._active_transcript_path
        try:
@@ -2016,6 +2039,8 @@ class BaseBot:
            self._heartbeat_stop.wait(STREAM_INTERVAL)
            if self._heartbeat_stop.is_set():
                break
+            if self._heartbeat_gen != gen:
+                break
            if self._is_pending_delivered():
                break
            if not self._tmux_session_exists():
@@ -2026,7 +2051,7 @@ class BaseBot:
            if new_text:
                buffer += new_text

-            if self._is_pending_delivered():
+            if self._is_pending_delivered() or self._heartbeat_gen != gen:
                break

            if not buffer:
@@ -2034,6 +2059,8 @@ class BaseBot:
                placeholder = f"Processing... ({self._format_elapsed(elapsed)})"
                now = time.time()
                if placeholder != last_sent and now >= retry_after_until:
+                    if self._is_pending_delivered() or self._heartbeat_gen != gen:
+                        break
                    ok, retry = self._stream_edit(chat_id, current_msg_id, placeholder)
                    retry_after_until = now + retry if retry > 0 else retry_after_until
                    last_sent = placeholder if ok else last_sent
@@ -2046,6 +2073,9 @@ class BaseBot:
            if now < retry_after_until:
                continue

+            if self._is_pending_delivered() or self._heartbeat_gen != gen:
+                break
+
            if len(buffer) > TELEGRAM_CHAR_LIMIT:
                break_at = buffer.rfind("\n", 0, TELEGRAM_CHAR_LIMIT)
                if break_at < TELEGRAM_CHAR_LIMIT // 2:
@@ -244,3 +244,121 @@ class TestReplyNotClobbered:

        # Heartbeat saw delivered immediately, never edited
        mock_edit.assert_not_called()
+
+
+# =============================================
+# 5. STALE THREAD CANNOT EDIT (generation counter)
+# =============================================
+
+
+class TestStaleThreadCannotEdit:
+    """A heartbeat thread from a previous generation must not edit after a new start."""
+
+    def test_stale_thread_blocked_by_gen(self, tmp_path, _patch_base_bot_deps):
+        bot = _make_bot(tmp_path, _patch_base_bot_deps)
+        bot.pending_file.parent.mkdir(parents=True, exist_ok=True)
+        bot.pending_file.write_text(json.dumps({"chat_id": 42}), encoding="utf-8")
+
+        edits_by_msg = {}
+
+        def track_edit(chat_id, msg_id, text):
+            edits_by_msg.setdefault(msg_id, []).append(text)
+
+        with (
+            patch.object(bot, "edit_message", side_effect=track_edit),
+            patch.object(bot, "_tmux_session_exists", return_value=True),
+            patch("aipass.skills.lib.telegram.apps.handlers.base_bot.HEARTBEAT_INTERVAL", 0.1),
+        ):
+            # Start heartbeat for msg 100
+            bot._start_heartbeat(42, 100)
+            time.sleep(0.3)
+            # Start new heartbeat for msg 200 (bumps gen, stops old)
+            bot._start_heartbeat(42, 200)
+            time.sleep(0.3)
+            bot._stop_heartbeat()
+
+        # msg 200 should have edits, msg 100 should have stopped
+        assert 200 in edits_by_msg
+        # After new start, no further edits to msg 100
+        edits_100_count = len(edits_by_msg.get(100, []))
+        edits_200_count = len(edits_by_msg.get(200, []))
+        assert edits_200_count >= 1
+        # Old thread may have gotten 1-2 edits before gen mismatch, but not indefinite
+        assert edits_100_count <= 3
+
+    def test_gen_increments_on_each_start(self, tmp_path, _patch_base_bot_deps):
+        bot = _make_bot(tmp_path, _patch_base_bot_deps)
+        bot.pending_file.parent.mkdir(parents=True, exist_ok=True)
+        bot.pending_file.write_text(json.dumps({"chat_id": 42, "delivered": True}), encoding="utf-8")
+
+        assert bot._heartbeat_gen == 0
+        with patch.object(bot, "edit_message"):
+            bot._start_heartbeat(42, 100)
+            assert bot._heartbeat_gen == 1
+            bot._start_heartbeat(42, 200)
+            assert bot._heartbeat_gen == 2
+            bot._stop_heartbeat()
+
+
+# =============================================
+# 6. RAPID-FIRE: stranded placeholder finalized
+# =============================================
+
+
+class TestRapidFireFinalize:
+    """When a new message overwrites undelivered pending, the old placeholder is finalized."""
+
+    def test_superseded_placeholder_edited(self, tmp_path, _patch_base_bot_deps):
+        bot = _make_bot(tmp_path, _patch_base_bot_deps)
+        bot.pending_file.parent.mkdir(parents=True, exist_ok=True)
+
+        # Simulate undelivered pending from msg 100 with processing_message_id 500
+        prev_pending = {
+            "chat_id": 42,
+            "message_id": 100,
+            "processing_message_id": 500,
+            "timestamp": time.time(),
+        }
+        bot.pending_file.write_text(json.dumps(prev_pending), encoding="utf-8")
+
+        with patch.object(bot, "edit_message") as mock_edit:
+            bot._finalize_superseded_pending(200)
+
+        mock_edit.assert_called_once_with(42, 500, "⏭ Superseded by newer message")
+
+    def test_no_finalize_when_delivered(self, tmp_path, _patch_base_bot_deps):
+        bot = _make_bot(tmp_path, _patch_base_bot_deps)
+        bot.pending_file.parent.mkdir(parents=True, exist_ok=True)
+
+        prev_pending = {
+            "chat_id": 42,
+            "message_id": 100,
+            "processing_message_id": 500,
+            "delivered": True,
+            "timestamp": time.time(),
+        }
+        bot.pending_file.write_text(json.dumps(prev_pending), encoding="utf-8")
+
+        with patch.object(bot, "edit_message") as mock_edit:
+            bot._finalize_superseded_pending(200)
+
+        mock_edit.assert_not_called()
+
+    def test_no_finalize_when_no_pending(self, tmp_path, _patch_base_bot_deps):
+        bot = _make_bot(tmp_path, _patch_base_bot_deps)
+        assert not bot.pending_file.exists()
+
+        with patch.object(bot, "edit_message") as mock_edit:
+            bot._finalize_superseded_pending(200)
+
+        mock_edit.assert_not_called()
+
+    def test_no_crash_on_corrupt_pending(self, tmp_path, _patch_base_bot_deps):
+        bot = _make_bot(tmp_path, _patch_base_bot_deps)
+        bot.pending_file.parent.mkdir(parents=True, exist_ok=True)
+        bot.pending_file.write_text("not json{{{", encoding="utf-8")
+
+        with patch.object(bot, "edit_message") as mock_edit:
+            bot._finalize_superseded_pending(200)
+
+        mock_edit.assert_not_called()
@@ -123,7 +123,7 @@ class TestFormatContentBlock:
        assert result is None

    def test_not_dict(self):
-        result = BaseBot._format_content_block("not a dict")
+        result = BaseBot._format_content_block("not a dict")  # type: ignore[arg-type]
        assert result is None


@@ -371,7 +371,7 @@ class TestBatchModeUnchanged:

        def delivered_after_one():
            call_count[0] += 1
-            return call_count[0] >= 2
+            return call_count[0] >= 3

        bot._is_pending_delivered = delivered_after_one
        bot._heartbeat_stop = threading.Event()
@@ -379,7 +379,7 @@ class TestBatchModeUnchanged:
        if bot._heartbeat_thread:
            bot._heartbeat_thread.join(timeout=35)
        bot._stream_edit.assert_not_called()
-        bot.edit_message.assert_called()
+        bot.edit_message.assert_called()  # type: ignore[union-attr]

    def test_stream_false_no_transcript_tail(self, tmp_path, _patch_deps):
        """When stream=False, _tail_transcript_bytes is never called."""
@@ -423,7 +423,7 @@ class TestStreamingLoop:

        def stop_after_three():
            call_count[0] += 1
-            return call_count[0] >= 3
+            return call_count[0] >= 4

        bot._is_pending_delivered = stop_after_three
        bot._tmux_session_exists = MagicMock(return_value=True)
@@ -439,7 +439,7 @@ class TestStreamingLoop:

        bot._heartbeat_stop = threading.Event()
        with patch.object(Path, "stat", fake_stat):
-            bot._streaming_loop(123, 456, time.time())
+            bot._streaming_loop(123, 456, time.time(), bot._heartbeat_gen)

        assert bot._stream_edit.call_count >= 1
        edit_text = bot._stream_edit.call_args[0][2]
@@ -463,7 +463,7 @@ class TestStreamingLoop:
        bot._stream_edit = MagicMock(return_value=(True, 0.0))
        bot._heartbeat_stop = threading.Event()

-        bot._streaming_loop(123, 456, time.time())
+        bot._streaming_loop(123, 456, time.time(), bot._heartbeat_gen)

        # Should show "Processing..." but not re-edit identical text
        first_call_text = bot._stream_edit.call_args_list[0][0][2] if bot._stream_edit.call_count > 0 else ""
@@ -487,14 +487,14 @@ class TestStreamingLoop:

        def stop_after_three():
            call_count[0] += 1
-            return call_count[0] >= 3
+            return call_count[0] >= 4

        bot._is_pending_delivered = stop_after_three
        bot._tmux_session_exists = MagicMock(return_value=True)
        bot._heartbeat_stop = threading.Event()

        with patch.object(Path, "stat", lambda s: type("S", (), {"st_size": 0})()):
-            bot._streaming_loop(123, 456, time.time())
+            bot._streaming_loop(123, 456, time.time(), bot._heartbeat_gen)

        # Should have called send_message for the rollover
        bot.send_message.assert_called()
@@ -522,14 +522,14 @@ class TestStreamingLoop:

        def stop_after_three():
            tick[0] += 1
-            return tick[0] >= 3
+            return tick[0] >= 4

        bot._is_pending_delivered = stop_after_three
        bot._tmux_session_exists = MagicMock(return_value=True)
        bot._heartbeat_stop = threading.Event()

        with patch.object(Path, "stat", lambda s: type("S", (), {"st_size": 0})()):
-            bot._streaming_loop(123, 456, time.time())
+            bot._streaming_loop(123, 456, time.time(), bot._heartbeat_gen)

        # Only 1 edit attempt — subsequent ticks skipped due to 429 backoff
        assert bot._stream_edit.call_count == 1
@@ -545,7 +545,7 @@ class TestStreamingLoop:

        def delivered_after_one():
            call_count[0] += 1
-            return call_count[0] >= 2
+            return call_count[0] >= 3

        bot._is_pending_delivered = delivered_after_one
        bot._heartbeat_stop = threading.Event()
@@ -554,7 +554,7 @@ class TestStreamingLoop:
            bot._heartbeat_thread.join(timeout=35)

        # Batch mode: edit_message called, _stream_edit not called
-        bot.edit_message.assert_called()
+        bot.edit_message.assert_called()  # type: ignore[union-attr]
        bot._stream_edit.assert_not_called()

    def test_mid_loop_delivery_breaks_without_edit(self, tmp_path, _patch_deps):
@@ -580,7 +580,7 @@ class TestStreamingLoop:
        bot._heartbeat_stop = threading.Event()

        with patch.object(Path, "stat", lambda s: type("S", (), {"st_size": 0})()):
-            bot._streaming_loop(123, 456, time.time())
+            bot._streaming_loop(123, 456, time.time(), bot._heartbeat_gen)

        bot._stream_edit.assert_not_called()