Fix bot response issues and enhance outbound message logging

+45 -10

bsky.py

··· 206 206 None: Failed with non-retryable error, move to errors directory 207 207 "no_reply": No reply was generated, move to no_reply directory 208 208 """ 209 + import uuid 210 + 211 + # Generate correlation ID for tracking this notification through the pipeline 212 + correlation_id = str(uuid.uuid4())[:8] 213 + 209 214 try: 210 - logger.debug(f"Starting process_mention with notification_data type: {type(notification_data)}") 215 + logger.info(f"[{correlation_id}] Starting process_mention", extra={ 216 + 'correlation_id': correlation_id, 217 + 'notification_type': type(notification_data).__name__ 218 + }) 211 219 212 220 # Handle both dict and object inputs for backwards compatibility 213 221 if isinstance(notification_data, dict): ··· 222 230 author_handle = notification_data.author.handle 223 231 author_name = notification_data.author.display_name or author_handle 224 232 225 - logger.debug(f"Extracted data - URI: {uri}, Author: @{author_handle}, Text: {mention_text[:50]}...") 233 + logger.info(f"[{correlation_id}] Processing mention from @{author_handle}", extra={ 234 + 'correlation_id': correlation_id, 235 + 'author_handle': author_handle, 236 + 'author_name': author_name, 237 + 'mention_uri': uri, 238 + 'mention_text_length': len(mention_text), 239 + 'mention_preview': mention_text[:100] if mention_text else '' 240 + }) 226 241 227 242 # Retrieve the entire thread associated with the mention 228 243 try: ··· 328 343 bot_check_result = check_known_bots(unique_handles, void_agent) 329 344 bot_check_data = json.loads(bot_check_result) 330 345 331 - if bot_check_data.get("bot_detected", False): 346 + # TEMPORARILY DISABLED: Bot detection causing issues with normal users 347 + # TODO: Re-enable after debugging why normal users are being flagged as bots 348 + if False: # bot_check_data.get("bot_detected", False): 332 349 detected_bots = bot_check_data.get("detected_bots", []) 333 350 logger.info(f"Bot detected in thread: {detected_bots}") 334 351 ··· 340 357 else: 341 358 logger.info(f"Responding to bot thread (10% response rate). Detected bots: {detected_bots}") 342 359 else: 343 - logger.debug("No known bots detected in thread") 360 + logger.debug("Bot detection disabled - processing all notifications") 344 361 345 362 except Exception as bot_check_error: 346 363 logger.warning(f"Error checking for bots: {bot_check_error}") ··· 809 826 client=atproto_client, 810 827 notification=notification_data, 811 828 reply_text=cleaned_text, 812 - lang=reply_lang 829 + lang=reply_lang, 830 + correlation_id=correlation_id 813 831 ) 814 832 else: 815 833 # Multiple replies - use new threaded function ··· 819 837 client=atproto_client, 820 838 notification=notification_data, 821 839 reply_messages=cleaned_messages, 822 - lang=reply_lang 840 + lang=reply_lang, 841 + correlation_id=correlation_id 823 842 ) 824 843 825 844 if response: 826 - logger.info(f"Successfully replied to @{author_handle}") 845 + logger.info(f"[{correlation_id}] Successfully replied to @{author_handle}", extra={ 846 + 'correlation_id': correlation_id, 847 + 'author_handle': author_handle, 848 + 'reply_count': len(reply_messages) 849 + }) 827 850 828 851 # Acknowledge the post we're replying to with stream.thought.ack 829 852 try: ··· 857 880 else: 858 881 # Check if notification was explicitly ignored 859 882 if ignored_notification: 860 - logger.info(f"Notification from @{author_handle} was explicitly ignored (category: {ignore_category})") 883 + logger.info(f"[{correlation_id}] Notification from @{author_handle} was explicitly ignored (category: {ignore_category})", extra={ 884 + 'correlation_id': correlation_id, 885 + 'author_handle': author_handle, 886 + 'ignore_category': ignore_category 887 + }) 861 888 return "ignored" 862 889 else: 863 - logger.warning(f"No add_post_to_bluesky_reply_thread tool calls found for mention from @{author_handle}, moving to no_reply folder") 890 + logger.warning(f"[{correlation_id}] No reply generated for mention from @{author_handle}, moving to no_reply folder", extra={ 891 + 'correlation_id': correlation_id, 892 + 'author_handle': author_handle 893 + }) 864 894 return "no_reply" 865 895 866 896 except Exception as e: 867 - logger.error(f"Error processing mention: {e}") 897 + logger.error(f"[{correlation_id}] Error processing mention: {e}", extra={ 898 + 'correlation_id': correlation_id, 899 + 'error': str(e), 900 + 'error_type': type(e).__name__, 901 + 'author_handle': author_handle if 'author_handle' in locals() else 'unknown' 902 + }) 868 903 return False 869 904 finally: 870 905 # Detach user blocks after agent response (success or failure)

+188 -51

bsky_utils.py

··· 1 1 import os 2 2 import logging 3 + import uuid 4 + import time 3 5 from typing import Optional, Dict, Any, List 4 6 from atproto_client import Client, Session, SessionEvent, models 5 7 ··· 235 237 236 238 237 239 def default_login() -> Client: 238 - username = os.getenv("BSKY_USERNAME") 239 - password = os.getenv("BSKY_PASSWORD") 240 - 241 - if username is None: 242 - logger.error( 243 - "No username provided. Please provide a username using the BSKY_USERNAME environment variable." 244 - ) 245 - exit() 246 - 247 - if password is None: 248 - logger.error( 249 - "No password provided. Please provide a password using the BSKY_PASSWORD environment variable." 250 - ) 251 - exit() 252 - 253 - return init_client(username, password) 240 + """Login using configuration from config.yaml or environment variables.""" 241 + try: 242 + from config_loader import get_bluesky_config 243 + bluesky_config = get_bluesky_config() 244 + 245 + username = bluesky_config['username'] 246 + password = bluesky_config['password'] 247 + pds_uri = bluesky_config.get('pds_uri', 'https://bsky.social') 248 + 249 + logger.info(f"Logging into Bluesky as {username} via {pds_uri}") 250 + 251 + # Use pds_uri from config 252 + client = Client(base_url=pds_uri) 253 + client.login(username, password) 254 + return client 255 + 256 + except Exception as e: 257 + logger.error(f"Failed to load Bluesky configuration: {e}") 258 + logger.error("Please check your config.yaml file or environment variables") 259 + exit(1) 254 260 255 261 def remove_outside_quotes(text: str) -> str: 256 262 """ ··· 277 283 278 284 return text 279 285 280 - def reply_to_post(client: Client, text: str, reply_to_uri: str, reply_to_cid: str, root_uri: Optional[str] = None, root_cid: Optional[str] = None, lang: Optional[str] = None) -> Dict[str, Any]: 286 + def reply_to_post(client: Client, text: str, reply_to_uri: str, reply_to_cid: str, root_uri: Optional[str] = None, root_cid: Optional[str] = None, lang: Optional[str] = None, correlation_id: Optional[str] = None) -> Dict[str, Any]: 281 287 """ 282 288 Reply to a post on Bluesky with rich text support. 283 289 ··· 289 295 root_uri: The URI of the root post (if replying to a reply). If None, uses reply_to_uri 290 296 root_cid: The CID of the root post (if replying to a reply). If None, uses reply_to_cid 291 297 lang: Language code for the post (e.g., 'en-US', 'es', 'ja') 298 + correlation_id: Unique ID for tracking this message through the pipeline 292 299 293 300 Returns: 294 301 The response from sending the post 295 302 """ 296 303 import re 297 304 305 + # Generate correlation ID if not provided 306 + if correlation_id is None: 307 + correlation_id = str(uuid.uuid4())[:8] 308 + 309 + # Enhanced logging with structured data 310 + logger.info(f"[{correlation_id}] Starting reply_to_post", extra={ 311 + 'correlation_id': correlation_id, 312 + 'text_length': len(text), 313 + 'text_preview': text[:100] + '...' if len(text) > 100 else text, 314 + 'reply_to_uri': reply_to_uri, 315 + 'root_uri': root_uri, 316 + 'lang': lang 317 + }) 318 + 319 + start_time = time.time() 320 + 298 321 # If root is not provided, this is a reply to the root post 299 322 if root_uri is None: 300 323 root_uri = reply_to_uri ··· 307 330 # Parse rich text facets (mentions and URLs) 308 331 facets = [] 309 332 text_bytes = text.encode("UTF-8") 333 + mentions_found = [] 334 + urls_found = [] 335 + 336 + logger.debug(f"[{correlation_id}] Parsing facets from text (length: {len(text_bytes)} bytes)") 310 337 311 338 # Parse mentions - fixed to handle @ at start of text 312 339 mention_regex = rb"(?:^|[$|\W])(@([a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)" 313 340 314 341 for m in re.finditer(mention_regex, text_bytes): 315 342 handle = m.group(1)[1:].decode("UTF-8") # Remove @ prefix 343 + mentions_found.append(handle) 316 344 # Adjust byte positions to account for the optional prefix 317 345 mention_start = m.start(1) 318 346 mention_end = m.end(1) ··· 329 357 features=[models.AppBskyRichtextFacet.Mention(did=resolve_resp.did)] 330 358 ) 331 359 ) 360 + logger.debug(f"[{correlation_id}] Resolved mention @{handle} -> {resolve_resp.did}") 332 361 except Exception as e: 333 - logger.debug(f"Failed to resolve handle {handle}: {e}") 362 + logger.warning(f"[{correlation_id}] Failed to resolve handle @{handle}: {e}") 334 363 continue 335 364 336 365 # Parse URLs - fixed to handle URLs at start of text ··· 338 367 339 368 for m in re.finditer(url_regex, text_bytes): 340 369 url = m.group(1).decode("UTF-8") 370 + urls_found.append(url) 341 371 # Adjust byte positions to account for the optional prefix 342 372 url_start = m.start(1) 343 373 url_end = m.end(1) ··· 350 380 features=[models.AppBskyRichtextFacet.Link(uri=url)] 351 381 ) 352 382 ) 383 + logger.debug(f"[{correlation_id}] Found URL: {url}") 384 + 385 + logger.debug(f"[{correlation_id}] Facet parsing complete", extra={ 386 + 'correlation_id': correlation_id, 387 + 'mentions_count': len(mentions_found), 388 + 'mentions': mentions_found, 389 + 'urls_count': len(urls_found), 390 + 'urls': urls_found, 391 + 'total_facets': len(facets) 392 + }) 353 393 354 394 # Send the reply with facets if any were found 355 - if facets: 356 - response = client.send_post( 357 - text=text, 358 - reply_to=models.AppBskyFeedPost.ReplyRef(parent=parent_ref, root=root_ref), 359 - facets=facets, 360 - langs=[lang] if lang else None 361 - ) 362 - else: 363 - response = client.send_post( 364 - text=text, 365 - reply_to=models.AppBskyFeedPost.ReplyRef(parent=parent_ref, root=root_ref), 366 - langs=[lang] if lang else None 367 - ) 368 - 369 - logger.info(f"Reply sent successfully: {response.uri}") 370 - return response 395 + logger.info(f"[{correlation_id}] Sending reply to Bluesky API", extra={ 396 + 'correlation_id': correlation_id, 397 + 'has_facets': bool(facets), 398 + 'facet_count': len(facets), 399 + 'lang': lang 400 + }) 401 + 402 + try: 403 + if facets: 404 + response = client.send_post( 405 + text=text, 406 + reply_to=models.AppBskyFeedPost.ReplyRef(parent=parent_ref, root=root_ref), 407 + facets=facets, 408 + langs=[lang] if lang else None 409 + ) 410 + else: 411 + response = client.send_post( 412 + text=text, 413 + reply_to=models.AppBskyFeedPost.ReplyRef(parent=parent_ref, root=root_ref), 414 + langs=[lang] if lang else None 415 + ) 416 + 417 + # Calculate response time 418 + response_time = time.time() - start_time 419 + 420 + # Extract post URL for user-friendly logging 421 + post_url = None 422 + if hasattr(response, 'uri') and response.uri: 423 + # Convert AT-URI to web URL 424 + # Format: at://did:plc:xxx/app.bsky.feed.post/xxx -> https://bsky.app/profile/handle/post/xxx 425 + try: 426 + uri_parts = response.uri.split('/') 427 + if len(uri_parts) >= 4 and uri_parts[3] == 'app.bsky.feed.post': 428 + rkey = uri_parts[4] 429 + # We'd need to resolve DID to handle, but for now just use the URI 430 + post_url = f"bsky://post/{rkey}" 431 + except: 432 + pass 433 + 434 + logger.info(f"[{correlation_id}] Reply sent successfully ({response_time:.3f}s) - URI: {response.uri}" + 435 + (f" - URL: {post_url}" if post_url else ""), extra={ 436 + 'correlation_id': correlation_id, 437 + 'response_time': round(response_time, 3), 438 + 'post_uri': response.uri, 439 + 'post_url': post_url, 440 + 'post_cid': getattr(response, 'cid', None), 441 + 'text_length': len(text) 442 + }) 443 + 444 + return response 445 + 446 + except Exception as e: 447 + response_time = time.time() - start_time 448 + logger.error(f"[{correlation_id}] Failed to send reply", extra={ 449 + 'correlation_id': correlation_id, 450 + 'error': str(e), 451 + 'error_type': type(e).__name__, 452 + 'response_time': round(response_time, 3), 453 + 'text_length': len(text) 454 + }) 455 + raise 371 456 372 457 373 458 def get_post_thread(client: Client, uri: str) -> Optional[Dict[str, Any]]: ··· 389 474 return None 390 475 391 476 392 - def reply_to_notification(client: Client, notification: Any, reply_text: str, lang: str = "en-US") -> Optional[Dict[str, Any]]: 477 + def reply_to_notification(client: Client, notification: Any, reply_text: str, lang: str = "en-US", correlation_id: Optional[str] = None) -> Optional[Dict[str, Any]]: 393 478 """ 394 479 Reply to a notification (mention or reply). 395 480 ··· 398 483 notification: The notification object from list_notifications 399 484 reply_text: The text to reply with 400 485 lang: Language code for the post (defaults to "en-US") 486 + correlation_id: Unique ID for tracking this message through the pipeline 401 487 402 488 Returns: 403 489 The response from sending the reply or None if failed 404 490 """ 491 + # Generate correlation ID if not provided 492 + if correlation_id is None: 493 + correlation_id = str(uuid.uuid4())[:8] 494 + 495 + logger.info(f"[{correlation_id}] Processing reply_to_notification", extra={ 496 + 'correlation_id': correlation_id, 497 + 'reply_length': len(reply_text), 498 + 'lang': lang 499 + }) 500 + 405 501 try: 406 502 # Get the post URI and CID from the notification (handle both dict and object) 407 503 if isinstance(notification, dict): ··· 461 557 reply_to_cid=post_cid, 462 558 root_uri=root_uri, 463 559 root_cid=root_cid, 464 - lang=lang 560 + lang=lang, 561 + correlation_id=correlation_id 465 562 ) 466 563 467 564 except Exception as e: 468 - logger.error(f"Error replying to notification: {e}") 565 + logger.error(f"[{correlation_id}] Error replying to notification: {e}", extra={ 566 + 'correlation_id': correlation_id, 567 + 'error': str(e), 568 + 'error_type': type(e).__name__ 569 + }) 469 570 return None 470 571 471 572 472 - def reply_with_thread_to_notification(client: Client, notification: Any, reply_messages: List[str], lang: str = "en-US") -> Optional[List[Dict[str, Any]]]: 573 + def reply_with_thread_to_notification(client: Client, notification: Any, reply_messages: List[str], lang: str = "en-US", correlation_id: Optional[str] = None) -> Optional[List[Dict[str, Any]]]: 473 574 """ 474 575 Reply to a notification with a threaded chain of messages (max 15). 475 576 ··· 478 579 notification: The notification object from list_notifications 479 580 reply_messages: List of reply texts (max 15 messages, each max 300 chars) 480 581 lang: Language code for the posts (defaults to "en-US") 582 + correlation_id: Unique ID for tracking this message through the pipeline 481 583 482 584 Returns: 483 585 List of responses from sending the replies or None if failed 484 586 """ 587 + # Generate correlation ID if not provided 588 + if correlation_id is None: 589 + correlation_id = str(uuid.uuid4())[:8] 590 + 591 + logger.info(f"[{correlation_id}] Starting threaded reply", extra={ 592 + 'correlation_id': correlation_id, 593 + 'message_count': len(reply_messages), 594 + 'total_length': sum(len(msg) for msg in reply_messages), 595 + 'lang': lang 596 + }) 597 + 485 598 try: 486 599 # Validate input 487 600 if not reply_messages or len(reply_messages) == 0: 488 - logger.error("Reply messages list cannot be empty") 601 + logger.error(f"[{correlation_id}] Reply messages list cannot be empty") 489 602 return None 490 603 if len(reply_messages) > 15: 491 - logger.error(f"Cannot send more than 15 reply messages (got {len(reply_messages)})") 604 + logger.error(f"[{correlation_id}] Cannot send more than 15 reply messages (got {len(reply_messages)})") 492 605 return None 493 606 494 607 # Get the post URI and CID from the notification (handle both dict and object) ··· 547 660 current_parent_cid = post_cid 548 661 549 662 for i, message in enumerate(reply_messages): 550 - logger.info(f"Sending reply {i+1}/{len(reply_messages)}: {message[:50]}...") 663 + thread_correlation_id = f"{correlation_id}-{i+1}" 664 + logger.info(f"[{thread_correlation_id}] Sending reply {i+1}/{len(reply_messages)}: {message[:50]}...") 551 665 552 666 # Send this reply 553 667 response = reply_to_post( ··· 557 671 reply_to_cid=current_parent_cid, 558 672 root_uri=root_uri, 559 673 root_cid=root_cid, 560 - lang=lang 674 + lang=lang, 675 + correlation_id=thread_correlation_id 561 676 ) 562 677 563 678 if not response: 564 - logger.error(f"Failed to send reply {i+1}, posting system failure message") 679 + logger.error(f"[{thread_correlation_id}] Failed to send reply {i+1}, posting system failure message") 565 680 # Try to post a system failure message 566 681 failure_response = reply_to_post( 567 682 client=client, ··· 570 685 reply_to_cid=current_parent_cid, 571 686 root_uri=root_uri, 572 687 root_cid=root_cid, 573 - lang=lang 688 + lang=lang, 689 + correlation_id=f"{thread_correlation_id}-FAIL" 574 690 ) 575 691 if failure_response: 576 692 responses.append(failure_response) 577 693 current_parent_uri = failure_response.uri 578 694 current_parent_cid = failure_response.cid 579 695 else: 580 - logger.error("Could not even send system failure message, stopping thread") 696 + logger.error(f"[{thread_correlation_id}] Could not even send system failure message, stopping thread") 581 697 return responses if responses else None 582 698 else: 583 699 responses.append(response) ··· 586 702 current_parent_uri = response.uri 587 703 current_parent_cid = response.cid 588 704 589 - logger.info(f"Successfully sent {len(responses)} threaded replies") 705 + logger.info(f"[{correlation_id}] Successfully sent {len(responses)} threaded replies", extra={ 706 + 'correlation_id': correlation_id, 707 + 'replies_sent': len(responses), 708 + 'replies_requested': len(reply_messages) 709 + }) 590 710 return responses 591 711 592 712 except Exception as e: 593 - logger.error(f"Error sending threaded reply to notification: {e}") 713 + logger.error(f"[{correlation_id}] Error sending threaded reply to notification: {e}", extra={ 714 + 'correlation_id': correlation_id, 715 + 'error': str(e), 716 + 'error_type': type(e).__name__, 717 + 'message_count': len(reply_messages) 718 + }) 594 719 return None 595 720 596 721 ··· 631 756 logger.error("Missing access token or DID from session") 632 757 return None 633 758 634 - pds_host = os.getenv("PDS_URI", "https://bsky.social") 759 + # Get PDS URI from config instead of environment variables 760 + from config_loader import get_bluesky_config 761 + bluesky_config = get_bluesky_config() 762 + pds_host = bluesky_config['pds_uri'] 635 763 636 764 # Create acknowledgment record with null subject 637 765 now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") ··· 705 833 logger.error("Missing access token or DID from session") 706 834 return None 707 835 708 - pds_host = os.getenv("PDS_URI", "https://bsky.social") 836 + # Get PDS URI from config instead of environment variables 837 + from config_loader import get_bluesky_config 838 + bluesky_config = get_bluesky_config() 839 + pds_host = bluesky_config['pds_uri'] 709 840 710 841 # Create acknowledgment record with stream.thought.ack type 711 842 now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") ··· 781 912 logger.error("Missing access token or DID from session") 782 913 return None 783 914 784 - pds_host = os.getenv("PDS_URI", "https://bsky.social") 915 + # Get PDS URI from config instead of environment variables 916 + from config_loader import get_bluesky_config 917 + bluesky_config = get_bluesky_config() 918 + pds_host = bluesky_config['pds_uri'] 785 919 786 920 # Create tool call record 787 921 now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") ··· 858 992 logger.error("Missing access token or DID from session") 859 993 return None 860 994 861 - pds_host = os.getenv("PDS_URI", "https://bsky.social") 995 + # Get PDS URI from config instead of environment variables 996 + from config_loader import get_bluesky_config 997 + bluesky_config = get_bluesky_config() 998 + pds_host = bluesky_config['pds_uri'] 862 999 863 1000 # Create reasoning record 864 1001 now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")

+4 -4

config_loader.py

··· 179 179 } 180 180 181 181 def get_bluesky_config() -> Dict[str, Any]: 182 - """Get Bluesky configuration.""" 182 + """Get Bluesky configuration, prioritizing config.yaml over environment variables.""" 183 183 config = get_config() 184 184 return { 185 - 'username': config.get_required('bluesky.username', 'BSKY_USERNAME'), 186 - 'password': config.get_required('bluesky.password', 'BSKY_PASSWORD'), 187 - 'pds_uri': config.get_with_env('bluesky.pds_uri', 'PDS_URI', 'https://bsky.social'), 185 + 'username': config.get_required('bluesky.username'), 186 + 'password': config.get_required('bluesky.password'), 187 + 'pds_uri': config.get('bluesky.pds_uri', 'https://bsky.social'), 188 188 } 189 189 190 190 def get_bot_config() -> Dict[str, Any]:

Configure Feed

Configure Feed