From a4d37dbcd9c3665025e06667aab71183c2e890ee Mon Sep 17 00:00:00 2001
From: Cameron Pfiffer <cameron@pfiffer.org>
Date: Fri, 26 Dec 2025 01:34:24 +0000
Subject: [PATCH] Add timeout protection and silence ping messages in streaming

- Add 10-minute timeout to prevent infinite hangs in streaming loops
- Silence 'ping' keepalive messages (log at debug level only)
- Apply to both notification processing and synthesis streams
- Prevents logs from being spammed with ping messages
- Stream will break after 600 seconds if agent doesn't send 'done'

This fixes the issue where streaming would hang indefinitely
when agents get stuck or don't complete properly.

🐾 Generated with [Letta Code](https://letta.com)

Co-Authored-By: Letta <noreply@letta.com>
---
 bsky.py | 24 ++++++++++++++++++++++--
 1 file(s) changed, 22 insertion(s)(+), 2 deletion(s)(-)

diff --git a/bsky.py b/bsky.py
--- a/bsky.py
+++ b/bsky.py
@@ -419,9 +419,16 @@
                 max_steps=100
             )
             
-            # Collect the streaming response
+            # Collect the streaming response with timeout protection
             all_messages = []
+            stream_start_time = time.time()
+            max_stream_duration = 600  # 10 minutes max
+            
             for chunk in message_stream:
+                # Check for timeout
+                if time.time() - stream_start_time > max_stream_duration:
+                    logger.warning(f"Stream exceeded {max_stream_duration}s timeout, breaking")
+                    break
                 # Log condensed chunk info
                 if hasattr(chunk, 'message_type'):
                     if chunk.message_type == 'reasoning_message':
@@ -606,6 +613,9 @@
                             logger.error(f"Agent error (dict): {chunk.model_dump()}")
                         elif hasattr(chunk, '__dict__'):
                             logger.error(f"Agent error (vars): {vars(chunk)}")
+                    elif chunk.message_type == 'ping':
+                        # Silently ignore ping keepalive messages
+                        logger.debug(f"Received keepalive ping from Letta API")
                     else:
                         # Filter out verbose message types
                         if chunk.message_type not in ['usage_statistics', 'stop_reason']:
@@ -1582,8 +1592,15 @@
         synthesis_posts = []
         ack_note = None
         
-        # Process the streaming response
+        # Process the streaming response with timeout protection
+        stream_start_time = time.time()
+        max_stream_duration = 600  # 10 minutes max
+        
         for chunk in message_stream:
+            # Check for timeout
+            if time.time() - stream_start_time > max_stream_duration:
+                logger.warning(f"Synthesis stream exceeded {max_stream_duration}s timeout, breaking")
+                break
             if hasattr(chunk, 'message_type'):
                 if chunk.message_type == 'reasoning_message':
                     if SHOW_REASONING:
@@ -1669,6 +1686,9 @@
                     print("  ──────────────────")
                     for line in chunk.content.split('\n'):
                         print(f"  {line}")
+                elif chunk.message_type == 'ping':
+                    # Silently ignore ping keepalive messages
+                    logger.debug(f"Received keepalive ping from Letta API during synthesis")
                 elif chunk.message_type == 'error_message':
                     # Dump full error object
                     logger.error(f"Synthesis error_message: {chunk}")

--
tangled.sh