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 <noreply@anthropic.com>

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(

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_<type>' 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
 # =============================================================================

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

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(

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(

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

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(

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": {