From 4f62dc2c482aa13837401d4e960b721a80bcc49c Mon Sep 17 00:00:00 2001 From: FrundlesTian <2929608755@qq.com> Date: Wed, 23 Sep 2026 21:23:41 +0800 Subject: [PATCH 1/2] fix: fall back when provider result is empty --- CHANGELOG.md | 1 + src/claude/sdk_integration.py | 12 ++++++++--- .../unit/test_claude/test_sdk_integration.py | 20 +++++++++++++++++++ 3 files changed, 30 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 874799ca..f1e7d883 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Fixed +- **OpenRouter empty-result fallback**: Anthropic-compatible providers that return an empty `ResultMessage.result` now fall back to the text in `AssistantMessage` instead of producing "No content to display" (#171) - **A run that stops early no longer reports success**: Claude produces no final text when the CLI kills a run at the turn limit, so the bot fell through to its "✅ Task completed" placeholder and told the user the work was done. A run that died at turn 10 mid-task and a run that finished were indistinguishable in Telegram (#172). `ResultMessage.subtype` is now read alongside the cost and session id, the placeholder claims completion only for `success`, and the reply carries a footer saying why the run ended — turn limit, cost budget, cancellation, or an unrecognised reason named by its raw subtype — with the turn count and a prompt to send another message to continue. Every place the bot renders a Claude reply carries it — typed messages, document, photo and voice input in agentic mode, the same four in classic mode, `/continue`, the Continue Session button and the quick-action buttons, whose `✅ … Complete` heading now reads `⚠️ … Stopped` when the run was cut short. Webhook-triggered and scheduled runs carry it too: nobody is watching those, so the reason a nightly job came back short is the whole of what its notification can say about it (#230) - **`ClaudeResponse.num_turns` is the turn count the CLI reports**: it was derived by counting `UserMessage` and `AssistantMessage` objects, which over-reports — every tool result arrives as another user message — so a run stopped at turn 10 could be recorded as having taken roughly twice that. It only reached the logs and the session store before; the stop-reason footer now shows it to the user, which made the approximation worth removing. `ResultMessage.num_turns` is used where the CLI supplies it, with the message count kept as the fallback for a result that carries none - **Inline code spans accept backtick runs of any length**: `markdown_to_telegram_html` matched a code span only between single backticks with no backtick inside, so ``` ``a`b`` ``` rendered as the span `a` followed by loose text. A run of N backticks now opens a span that closes on a run of N, and one space is stripped from each end when both are present, as CommonMark specifies. Pairing the runs is done by scanning rather than by a backreferencing regex: `` (`+)([^\n]*?)\1 `` re-scans the rest of the line for every opener that never closes, which is superlinear on a reply whose backtick runs are all of different lengths — 2.5 seconds on a 256KB reply, on the event loop, against under a millisecond for the scan. This is what lets the blocked-tool-call line print an argument containing backticks: `` Bash(`echo `whoami``) `` names the command that was actually refused, where dropping the backticks would have named a different one diff --git a/src/claude/sdk_integration.py b/src/claude/sdk_integration.py index c67b16f8..f4231004 100644 --- a/src/claude/sdk_integration.py +++ b/src/claude/sdk_integration.py @@ -781,9 +781,15 @@ async def _cancel_on_interrupt() -> None: previous_session_id=session_id, ) - # Use ResultMessage.result if available, fall back to message extraction - if result_content is not None: - content = str(result_content).strip() + # Some Anthropic-compatible providers (including OpenRouter) return + # an empty ResultMessage.result even though the AssistantMessage + # contains the actual reply. Only prefer a non-empty result so those + # providers still reach the message-extraction fallback (#171). + result_text = ( + str(result_content).strip() if result_content is not None else "" + ) + if result_text: + content = result_text else: content_parts = [] for msg in messages: diff --git a/tests/unit/test_claude/test_sdk_integration.py b/tests/unit/test_claude/test_sdk_integration.py index 5c81dadc..9b013431 100644 --- a/tests/unit/test_claude/test_sdk_integration.py +++ b/tests/unit/test_claude/test_sdk_integration.py @@ -218,6 +218,26 @@ async def test_execute_command_falls_back_to_messages(self, sdk_manager): assert response.content == "Extracted from messages" + @pytest.mark.parametrize("empty_result", ["", " \n"]) + async def test_execute_command_falls_back_when_result_is_empty( + self, sdk_manager, empty_result + ): + """Empty provider results must not hide AssistantMessage text (#171).""" + mock_factory = _mock_client_factory( + _make_assistant_message("OpenRouter response"), + _make_result_message(result=empty_result), + ) + + with patch( + "src.claude.sdk_integration.ClaudeSDKClient", side_effect=mock_factory + ): + response = await sdk_manager.execute_command( + prompt="Test prompt", + working_directory=Path("/test"), + ) + + assert response.content == "OpenRouter response" + async def test_execute_command_with_streaming(self, sdk_manager): """Test command execution with streaming callback.""" stream_updates = [] From 0b8c8e2107f0cf8c5414321cd4ed71dc3301733e Mon Sep 17 00:00:00 2001 From: FrundlesTian <2929608755@qq.com> Date: Wed, 23 Sep 2026 22:05:05 +0800 Subject: [PATCH 2/2] fix: retry transient reply failures --- CHANGELOG.md | 1 + src/bot/core.py | 6 +- src/bot/handlers/message.py | 91 ++++++++++++------- src/bot/orchestrator.py | 63 ++++++++----- src/bot/utils/telegram_retry.py | 50 ++++++++++ src/claude/sdk_integration.py | 23 +++++ tests/unit/test_bot/test_core_proxy.py | 6 +- tests/unit/test_bot/test_telegram_retry.py | 43 +++++++++ .../unit/test_claude/test_sdk_integration.py | 76 ++++++++++++++-- 9 files changed, 290 insertions(+), 69 deletions(-) create mode 100644 src/bot/utils/telegram_retry.py create mode 100644 tests/unit/test_bot/test_telegram_retry.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f1e7d883..f6705382 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Fixed +- **Transient connection failures no longer silently drop replies**: Telegram progress, final-response, and global error messages now retry transport failures with bounded exponential backoff while invalid formatting still falls back immediately to plain text. Explicit proxies are applied to the separate `getUpdates` client as well as normal Bot API requests. Claude SDK streams must now end with a `ResultMessage`; an empty stream is retried, while a partially observed stream is not replayed because its tool calls may already have caused side effects (#241) - **OpenRouter empty-result fallback**: Anthropic-compatible providers that return an empty `ResultMessage.result` now fall back to the text in `AssistantMessage` instead of producing "No content to display" (#171) - **A run that stops early no longer reports success**: Claude produces no final text when the CLI kills a run at the turn limit, so the bot fell through to its "✅ Task completed" placeholder and told the user the work was done. A run that died at turn 10 mid-task and a run that finished were indistinguishable in Telegram (#172). `ResultMessage.subtype` is now read alongside the cost and session id, the placeholder claims completion only for `success`, and the reply carries a footer saying why the run ended — turn limit, cost budget, cancellation, or an unrecognised reason named by its raw subtype — with the turn count and a prompt to send another message to continue. Every place the bot renders a Claude reply carries it — typed messages, document, photo and voice input in agentic mode, the same four in classic mode, `/continue`, the Continue Session button and the quick-action buttons, whose `✅ … Complete` heading now reads `⚠️ … Stopped` when the run was cut short. Webhook-triggered and scheduled runs carry it too: nobody is watching those, so the reason a nightly job came back short is the whole of what its notification can say about it (#230) - **`ClaudeResponse.num_turns` is the turn count the CLI reports**: it was derived by counting `UserMessage` and `AssistantMessage` objects, which over-reports — every tool result arrives as another user message — so a run stopped at turn 10 could be recorded as having taken roughly twice that. It only reached the logs and the session store before; the stop-reason footer now shows it to the user, which made the approximation worth removing. `ResultMessage.num_turns` is used where the CLI supplies it, with the message count kept as the fallback for a result that carries none diff --git a/src/bot/core.py b/src/bot/core.py index 042aac72..31f03d51 100644 --- a/src/bot/core.py +++ b/src/bot/core.py @@ -27,6 +27,7 @@ from ..exceptions import ClaudeCodeTelegramError from .features.registry import FeatureRegistry from .orchestrator import MessageOrchestrator +from .utils.telegram_retry import retry_telegram_network logger = structlog.get_logger() @@ -105,6 +106,7 @@ async def initialize(self) -> None: proxy_url = os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY") if proxy_url: builder.proxy(proxy_url) + builder.get_updates_proxy(proxy_url) logger.info("Proxy configured", proxy=_redact_proxy_url(proxy_url)) self.app = builder.build() @@ -343,7 +345,9 @@ async def _error_handler( # Try to notify user if update and update.effective_message: try: - await update.effective_message.reply_text(user_message) + await retry_telegram_network( + lambda: update.effective_message.reply_text(user_message) + ) except Exception: logger.exception("Failed to send error message to user") diff --git a/src/bot/handlers/message.py b/src/bot/handlers/message.py index 2a192d34..c320d925 100644 --- a/src/bot/handlers/message.py +++ b/src/bot/handlers/message.py @@ -5,6 +5,7 @@ import structlog from telegram import InputMediaPhoto, Update +from telegram.error import BadRequest, NetworkError from telegram.ext import ContextTypes from ...claude.exceptions import ( @@ -26,6 +27,7 @@ should_send_as_photo, validate_image_path, ) +from ..utils.telegram_retry import retry_telegram_network logger = structlog.get_logger() @@ -305,6 +307,7 @@ async def handle_text_message( # Get services rate_limiter: Optional[RateLimiter] = context.bot_data.get("rate_limiter") audit_logger: Optional[AuditLogger] = context.bot_data.get("audit_logger") + progress_msg = None logger.info( "Processing text message", user_id=user_id, message_length=len(message_text) @@ -323,12 +326,18 @@ async def handle_text_message( return # Send typing indicator - await update.message.chat.send_action("typing") + try: + await update.message.chat.send_action("typing") + except NetworkError as exc: + # Typing indicators are cosmetic and must not abort a real request. + logger.debug("Failed to send typing action, ignoring", error=str(exc)) # Create progress message - progress_msg = await update.message.reply_text( - "🤔 Processing your request...", - reply_to_message_id=update.message.message_id, + progress_msg = await retry_telegram_network( + lambda: update.message.reply_text( + "🤔 Processing your request...", + reply_to_message_id=update.message.message_id, + ) ) # Get Claude integration and storage from context @@ -438,7 +447,10 @@ async def stream_handler(update_obj): ] # Delete progress message - await progress_msg.delete() + try: + await progress_msg.delete() + except Exception as exc: + logger.debug("Failed to delete progress message, ignoring", error=str(exc)) # Use MCP-collected images (from send_image_to_user tool calls) images: list[ImageAttachment] = mcp_images @@ -494,42 +506,49 @@ async def stream_handler(update_obj): # Send formatted responses (may be multiple messages) for i, message in enumerate(formatted_messages): try: - await update.message.reply_text( - message.text, - parse_mode=message.parse_mode, - reply_markup=message.reply_markup, - reply_to_message_id=( - update.message.message_id if i == 0 else None - ), + await retry_telegram_network( + lambda: update.message.reply_text( + message.text, + parse_mode=message.parse_mode, + reply_markup=message.reply_markup, + reply_to_message_id=( + update.message.message_id if i == 0 else None + ), + ) ) if i < len(formatted_messages) - 1: await asyncio.sleep(0.5) - except Exception as send_err: + except BadRequest as send_err: logger.warning( - "Failed to send HTML response, retrying as plain text", + "Failed to send formatted response, retrying as plain text", error=str(send_err), message_index=i, ) try: - await update.message.reply_text( - message.text, - reply_markup=message.reply_markup, - reply_to_message_id=( - update.message.message_id if i == 0 else None - ), + await retry_telegram_network( + lambda: update.message.reply_text( + message.text, + reply_markup=message.reply_markup, + reply_to_message_id=( + update.message.message_id if i == 0 else None + ), + ) ) except Exception as plain_err: + plain_error_text = str(plain_err)[:150] logger.error( "Failed to send plain text fallback response", - error=str(plain_err), + error=plain_error_text, ) - await update.message.reply_text( - f"Failed to deliver response " - f"(Telegram error: {str(plain_err)[:150]}). " - f"Please try again.", - reply_to_message_id=( - update.message.message_id if i == 0 else None - ), + await retry_telegram_network( + lambda: update.message.reply_text( + f"Failed to deliver response " + f"(Telegram error: {plain_error_text}). " + f"Please try again.", + reply_to_message_id=( + update.message.message_id if i == 0 else None + ), + ) ) # Send images separately @@ -635,12 +654,18 @@ async def stream_handler(update_obj): except Exception as e: # Clean up progress message if it exists - try: - await progress_msg.delete() - except Exception as delete_error: - logger.debug("Failed to delete progress message", error=str(delete_error)) + if progress_msg is not None: + try: + await progress_msg.delete() + except Exception as delete_error: + logger.debug( + "Failed to delete progress message", error=str(delete_error) + ) - await update.message.reply_text(_format_error_message(e), parse_mode="HTML") + error_message = _format_error_message(e) + await retry_telegram_network( + lambda: update.message.reply_text(error_message, parse_mode="HTML") + ) # Log failed processing if audit_logger: diff --git a/src/bot/orchestrator.py b/src/bot/orchestrator.py index 5801772e..c3f72b7e 100644 --- a/src/bot/orchestrator.py +++ b/src/bot/orchestrator.py @@ -21,6 +21,7 @@ InputMediaPhoto, Update, ) +from telegram.error import BadRequest, NetworkError from telegram.ext import ( Application, CallbackQueryHandler, @@ -40,6 +41,7 @@ should_send_as_photo, validate_image_path, ) +from .utils.telegram_retry import retry_telegram_network logger = structlog.get_logger() @@ -993,7 +995,11 @@ async def agentic_text( return chat = update.message.chat - await chat.send_action("typing") + try: + await chat.send_action("typing") + except NetworkError as exc: + # Typing indicators are cosmetic and must not abort a real request. + logger.debug("Failed to send typing action, ignoring", error=str(exc)) verbose_level = self._get_verbose_level(context) @@ -1002,8 +1008,8 @@ async def agentic_text( stop_kb = InlineKeyboardMarkup( [[InlineKeyboardButton("Stop", callback_data=f"stop:{user_id}")]] ) - progress_msg = await update.message.reply_text( - "Working...", reply_markup=stop_kb + progress_msg = await retry_telegram_network( + lambda: update.message.reply_text("Working...", reply_markup=stop_kb) ) # Register active request for stop callback @@ -1171,38 +1177,45 @@ async def agentic_text( if not message.text or not message.text.strip(): continue try: - await update.message.reply_text( - message.text, - parse_mode=message.parse_mode, - reply_markup=None, # No keyboards in agentic mode - reply_to_message_id=( - update.message.message_id if i == 0 else None - ), + await retry_telegram_network( + lambda: update.message.reply_text( + message.text, + parse_mode=message.parse_mode, + reply_markup=None, # No keyboards in agentic mode + reply_to_message_id=( + update.message.message_id if i == 0 else None + ), + ) ) if i < len(formatted_messages) - 1: await asyncio.sleep(0.5) - except Exception as send_err: + except BadRequest as send_err: logger.warning( - "Failed to send HTML response, retrying as plain text", + "Failed to send formatted response, retrying as plain text", error=str(send_err), message_index=i, ) try: - await update.message.reply_text( - message.text, - reply_markup=None, - reply_to_message_id=( - update.message.message_id if i == 0 else None - ), + await retry_telegram_network( + lambda: update.message.reply_text( + message.text, + reply_markup=None, + reply_to_message_id=( + update.message.message_id if i == 0 else None + ), + ) ) except Exception as plain_err: - await update.message.reply_text( - f"Failed to deliver response " - f"(Telegram error: {str(plain_err)[:150]}). " - f"Please try again.", - reply_to_message_id=( - update.message.message_id if i == 0 else None - ), + plain_error_text = str(plain_err)[:150] + await retry_telegram_network( + lambda: update.message.reply_text( + f"Failed to deliver response " + f"(Telegram error: {plain_error_text}). " + f"Please try again.", + reply_to_message_id=( + update.message.message_id if i == 0 else None + ), + ) ) # Send images separately if caption wasn't used diff --git a/src/bot/utils/telegram_retry.py b/src/bot/utils/telegram_retry.py new file mode 100644 index 00000000..e5ca3c6b --- /dev/null +++ b/src/bot/utils/telegram_retry.py @@ -0,0 +1,50 @@ +"""Retry helpers for transient Telegram transport failures.""" + +import asyncio +from typing import Awaitable, Callable, TypeVar + +import structlog +from telegram.error import BadRequest, NetworkError + +logger = structlog.get_logger() + +T = TypeVar("T") + + +async def retry_telegram_network( + operation: Callable[[], Awaitable[T]], + *, + max_attempts: int = 3, + base_delay: float = 0.5, +) -> T: + """Run a Telegram request again after transient transport failures. + + ``BadRequest`` inherits from PTB's ``NetworkError`` even though it represents + a permanent request problem (for example invalid HTML), so it must never be + retried here. Callers can handle it separately with a formatting fallback. + """ + if max_attempts < 1: + raise ValueError("max_attempts must be at least 1") + if base_delay < 0: + raise ValueError("base_delay must not be negative") + + for attempt in range(max_attempts): + try: + return await operation() + except BadRequest: + raise + except NetworkError as exc: + if attempt == max_attempts - 1: + raise + + delay = base_delay * (2**attempt) + logger.warning( + "Transient Telegram request failure, retrying", + attempt=attempt + 1, + max_attempts=max_attempts, + delay_seconds=delay, + error_type=type(exc).__name__, + ) + await asyncio.sleep(delay) + + raise AssertionError("Telegram retry loop exited unexpectedly") diff --git a/src/claude/sdk_integration.py b/src/claude/sdk_integration.py index f4231004..2bbdc641 100644 --- a/src/claude/sdk_integration.py +++ b/src/claude/sdk_integration.py @@ -63,6 +63,14 @@ RESULT_SUBTYPE_SUCCESS = "success" +class _IncompleteMessageStreamError(CLIConnectionError): + """The SDK transport closed before delivering its terminal result.""" + + def __init__(self, *, retry_safe: bool): + super().__init__("Claude message stream ended before ResultMessage") + self.retry_safe = retry_safe + + def _as_error_list(value: Any) -> List[str]: """Normalise ResultMessage.errors into a list of strings.""" if not value: @@ -394,6 +402,8 @@ def _is_retryable_error(self, exc: BaseException) -> bool: Only non-MCP CLIConnectionError is considered transient. """ if isinstance(exc, CLIConnectionError): + if isinstance(exc, _IncompleteMessageStreamError): + return exc.retry_safe msg = str(exc).lower() return "mcp" not in msg # "server" alone is too broad return False @@ -568,6 +578,8 @@ def _stderr_callback(line: str) -> None: async def _run_client() -> None: client = ClaudeSDKClient(options) + received_raw_message = False + received_result = False try: await client.connect() @@ -603,6 +615,7 @@ async def _multimodal_prompt() -> AsyncIterator[Dict[str, Any]]: await client.query(prompt) async for raw_data in client._query.receive_messages(): + received_raw_message = True try: message = parse_message(raw_data) except MessageParseError as e: @@ -615,6 +628,7 @@ async def _multimodal_prompt() -> AsyncIterator[Dict[str, Any]]: messages.append(message) if isinstance(message, ResultMessage): + received_result = True break # Handle streaming callback @@ -629,6 +643,15 @@ async def _multimodal_prompt() -> AsyncIterator[Dict[str, Any]]: error=str(callback_error), error_type=type(callback_error).__name__, ) + + if not received_result: + # An SDK "end" sentinel (or transport EOF) can close the + # iterator without raising. Retrying is safe only when no + # message was observed; replaying after partial output may + # duplicate tool side effects. + raise _IncompleteMessageStreamError( + retry_safe=not received_raw_message + ) finally: await client.disconnect() diff --git a/tests/unit/test_bot/test_core_proxy.py b/tests/unit/test_bot/test_core_proxy.py index 0ccd795d..f6a9f7a9 100644 --- a/tests/unit/test_bot/test_core_proxy.py +++ b/tests/unit/test_bot/test_core_proxy.py @@ -88,7 +88,7 @@ def test_unparsable_url_is_not_echoed(self): async def test_initialize_configures_proxy_from_environment( bot_with_builder, monkeypatch ): - """HTTPS_PROXY in the environment must reach builder.proxy().""" + """HTTPS_PROXY must reach both PTB request clients.""" monkeypatch.setenv("HTTPS_PROXY", "http://alice:s3cret@proxy.internal:3128") monkeypatch.delenv("HTTP_PROXY", raising=False) bot, builder = bot_with_builder @@ -96,6 +96,9 @@ async def test_initialize_configures_proxy_from_environment( await bot.initialize() builder.proxy.assert_called_once_with("http://alice:s3cret@proxy.internal:3128") + builder.get_updates_proxy.assert_called_once_with( + "http://alice:s3cret@proxy.internal:3128" + ) @pytest.mark.asyncio @@ -132,3 +135,4 @@ async def test_initialize_skips_proxy_when_unset(bot_with_builder, monkeypatch): await bot.initialize() builder.proxy.assert_not_called() + builder.get_updates_proxy.assert_not_called() diff --git a/tests/unit/test_bot/test_telegram_retry.py b/tests/unit/test_bot/test_telegram_retry.py new file mode 100644 index 00000000..885bb4f6 --- /dev/null +++ b/tests/unit/test_bot/test_telegram_retry.py @@ -0,0 +1,43 @@ +"""Tests for transient Telegram request retries.""" + +from unittest.mock import AsyncMock + +import pytest +from telegram.error import BadRequest, TimedOut + +from src.bot.utils.telegram_retry import retry_telegram_network + + +async def test_retries_transient_network_failures() -> None: + operation = AsyncMock(side_effect=[TimedOut("slow"), TimedOut("slow"), "sent"]) + sleep = AsyncMock() + + with pytest.MonkeyPatch.context() as monkeypatch: + monkeypatch.setattr("src.bot.utils.telegram_retry.asyncio.sleep", sleep) + result = await retry_telegram_network(operation) + + assert result == "sent" + assert operation.await_count == 3 + assert [call.args[0] for call in sleep.await_args_list] == [0.5, 1.0] + + +async def test_does_not_retry_bad_request() -> None: + operation = AsyncMock(side_effect=BadRequest("invalid HTML")) + + with pytest.raises(BadRequest, match="invalid HTML"): + await retry_telegram_network(operation) + + operation.assert_awaited_once() + + +async def test_raises_after_transient_retry_budget_is_exhausted() -> None: + operation = AsyncMock(side_effect=TimedOut("still unavailable")) + sleep = AsyncMock() + + with pytest.MonkeyPatch.context() as monkeypatch: + monkeypatch.setattr("src.bot.utils.telegram_retry.asyncio.sleep", sleep) + with pytest.raises(TimedOut, match="still unavailable"): + await retry_telegram_network(operation) + + assert operation.await_count == 3 + assert sleep.await_count == 2 diff --git a/tests/unit/test_claude/test_sdk_integration.py b/tests/unit/test_claude/test_sdk_integration.py index 9b013431..b0809f31 100644 --- a/tests/unit/test_claude/test_sdk_integration.py +++ b/tests/unit/test_claude/test_sdk_integration.py @@ -426,7 +426,7 @@ async def flaky_receive(): if call_count == 1: raise CLIConnectionError("connection reset") # Second attempt succeeds - yield a ResultMessage - yield + yield _make_result_message() # Use a config with 2 attempts sdk_manager.config.claude_retry_max_attempts = 2 @@ -439,17 +439,75 @@ async def flaky_receive(): query_mock.receive_messages = flaky_receive client._query = query_mock - # Should not raise - second attempt succeeds with patch("src.claude.sdk_integration.ClaudeSDKClient", return_value=client): with patch("asyncio.sleep", new_callable=AsyncMock): - try: - await sdk_manager.execute_command( - prompt="Test", - working_directory=Path("/test"), - ) - except Exception: - pass # Response parsing may fail - what matters is retry happened + response = await sdk_manager.execute_command( + prompt="Test", + working_directory=Path("/test"), + ) + + assert call_count == 2 + assert response.content == "Success" + + async def test_retry_when_stream_ends_before_any_message(self, sdk_manager): + """A clean EOF without ResultMessage is a retryable transport failure.""" + call_count = 0 + + async def flaky_receive(): + nonlocal call_count + call_count += 1 + if call_count == 1: + return + yield _make_assistant_message("Recovered response") + yield _make_result_message(result="Recovered response") + + sdk_manager.config.claude_retry_max_attempts = 2 + client = AsyncMock() + client.connect = AsyncMock() + client.disconnect = AsyncMock() + client.query = AsyncMock() + query_mock = AsyncMock() + query_mock.receive_messages = flaky_receive + client._query = query_mock + + with patch("src.claude.sdk_integration.ClaudeSDKClient", return_value=client): + with patch("asyncio.sleep", new_callable=AsyncMock): + response = await sdk_manager.execute_command( + prompt="Test", + working_directory=Path("/test"), + ) + assert call_count == 2 + assert response.content == "Recovered response" + + async def test_does_not_replay_after_partial_stream(self, sdk_manager): + """Partial output may include tool side effects and must not be replayed.""" + from src.claude.exceptions import ClaudeProcessError + + call_count = 0 + + async def partial_receive(): + nonlocal call_count + call_count += 1 + yield _make_assistant_message("Partial response") + + sdk_manager.config.claude_retry_max_attempts = 3 + client = AsyncMock() + client.connect = AsyncMock() + client.disconnect = AsyncMock() + client.query = AsyncMock() + query_mock = AsyncMock() + query_mock.receive_messages = partial_receive + client._query = query_mock + + with patch("src.claude.sdk_integration.ClaudeSDKClient", return_value=client): + with pytest.raises(ClaudeProcessError, match="before ResultMessage"): + await sdk_manager.execute_command( + prompt="Test", + working_directory=Path("/test"), + ) + + assert call_count == 1 async def test_no_retry_on_mcp_connection_error(self, sdk_manager): """Test that MCP CLIConnectionError is NOT retried."""