fix: 增强异常处理和Issue#12修复

- consumer.py:
  * 添加lambda_handler的JSON解析异常处理
  * 包装asyncio.run的异常捕获和重新抛出
  * 初始化result默认值防止未定义错误
  * 添加message为None的检查
  * 增强异常处理的日志记录
  * 修复Issue#12：使用 or 运算符处理空response

- handler.py:
  * 包装asyncio.run的异常处理，返回结构化错误响应
  * 添加session操作的异常处理
  * 关键操作失败返回500，非关键操作log后继续
  * 完整的日志记录便于调试

确保系统在异常情况下不会导致SQS消息卡死，并正确处理Agent返回空响应的场景。

Author: BukeLy

agent-sdk-client/consumer.py

@@ -18,28 +18,58 @@
 def lambda_handler(event: dict, context: Any) -> dict:
     """SQS Consumer Lambda entry point."""
     for record in event['Records']:
-        message_data = json.loads(record['body'])
-        asyncio.run(process_message(message_data))
+        try:
+            message_data = json.loads(record['body'])
+        except json.JSONDecodeError as e:
+            # Invalid message format - log and skip
+            import logging
+            logger = logging.getLogger()
+            logger.error(f"Failed to parse SQS message: {e}")
+            continue
+
+        try:
+            asyncio.run(process_message(message_data))
+        except Exception as e:
+            # Log and let SQS retry on failure
+            import logging
+            logger = logging.getLogger()
+            logger.exception(f"Failed to process message: {e}")
+            raise  # Re-raise to fail the batch item
 
     return {'statusCode': 200}
 
 
 async def process_message(message_data: dict) -> None:
     """Process single message from SQS queue."""
+    import logging
+    logger = logging.getLogger()
+
     config = Config.from_env()
     bot = Bot(config.telegram_token)
 
     # Reconstruct Update object from stored data
     update = Update.de_json(message_data['telegram_update'], bot)
     message = update.message or update.edited_message
 
+    if not message:
+        logger.warning("Received update with no message or edited_message")
+        return
+
     # Send typing indicator
     await bot.send_chat_action(
         chat_id=message.chat_id,
         action=ChatAction.TYPING,
         message_thread_id=message.message_thread_id,
     )
 
+    # Initialize result with default error response
+    # This ensures result is always defined, even if Agent Server call fails
+    result = {
+        'response': '',
+        'is_error': True,
+        'error_message': 'Failed to get response from Agent Server'
+    }
+
     # Call Agent Server
     try:
         async with httpx.AsyncClient(timeout=600.0) as client:
@@ -59,6 +89,7 @@ async def process_message(message_data: dict) -> None:
             result = response.json()
 
     except httpx.TimeoutException:
+        logger.warning(f"Agent Server timeout for chat_id={message.chat_id}")
         await bot.send_message(
             chat_id=message.chat_id,
             text="Request timed out.",
@@ -67,19 +98,23 @@ async def process_message(message_data: dict) -> None:
         raise  # Re-raise to trigger SQS retry for transient errors
 
     except Exception as e:
-        await bot.send_message(
-            chat_id=message.chat_id,
-            text=f"Error: {str(e)[:200]}",
-            message_thread_id=message.message_thread_id,
-        )
-        # Don't re-raise for general exceptions - error message already sent
-        # to user, retrying would cause duplicate messages
+        logger.exception(f"Agent Server error for chat_id={message.chat_id}")
+        error_text = f"Error: {str(e)[:200]}"
+        try:
+            await bot.send_message(
+                chat_id=message.chat_id,
+                text=error_text,
+                message_thread_id=message.message_thread_id,
+            )
+        except Exception as send_error:
+            logger.error(f"Failed to send error message to Telegram: {send_error}")
+        # Don't re-raise - error message already sent to user, retrying would cause duplicate messages
 
-    # Format response
+    # Format response (result is guaranteed to be defined now)
     if result.get('is_error'):
         text = f"Agent error: {result.get('error_message', 'Unknown')}"
     else:
-        text = result.get('response', 'No response')
+        text = result.get('response') or 'No response'
 
     if len(text) > 4000:
         text = text[:4000] + "\n\n... (truncated)"

agent-sdk-server/handler.py

@@ -4,12 +4,16 @@
 """
 import asyncio
 import json
+import logging
 from typing import Any
 
 from config import Config
 from session_store import SessionStore
 from agent_session import process_message
 
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
 
 def lambda_handler(event: dict, context: Any) -> dict:
     """Lambda entry point.
@@ -76,26 +80,56 @@ def lambda_handler(event: dict, context: Any) -> dict:
 
     # Download session files if resuming
     if session_id:
-        store.download_session_files(session_id)
+        try:
+            store.download_session_files(session_id)
+        except Exception as e:
+            logger.error(f"Failed to download session files for session_id={session_id}: {e}")
+            return {
+                'statusCode': 500,
+                'body': json.dumps({'error': 'Failed to download session'})
+            }
 
     # Process message with Agent SDK
-    result = asyncio.run(process_message(
-        user_message=user_message,
-        session_id=session_id,
-        model=model,
-    ))
+    try:
+        result = asyncio.run(process_message(
+            user_message=user_message,
+            session_id=session_id,
+            model=model,
+        ))
+    except Exception as e:
+        logger.exception(f"Agent SDK processing failed for chat_id={chat_id}")
+        return {
+            'statusCode': 500,
+            'headers': {'Content-Type': 'application/json'},
+            'body': json.dumps({
+                'response': '',
+                'session_id': session_id,
+                'cost_usd': 0.0,
+                'num_turns': 0,
+                'is_error': True,
+                'error_message': f'Agent processing error: {str(e)[:200]}'
+            })
+        }
 
     # Get session_id from result (SDK generates it for new sessions)
     result_session_id = result.get('session_id', '')
 
     # Save session mapping if new session
     if result_session_id and result_session_id != session_id:
-        store.save_session_id(chat_id, thread_id, result_session_id)
+        try:
+            store.save_session_id(chat_id, thread_id, result_session_id)
+        except Exception as e:
+            logger.error(f"Failed to save session mapping for chat_id={chat_id}: {e}")
+            # Continue anyway - session can still work with S3 files
 
     # Upload session files
     if result_session_id:
-        store.upload_session_files(result_session_id)
-        store.update_session_timestamp(chat_id, thread_id)
+        try:
+            store.upload_session_files(result_session_id)
+            store.update_session_timestamp(chat_id, thread_id)
+        except Exception as e:
+            logger.error(f"Failed to upload session files for session_id={result_session_id}: {e}")
+            # Continue anyway - result is already available to return
 
     return {
         'statusCode': 200,