diff --git a/CHANGELOG.md b/CHANGELOG.md index effc2811..0bab8d3d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/src/aipass/skills/lib/telegram/apps/handlers/base_bot.py b/src/aipass/skills/lib/telegram/apps/handlers/base_bot.py index db0fe9ba..6dc8c45a 100644 --- a/src/aipass/skills/lib/telegram/apps/handlers/base_bot.py +++ b/src/aipass/skills/lib/telegram/apps/handlers/base_bot.py @@ -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: diff --git a/src/aipass/skills/lib/telegram/tests/test_heartbeat_delivered.py b/src/aipass/skills/lib/telegram/tests/test_heartbeat_delivered.py index b0b7c9a5..bf164eaf 100644 --- a/src/aipass/skills/lib/telegram/tests/test_heartbeat_delivered.py +++ b/src/aipass/skills/lib/telegram/tests/test_heartbeat_delivered.py @@ -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() diff --git a/src/aipass/skills/lib/telegram/tests/test_streaming.py b/src/aipass/skills/lib/telegram/tests/test_streaming.py index 39a10c37..1070bf4a 100644 --- a/src/aipass/skills/lib/telegram/tests/test_streaming.py +++ b/src/aipass/skills/lib/telegram/tests/test_streaming.py @@ -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()