From 4a75ea9f998cdf09ab3790165ffe07f94755d74f Mon Sep 17 00:00:00 2001 From: AndyMik90 Date: Fri, 20 Feb 2026 11:16:59 +0100 Subject: [PATCH] fix: handle unknown SDK message types (rate_limit_event) to prevent session crashes The Claude CLI emits message types like rate_limit_event that the SDK's message_parser doesn't recognize, causing MessageParseError to kill the entire agent session stream. This adds two layers of defense: 1. Monkey-patch SDK's parse_message to convert unknown types into safe SystemMessage objects instead of raising 2. safe_receive_messages() wrapper around receive_response() that filters patched messages and catches stream-level errors gracefully Applied to all agent consumers: session, qa_reviewer, qa_fixer, and agent_runner. Also bumps minimum SDK to >=0.1.39. Co-Authored-By: Claude Opus 4.6 --- apps/backend/agents/session.py | 3 +- apps/backend/core/client.py | 71 ++++++++++++++++++++++ apps/backend/core/error_utils.py | 68 +++++++++++++++++++++ apps/backend/qa/fixer.py | 8 ++- apps/backend/qa/reviewer.py | 8 ++- apps/backend/requirements.txt | 7 ++- apps/backend/spec/pipeline/agent_runner.py | 3 +- package-lock.json | 6 +- 8 files changed, 162 insertions(+), 12 deletions(-) diff --git a/apps/backend/agents/session.py b/apps/backend/agents/session.py index 437628db..81fdf261 100644 --- a/apps/backend/agents/session.py +++ b/apps/backend/agents/session.py @@ -14,6 +14,7 @@ from core.error_utils import ( is_authentication_error, is_rate_limit_error, is_tool_concurrency_error, + safe_receive_messages, ) from core.file_utils import write_json_atomic from debug import debug, debug_detailed, debug_error, debug_section, debug_success @@ -490,7 +491,7 @@ async def run_agent_session( # Collect response text and show tool use response_text = "" debug("session", "Starting to receive response stream...") - async for msg in client.receive_response(): + async for msg in safe_receive_messages(client, caller="session"): msg_type = type(msg).__name__ message_count += 1 debug_detailed( diff --git a/apps/backend/core/client.py b/apps/backend/core/client.py index db1d8ab4..a21e3959 100644 --- a/apps/backend/core/client.py +++ b/apps/backend/core/client.py @@ -29,6 +29,77 @@ from core.platform import ( logger = logging.getLogger(__name__) +# ============================================================================= +# SDK Message Parser Patch +# ============================================================================= +# The Claude Agent SDK's message_parser raises MessageParseError for unknown +# message types (e.g., "rate_limit_event"). Since parse_message runs inside an +# async generator, the exception kills the entire agent session stream. +# Patch to log a warning and return a SystemMessage instead of crashing. +# This is needed until the SDK natively handles all CLI message types. + + +def _patch_sdk_message_parser() -> None: + """Patch the SDK's parse_message to handle unknown message types gracefully. + + The Claude CLI may emit message types that the installed SDK version doesn't + recognize (e.g., rate_limit_event, usage_event). Without this patch, any + unrecognized type raises MessageParseError inside the SDK's async generator, + which terminates the entire response stream and kills the agent session. + + The patch converts unknown types into SystemMessage objects with a + 'unknown_' subtype, which all message consumers silently skip. + """ + try: + import claude_agent_sdk._internal.message_parser as _parser + from claude_agent_sdk._errors import MessageParseError + from claude_agent_sdk.types import SystemMessage + + _original_parse = _parser.parse_message + + def _patched_parse(data): + try: + return _original_parse(data) + except MessageParseError as e: + msg = str(e) + if "Unknown message type" in msg: + msg_type = ( + data.get("type", "unknown") + if isinstance(data, dict) + else "unknown" + ) + # Rate limit events deserve a visible warning; others just debug-level + if "rate_limit" in msg_type: + retry_after = ( + data.get("retry_after") + or data.get("data", {}).get("retry_after") + if isinstance(data, dict) + else None + ) + retry_info = ( + f" (retry_after={retry_after}s)" if retry_after else "" + ) + logger.warning( + f"Rate limit event received from CLI{retry_info} — " + f"the SDK will handle backoff automatically" + ) + else: + logger.debug( + f"SDK received unhandled message type '{msg_type}', skipping" + ) + return SystemMessage( + subtype=f"unknown_{msg_type}", + data=data if isinstance(data, dict) else {}, + ) + raise + + _parser.parse_message = _patched_parse + except Exception as e: + logger.warning(f"Failed to patch SDK message parser: {e}") + + +_patch_sdk_message_parser() + # ============================================================================= # Windows System Prompt Limits # ============================================================================= diff --git a/apps/backend/core/error_utils.py b/apps/backend/core/error_utils.py index 28c1b155..120db0d9 100644 --- a/apps/backend/core/error_utils.py +++ b/apps/backend/core/error_utils.py @@ -6,7 +6,17 @@ Common error detection and classification functions used across agent sessions, QA, and other modules. """ +from __future__ import annotations + +import logging import re +from collections.abc import AsyncIterator +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from claude_agent_sdk.types import Message + +logger = logging.getLogger(__name__) def is_tool_concurrency_error(error: Exception) -> bool: @@ -118,3 +128,61 @@ def is_authentication_error(error: Exception) -> bool: "please login again", ] ) + + +async def safe_receive_messages( + client, + *, + caller: str = "agent", +) -> AsyncIterator[Message]: + """Iterate over SDK messages with resilience against unexpected errors. + + The SDK's ``receive_response()`` async generator can terminate early if: + 1. An unhandled message type slips past the monkey-patch (e.g., SDK upgrade + removes the patch surface). + 2. A transient parse error corrupts a single message in the stream. + 3. An unexpected ``StopAsyncIteration`` or runtime error occurs mid-stream. + + This wrapper catches per-message errors, logs them, and continues yielding + subsequent messages so the agent session can complete its work. + + It also detects rate-limit events (surfaced as ``SystemMessage`` with + subtype ``unknown_rate_limit_event``) and logs a user-visible warning. + + Args: + client: A ``ClaudeSDKClient`` instance (must be inside ``async with``). + caller: Label for log messages (e.g., "session", "agent_runner"). + + Yields: + Parsed ``Message`` objects from the SDK response stream. + """ + try: + async for msg in client.receive_response(): + # Detect rate-limit events surfaced by the monkey-patch + msg_type = type(msg).__name__ + if msg_type == "SystemMessage": + subtype = getattr(msg, "subtype", "") + if subtype.startswith("unknown_"): + original_type = subtype[len("unknown_") :] + if "rate_limit" in original_type: + data = getattr(msg, "data", {}) + retry_after = data.get("retry_after") or data.get( + "data", {} + ).get("retry_after") + retry_info = ( + f" (retry in {retry_after}s)" if retry_after else "" + ) + logger.warning(f"[{caller}] Rate limit event{retry_info}") + else: + logger.debug( + f"[{caller}] Skipping unknown SDK message type: {original_type}" + ) + continue + yield msg + except GeneratorExit: + return + except Exception as e: + # If the generator itself raises (e.g., transport error), log and stop + # gracefully so callers can process whatever was collected so far. + logger.error(f"[{caller}] SDK response stream terminated unexpectedly: {e}") + return diff --git a/apps/backend/qa/fixer.py b/apps/backend/qa/fixer.py index 815724ff..290983f8 100644 --- a/apps/backend/qa/fixer.py +++ b/apps/backend/qa/fixer.py @@ -15,7 +15,11 @@ from pathlib import Path from agents.base import sanitize_error_message from agents.memory_manager import get_graphiti_context, save_session_memory from claude_agent_sdk import ClaudeSDKClient -from core.error_utils import is_rate_limit_error, is_tool_concurrency_error +from core.error_utils import ( + is_rate_limit_error, + is_tool_concurrency_error, + safe_receive_messages, +) from debug import debug, debug_detailed, debug_error, debug_section, debug_success from security.tool_input_validator import get_safe_tool_input from task_logger import ( @@ -141,7 +145,7 @@ async def run_qa_fixer_session( response_text = "" debug("qa_fixer", "Starting to receive response stream...") - async for msg in client.receive_response(): + async for msg in safe_receive_messages(client, caller="qa_fixer"): msg_type = type(msg).__name__ message_count += 1 debug_detailed( diff --git a/apps/backend/qa/reviewer.py b/apps/backend/qa/reviewer.py index 229625be..6bbdcd9c 100644 --- a/apps/backend/qa/reviewer.py +++ b/apps/backend/qa/reviewer.py @@ -16,7 +16,11 @@ from pathlib import Path from agents.base import sanitize_error_message from agents.memory_manager import get_graphiti_context, save_session_memory from claude_agent_sdk import ClaudeSDKClient -from core.error_utils import is_rate_limit_error, is_tool_concurrency_error +from core.error_utils import ( + is_rate_limit_error, + is_tool_concurrency_error, + safe_receive_messages, +) from debug import debug, debug_detailed, debug_error, debug_section, debug_success from prompts_pkg import get_qa_reviewer_prompt from security.tool_input_validator import get_safe_tool_input @@ -195,7 +199,7 @@ This is attempt {previous_error.get("consecutive_errors", 1) + 1}. If you fail t response_text = "" debug("qa_reviewer", "Starting to receive response stream...") - async for msg in client.receive_response(): + async for msg in safe_receive_messages(client, caller="qa_reviewer"): msg_type = type(msg).__name__ message_count += 1 debug_detailed( diff --git a/apps/backend/requirements.txt b/apps/backend/requirements.txt index 2cd44d8e..47ee7e92 100644 --- a/apps/backend/requirements.txt +++ b/apps/backend/requirements.txt @@ -1,7 +1,8 @@ # Auto-Build Framework Dependencies -# SDK 0.1.33+ required for Opus 4.6 adaptive thinking support -# Earlier versions lacked effort parameter and thinking type configuration -claude-agent-sdk>=0.1.33 +# SDK 0.1.39+ required for Opus 4.6 adaptive thinking support and stability fixes +# Earlier versions lacked effort parameter, thinking type configuration, +# and crashed on unhandled CLI message types (e.g., rate_limit_event) +claude-agent-sdk>=0.1.39 python-dotenv>=1.0.0 # TOML parsing fallback for Python < 3.11 diff --git a/apps/backend/spec/pipeline/agent_runner.py b/apps/backend/spec/pipeline/agent_runner.py index 3b0f621c..4ebe0ff6 100644 --- a/apps/backend/spec/pipeline/agent_runner.py +++ b/apps/backend/spec/pipeline/agent_runner.py @@ -12,6 +12,7 @@ from ui.capabilities import configure_safe_encoding configure_safe_encoding() +from core.error_utils import safe_receive_messages from debug import debug, debug_detailed, debug_error, debug_section, debug_success from security.tool_input_validator import get_safe_tool_input from task_logger import ( @@ -162,7 +163,7 @@ class AgentRunner: response_text = "" debug("agent_runner", "Starting to receive response stream...") - async for msg in client.receive_response(): + async for msg in safe_receive_messages(client, caller="agent_runner"): msg_type = type(msg).__name__ message_count += 1 debug_detailed( diff --git a/package-lock.json b/package-lock.json index 2c05a728..31ab465a 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "auto-claude", - "version": "2.7.6-beta.3", + "version": "2.7.6-beta.6", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "auto-claude", - "version": "2.7.6-beta.3", + "version": "2.7.6-beta.6", "license": "AGPL-3.0", "workspaces": [ "apps/*", @@ -25,7 +25,7 @@ }, "apps/frontend": { "name": "auto-claude-ui", - "version": "2.7.6-beta.3", + "version": "2.7.6-beta.6", "hasInstallScript": true, "license": "AGPL-3.0", "dependencies": {