Implement robust notification tracking system

Major improvements:
- Added SQLite database for reliable notification tracking
- Removed problematic is_read filter that was blocking all notifications
- Implemented timestamp-based filtering for continuity between runs
- Added comprehensive debug logging to understand notification flow
- Created recovery tools for missed notifications
- Added queue health monitoring and alerts
- Auto-migration from old JSON format to database
- Periodic cleanup of old notification records

This should significantly improve notification reliability and provide better visibility into what's being processed.

bsky.py

···
 import bsky_utils
 from tools.blocks import attach_user_blocks, detach_user_blocks
 from datetime import date
+from notification_db import NotificationDB
 
 def extract_handles_from_data(data):
     """Recursively extract all unique handles from nested data structure."""
···
 
 # Synthesis message tracking
 last_synthesis_time = time.time()
+
+# Database for notification tracking
+NOTIFICATION_DB = None
 
 def export_agent_state(client, agent, skip_git=False):
     """Export agent state to agent_archive/ (timestamped) and agents/ (current)."""
···
                         logger.info(f"Deleted queue file: {queue_filepath.name}")
                         
                         # Also mark as processed to avoid reprocessing
-                        processed_uris = load_processed_notifications()
-                        processed_uris.add(notification_data.get('uri', ''))
-                        save_processed_notifications(processed_uris)
+                        if NOTIFICATION_DB:
+                            NOTIFICATION_DB.mark_processed(notification_data.get('uri', ''), status='processed')
+                        else:
+                            processed_uris = load_processed_notifications()
+                            processed_uris.add(notification_data.get('uri', ''))
+                            save_processed_notifications(processed_uris)
                     
                     # Export agent state before terminating
                     export_agent_state(CLIENT, void_agent, skip_git=SKIP_GIT)
···
 
 
 def load_processed_notifications():
-    """Load the set of processed notification URIs."""
-    if PROCESSED_NOTIFICATIONS_FILE.exists():
-        try:
-            with open(PROCESSED_NOTIFICATIONS_FILE, 'r') as f:
-                data = json.load(f)
-                # Keep only recent entries (last MAX_PROCESSED_NOTIFICATIONS)
-                if len(data) > MAX_PROCESSED_NOTIFICATIONS:
-                    data = data[-MAX_PROCESSED_NOTIFICATIONS:]
-                    save_processed_notifications(data)
-                return set(data)
-        except Exception as e:
-            logger.error(f"Error loading processed notifications: {e}")
+    """Load the set of processed notification URIs from database."""
+    global NOTIFICATION_DB
+    if NOTIFICATION_DB:
+        return NOTIFICATION_DB.get_processed_uris(limit=MAX_PROCESSED_NOTIFICATIONS)
     return set()
 
 
 def save_processed_notifications(processed_set):
-    """Save the set of processed notification URIs."""
-    try:
-        with open(PROCESSED_NOTIFICATIONS_FILE, 'w') as f:
-            json.dump(list(processed_set), f)
-    except Exception as e:
-        logger.error(f"Error saving processed notifications: {e}")
+    """Save the set of processed notification URIs to database."""
+    # This is now handled by marking individual notifications in the DB
+    # Keeping function for compatibility but it doesn't need to do anything
+    pass
 
 
 def save_notification_to_queue(notification, is_priority=None):
     """Save a notification to the queue directory with priority-based filename."""
     try:
-        # Check if already processed
-        processed_uris = load_processed_notifications()
+        global NOTIFICATION_DB
         
         # Handle both notification objects and dicts
         if isinstance(notification, dict):
···
         else:
             notif_dict = notification_to_dict(notification)
             notification_uri = notification.uri
-            
-        if notification_uri in processed_uris:
-            logger.debug(f"Notification already processed: {notification_uri}")
-            return False
+        
+        # Check if already processed (using database if available)
+        if NOTIFICATION_DB:
+            if NOTIFICATION_DB.is_processed(notification_uri):
+                logger.debug(f"Notification already processed (DB): {notification_uri}")
+                return False
+            # Add to database
+            NOTIFICATION_DB.add_notification(notif_dict)
+        else:
+            # Fall back to old JSON method
+            processed_uris = load_processed_notifications()
+            if notification_uri in processed_uris:
+                logger.debug(f"Notification already processed: {notification_uri}")
+                return False
 
         # Create JSON string
         notif_json = json.dumps(notif_dict, sort_keys=True)
···
                         logger.info(f"Successfully processed and removed: {filepath.name}")
                         
                         # Mark as processed to avoid reprocessing
-                        processed_uris = load_processed_notifications()
-                        processed_uris.add(notif_data['uri'])
-                        save_processed_notifications(processed_uris)
+                        if NOTIFICATION_DB:
+                            NOTIFICATION_DB.mark_processed(notif_data['uri'], status='processed')
+                        else:
+                            processed_uris = load_processed_notifications()
+                            processed_uris.add(notif_data['uri'])
+                            save_processed_notifications(processed_uris)
                     
                 elif success is None:  # Special case for moving to error directory
                     error_path = QUEUE_ERROR_DIR / filepath.name
···
                     logger.warning(f"Moved {filepath.name} to errors directory")
                     
                     # Also mark as processed to avoid retrying
-                    processed_uris = load_processed_notifications()
-                    processed_uris.add(notif_data['uri'])
-                    save_processed_notifications(processed_uris)
+                    if NOTIFICATION_DB:
+                        NOTIFICATION_DB.mark_processed(notif_data['uri'], status='error')
+                    else:
+                        processed_uris = load_processed_notifications()
+                        processed_uris.add(notif_data['uri'])
+                        save_processed_notifications(processed_uris)
                     
                 elif success == "no_reply":  # Special case for moving to no_reply directory
                     no_reply_path = QUEUE_NO_REPLY_DIR / filepath.name
···
                     logger.info(f"Moved {filepath.name} to no_reply directory")
                     
                     # Also mark as processed to avoid retrying
-                    processed_uris = load_processed_notifications()
-                    processed_uris.add(notif_data['uri'])
-                    save_processed_notifications(processed_uris)
+                    if NOTIFICATION_DB:
+                        NOTIFICATION_DB.mark_processed(notif_data['uri'], status='error')
+                    else:
+                        processed_uris = load_processed_notifications()
+                        processed_uris.add(notif_data['uri'])
+                        save_processed_notifications(processed_uris)
                     
                 elif success == "ignored":  # Special case for explicitly ignored notifications
                     # For ignored notifications, we just delete them (not move to no_reply)
···
                     logger.info(f"🚫 Deleted ignored notification: {filepath.name}")
                     
                     # Also mark as processed to avoid retrying
-                    processed_uris = load_processed_notifications()
-                    processed_uris.add(notif_data['uri'])
-                    save_processed_notifications(processed_uris)
+                    if NOTIFICATION_DB:
+                        NOTIFICATION_DB.mark_processed(notif_data['uri'], status='error')
+                    else:
+                        processed_uris = load_processed_notifications()
+                        processed_uris.add(notif_data['uri'])
+                        save_processed_notifications(processed_uris)
                     
                 else:
                     logger.warning(f"⚠️  Failed to process {filepath.name}, keeping in queue for retry")
···
 def fetch_and_queue_new_notifications(atproto_client):
     """Fetch new notifications and queue them without processing."""
     try:
+        global NOTIFICATION_DB
+        
         # Get current time for marking notifications as seen
         logger.debug("Getting current time for notification marking...")
         last_seen_at = atproto_client.get_current_time_iso()
+        
+        # Get timestamp of last processed notification for filtering
+        last_processed_time = None
+        if NOTIFICATION_DB:
+            last_processed_time = NOTIFICATION_DB.get_latest_processed_time()
+            if last_processed_time:
+                logger.debug(f"Last processed notification was at: {last_processed_time}")
 
         # Fetch ALL notifications using pagination
         all_notifications = []
···
         # Now process all fetched notifications
         new_count = 0
         if all_notifications:
+            logger.info(f"📥 Fetched {len(all_notifications)} total notifications from API")
+            
             # Mark as seen first
             try:
                 atproto_client.app.bsky.notification.update_seen(
···
             except Exception as e:
                 logger.error(f"Error marking notifications as seen: {e}")
             
-            # Queue all new notifications (except likes and already read)
+            # Debug counters
+            skipped_read = 0
+            skipped_likes = 0
+            skipped_processed = 0
+            skipped_old_timestamp = 0
+            processed_uris = load_processed_notifications()
+            
+            # Queue all new notifications (except likes)
             for notif in all_notifications:
-                # Skip if already read or if it's a like
-                if (hasattr(notif, 'is_read') and notif.is_read) or (hasattr(notif, 'reason') and notif.reason == 'like'):
+                # Skip if older than last processed (when we have timestamp filtering)
+                if last_processed_time and hasattr(notif, 'indexed_at'):
+                    if notif.indexed_at <= last_processed_time:
+                        skipped_old_timestamp += 1
+                        logger.debug(f"Skipping old notification (indexed_at {notif.indexed_at} <= {last_processed_time})")
+                        continue
+                
+                # Debug: Log is_read status but DON'T skip based on it
+                if hasattr(notif, 'is_read') and notif.is_read:
+                    skipped_read += 1
+                    logger.debug(f"Notification has is_read=True (but processing anyway): {notif.uri if hasattr(notif, 'uri') else 'unknown'}")
+                
+                # Skip likes
+                if hasattr(notif, 'reason') and notif.reason == 'like':
+                    skipped_likes += 1
                     continue
                     
                 notif_dict = notif.model_dump() if hasattr(notif, 'model_dump') else notif
···
                 if notif_dict.get('reason') == 'like':
                     continue
                 
+                # Check if already processed
+                notif_uri = notif_dict.get('uri', '')
+                if notif_uri in processed_uris:
+                    skipped_processed += 1
+                    logger.debug(f"Skipping already processed: {notif_uri}")
+                    continue
+                
                 # Check if it's a priority notification
                 is_priority = False
                 
···
                 
                 if save_notification_to_queue(notif_dict, is_priority=is_priority):
                     new_count += 1
+                    logger.debug(f"Queued notification from @{author_handle}: {notif_dict.get('reason', 'unknown')}")
             
-            logger.info(f"Queued {new_count} new notifications and marked as seen")
+            # Log summary of filtering
+            logger.info(f"📊 Notification processing summary:")
+            logger.info(f"  • Total fetched: {len(all_notifications)}")
+            logger.info(f"  • Had is_read=True: {skipped_read} (not skipped)")
+            logger.info(f"  • Skipped (likes): {skipped_likes}")
+            logger.info(f"  • Skipped (old timestamp): {skipped_old_timestamp}")
+            logger.info(f"  • Skipped (already processed): {skipped_processed}")
+            logger.info(f"  • Queued for processing: {new_count}")
         else:
             logger.debug("No new notifications to queue")
             
···
     void_agent = initialize_void()
     logger.info(f"Void agent initialized: {void_agent.id}")
     
+    # Initialize notification database
+    global NOTIFICATION_DB
+    logger.info("Initializing notification database...")
+    NOTIFICATION_DB = NotificationDB()
+    
+    # Migrate from old JSON format if it exists
+    if PROCESSED_NOTIFICATIONS_FILE.exists():
+        logger.info("Found old processed_notifications.json, migrating to database...")
+        NOTIFICATION_DB.migrate_from_json(str(PROCESSED_NOTIFICATIONS_FILE))
+    
+    # Log database stats
+    db_stats = NOTIFICATION_DB.get_stats()
+    logger.info(f"Database initialized - Total notifications: {db_stats.get('total', 0)}, Recent (24h): {db_stats.get('recent_24h', 0)}")
+    
+    # Clean up old records
+    NOTIFICATION_DB.cleanup_old_records(days=7)
+    
     # Ensure correct tools are attached for Bluesky
     logger.info("Configuring tools for Bluesky platform...")
     try:
···
             if CLEANUP_INTERVAL > 0 and cycle_count % CLEANUP_INTERVAL == 0:
                 logger.debug(f"Running periodic user block cleanup (cycle {cycle_count})")
                 periodic_user_block_cleanup(CLIENT, void_agent.id)
+                
+                # Also check database health when doing cleanup
+                if NOTIFICATION_DB:
+                    db_stats = NOTIFICATION_DB.get_stats()
+                    pending = db_stats.get('status_pending', 0)
+                    errors = db_stats.get('status_error', 0)
+                    
+                    if pending > 50:
+                        logger.warning(f"⚠️ Queue health check: {pending} pending notifications (may be stuck)")
+                    if errors > 20:
+                        logger.warning(f"⚠️ Queue health check: {errors} error notifications")
+                    
+                    # Periodic cleanup of old records
+                    if cycle_count % (CLEANUP_INTERVAL * 10) == 0:  # Every 100 cycles
+                        logger.info("Running database cleanup of old records...")
+                        NOTIFICATION_DB.cleanup_old_records(days=7)
             
             # Log cycle completion with stats
             elapsed_time = time.time() - start_time
···
             logger.info(f"   - {message_counters['follows']} follows")
             logger.info(f"   - {message_counters['reposts_skipped']} reposts skipped")
             logger.info(f"   - Average rate: {messages_per_minute:.1f} messages/minute")
+            
+            # Close database connection
+            if NOTIFICATION_DB:
+                logger.info("Closing database connection...")
+                NOTIFICATION_DB.close()
+            
             break
         except Exception as e:
             logger.error(f"=== ERROR IN MAIN LOOP CYCLE {cycle_count} ===")

notification_db.py

···
+#!/usr/bin/env python3
+"""SQLite database for robust notification tracking."""
+
+import sqlite3
+import json
+from pathlib import Path
+from datetime import datetime, timedelta
+from typing import Set, Dict, List, Optional, Tuple
+import logging
+
+logger = logging.getLogger(__name__)
+
+class NotificationDB:
+    """Database for tracking notification processing state."""
+    
+    def __init__(self, db_path: str = "queue/notifications.db"):
+        """Initialize the notification database."""
+        self.db_path = Path(db_path)
+        self.db_path.parent.mkdir(exist_ok=True, parents=True)
+        self.conn = None
+        self._init_db()
+    
+    def _init_db(self):
+        """Initialize database schema."""
+        self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
+        self.conn.row_factory = sqlite3.Row
+        
+        # Create main notifications table
+        self.conn.execute("""
+            CREATE TABLE IF NOT EXISTS notifications (
+                uri TEXT PRIMARY KEY,
+                indexed_at TEXT NOT NULL,
+                processed_at TEXT,
+                status TEXT NOT NULL DEFAULT 'pending',
+                reason TEXT,
+                author_handle TEXT,
+                author_did TEXT,
+                text TEXT,
+                parent_uri TEXT,
+                root_uri TEXT,
+                error TEXT,
+                metadata TEXT
+            )
+        """)
+        
+        # Create indexes for faster lookups
+        self.conn.execute("""
+            CREATE INDEX IF NOT EXISTS idx_indexed_at 
+            ON notifications(indexed_at DESC)
+        """)
+        
+        self.conn.execute("""
+            CREATE INDEX IF NOT EXISTS idx_status 
+            ON notifications(status)
+        """)
+        
+        self.conn.execute("""
+            CREATE INDEX IF NOT EXISTS idx_author_handle 
+            ON notifications(author_handle)
+        """)
+        
+        # Create session tracking table
+        self.conn.execute("""
+            CREATE TABLE IF NOT EXISTS sessions (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                started_at TEXT NOT NULL,
+                ended_at TEXT,
+                last_seen_at TEXT,
+                notifications_processed INTEGER DEFAULT 0,
+                notifications_skipped INTEGER DEFAULT 0,
+                notifications_error INTEGER DEFAULT 0
+            )
+        """)
+        
+        self.conn.commit()
+    
+    def add_notification(self, notif_dict: Dict) -> bool:
+        """Add a notification to the database."""
+        try:
+            # Handle None input
+            if not notif_dict:
+                return False
+                
+            # Extract key fields
+            uri = notif_dict.get('uri', '')
+            if not uri:
+                return False
+            
+            indexed_at = notif_dict.get('indexed_at', '')
+            reason = notif_dict.get('reason', '')
+            author = notif_dict.get('author', {}) if notif_dict.get('author') else {}
+            author_handle = author.get('handle', '') if author else ''
+            author_did = author.get('did', '') if author else ''
+            
+            # Extract text from record if available
+            record = notif_dict.get('record', {})
+            text = record.get('text', '')[:500]  # Limit text length
+            
+            # Extract thread info
+            parent_uri = None
+            root_uri = None
+            if 'reply' in record:
+                parent_uri = record['reply'].get('parent', {}).get('uri')
+                root_uri = record['reply'].get('root', {}).get('uri')
+            
+            # Store additional metadata as JSON
+            metadata = {
+                'cid': notif_dict.get('cid'),
+                'labels': notif_dict.get('labels', []),
+                'is_read': notif_dict.get('is_read', False)
+            }
+            
+            self.conn.execute("""
+                INSERT OR IGNORE INTO notifications 
+                (uri, indexed_at, reason, author_handle, author_did, text, 
+                 parent_uri, root_uri, status, metadata)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?)
+            """, (uri, indexed_at, reason, author_handle, author_did, text,
+                  parent_uri, root_uri, json.dumps(metadata)))
+            
+            self.conn.commit()
+            return True
+        
+        except Exception as e:
+            logger.error(f"Error adding notification to DB: {e}")
+            return False
+    
+    def is_processed(self, uri: str) -> bool:
+        """Check if a notification has been processed."""
+        cursor = self.conn.execute("""
+            SELECT status FROM notifications WHERE uri = ?
+        """, (uri,))
+        row = cursor.fetchone()
+        
+        if row:
+            return row['status'] in ['processed', 'ignored', 'no_reply']
+        return False
+    
+    def mark_processed(self, uri: str, status: str = 'processed', error: str = None):
+        """Mark a notification as processed."""
+        try:
+            self.conn.execute("""
+                UPDATE notifications 
+                SET status = ?, processed_at = ?, error = ?
+                WHERE uri = ?
+            """, (status, datetime.now().isoformat(), error, uri))
+            self.conn.commit()
+        except Exception as e:
+            logger.error(f"Error marking notification processed: {e}")
+    
+    def get_unprocessed(self, limit: int = 100) -> List[Dict]:
+        """Get unprocessed notifications."""
+        cursor = self.conn.execute("""
+            SELECT * FROM notifications 
+            WHERE status = 'pending'
+            ORDER BY indexed_at ASC
+            LIMIT ?
+        """, (limit,))
+        
+        return [dict(row) for row in cursor]
+    
+    def get_latest_processed_time(self) -> Optional[str]:
+        """Get the timestamp of the most recently processed notification."""
+        cursor = self.conn.execute("""
+            SELECT MAX(indexed_at) as latest 
+            FROM notifications 
+            WHERE status IN ('processed', 'ignored', 'no_reply')
+        """)
+        row = cursor.fetchone()
+        return row['latest'] if row and row['latest'] else None
+    
+    def cleanup_old_records(self, days: int = 7):
+        """Remove records older than specified days."""
+        cutoff_date = (datetime.now() - timedelta(days=days)).isoformat()
+        
+        deleted = self.conn.execute("""
+            DELETE FROM notifications 
+            WHERE indexed_at < ? 
+            AND status IN ('processed', 'ignored', 'no_reply', 'error')
+        """, (cutoff_date,)).rowcount
+        
+        self.conn.commit()
+        
+        if deleted > 0:
+            logger.info(f"Cleaned up {deleted} old notification records")
+            # Vacuum to reclaim space
+            self.conn.execute("VACUUM")
+    
+    def get_stats(self) -> Dict:
+        """Get database statistics."""
+        stats = {}
+        
+        # Count by status
+        cursor = self.conn.execute("""
+            SELECT status, COUNT(*) as count 
+            FROM notifications 
+            GROUP BY status
+        """)
+        
+        for row in cursor:
+            stats[f"status_{row['status']}"] = row['count']
+        
+        # Total count
+        cursor = self.conn.execute("SELECT COUNT(*) as total FROM notifications")
+        stats['total'] = cursor.fetchone()['total']
+        
+        # Recent activity (last 24h)
+        yesterday = (datetime.now() - timedelta(days=1)).isoformat()
+        cursor = self.conn.execute("""
+            SELECT COUNT(*) as recent 
+            FROM notifications 
+            WHERE indexed_at > ?
+        """, (yesterday,))
+        stats['recent_24h'] = cursor.fetchone()['recent']
+        
+        return stats
+    
+    def start_session(self) -> int:
+        """Start a new processing session."""
+        cursor = self.conn.execute("""
+            INSERT INTO sessions (started_at, last_seen_at)
+            VALUES (?, ?)
+        """, (datetime.now().isoformat(), datetime.now().isoformat()))
+        self.conn.commit()
+        return cursor.lastrowid
+    
+    def update_session(self, session_id: int, processed: int = 0, skipped: int = 0, error: int = 0):
+        """Update session statistics."""
+        self.conn.execute("""
+            UPDATE sessions 
+            SET last_seen_at = ?,
+                notifications_processed = notifications_processed + ?,
+                notifications_skipped = notifications_skipped + ?,
+                notifications_error = notifications_error + ?
+            WHERE id = ?
+        """, (datetime.now().isoformat(), processed, skipped, error, session_id))
+        self.conn.commit()
+    
+    def end_session(self, session_id: int):
+        """End a processing session."""
+        self.conn.execute("""
+            UPDATE sessions 
+            SET ended_at = ?
+            WHERE id = ?
+        """, (datetime.now().isoformat(), session_id))
+        self.conn.commit()
+    
+    def get_processed_uris(self, limit: int = 10000) -> Set[str]:
+        """Get set of processed URIs for compatibility with existing code."""
+        cursor = self.conn.execute("""
+            SELECT uri FROM notifications 
+            WHERE status IN ('processed', 'ignored', 'no_reply')
+            ORDER BY processed_at DESC
+            LIMIT ?
+        """, (limit,))
+        
+        return {row['uri'] for row in cursor}
+    
+    def migrate_from_json(self, json_path: str = "queue/processed_notifications.json"):
+        """Migrate data from the old JSON format."""
+        json_file = Path(json_path)
+        if not json_file.exists():
+            return
+        
+        try:
+            with open(json_file, 'r') as f:
+                uris = json.load(f)
+            
+            migrated = 0
+            for uri in uris:
+                # Add as processed with unknown timestamp
+                self.conn.execute("""
+                    INSERT OR IGNORE INTO notifications 
+                    (uri, indexed_at, status, processed_at)
+                    VALUES (?, ?, 'processed', ?)
+                """, (uri, datetime.now().isoformat(), datetime.now().isoformat()))
+                migrated += 1
+            
+            self.conn.commit()
+            logger.info(f"Migrated {migrated} URIs from JSON to database")
+            
+            # Rename old file to backup
+            backup_path = json_file.with_suffix('.json.backup')
+            json_file.rename(backup_path)
+            logger.info(f"Renamed old JSON file to {backup_path}")
+            
+        except Exception as e:
+            logger.error(f"Error migrating from JSON: {e}")
+    
+    def close(self):
+        """Close database connection."""
+        if self.conn:
+            self.conn.close()

notification_recovery.py

···
+#!/usr/bin/env python3
+"""Recovery tools for missed notifications."""
+
+import argparse
+import logging
+from datetime import datetime, timedelta
+from pathlib import Path
+import json
+import bsky_utils
+from notification_db import NotificationDB
+from bsky import save_notification_to_queue, notification_to_dict
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+
+def recover_notifications(hours=24, dry_run=True):
+    """
+    Recover notifications from the past N hours.
+    
+    Args:
+        hours: Number of hours back to check for notifications
+        dry_run: If True, only show what would be recovered without saving
+    """
+    logger.info(f"Recovering notifications from the past {hours} hours")
+    logger.info(f"Dry run: {dry_run}")
+    
+    # Initialize Bluesky client
+    client = bsky_utils.default_login()
+    logger.info("Connected to Bluesky")
+    
+    # Initialize database
+    db = NotificationDB()
+    logger.info("Database initialized")
+    
+    # Fetch notifications
+    all_notifications = []
+    cursor = None
+    page_count = 0
+    max_pages = 50  # More pages for recovery
+    
+    cutoff_time = datetime.now() - timedelta(hours=hours)
+    cutoff_iso = cutoff_time.isoformat()
+    logger.info(f"Looking for notifications since: {cutoff_iso}")
+    
+    while page_count < max_pages:
+        try:
+            # Fetch notifications page
+            if cursor:
+                response = client.app.bsky.notification.list_notifications(
+                    params={'cursor': cursor, 'limit': 100}
+                )
+            else:
+                response = client.app.bsky.notification.list_notifications(
+                    params={'limit': 100}
+                )
+            
+            page_count += 1
+            page_notifications = response.notifications
+            
+            if not page_notifications:
+                break
+            
+            # Filter by time
+            for notif in page_notifications:
+                if hasattr(notif, 'indexed_at') and notif.indexed_at >= cutoff_iso:
+                    all_notifications.append(notif)
+                elif hasattr(notif, 'indexed_at') and notif.indexed_at < cutoff_iso:
+                    # We've gone past our cutoff, stop fetching
+                    logger.info(f"Reached notifications older than {hours} hours, stopping")
+                    cursor = None
+                    break
+            
+            # Check if there are more pages
+            if cursor is None:
+                break
+            cursor = getattr(response, 'cursor', None)
+            if not cursor:
+                break
+                
+        except Exception as e:
+            logger.error(f"Error fetching notifications page {page_count}: {e}")
+            break
+    
+    logger.info(f"Found {len(all_notifications)} notifications in the time range")
+    
+    # Process notifications
+    recovered = 0
+    skipped_likes = 0
+    already_processed = 0
+    
+    for notif in all_notifications:
+        # Skip likes
+        if hasattr(notif, 'reason') and notif.reason == 'like':
+            skipped_likes += 1
+            continue
+        
+        # Check if already processed
+        notif_dict = notification_to_dict(notif)
+        uri = notif_dict.get('uri', '')
+        
+        if db.is_processed(uri):
+            already_processed += 1
+            logger.debug(f"Already processed: {uri}")
+            continue
+        
+        # Log what we would recover
+        author = notif_dict.get('author', {})
+        author_handle = author.get('handle', 'unknown')
+        reason = notif_dict.get('reason', 'unknown')
+        indexed_at = notif_dict.get('indexed_at', '')
+        
+        logger.info(f"Would recover: {reason} from @{author_handle} at {indexed_at}")
+        
+        if not dry_run:
+            # Save to queue
+            if save_notification_to_queue(notif_dict):
+                recovered += 1
+                logger.info(f"Recovered notification from @{author_handle}")
+            else:
+                logger.warning(f"Failed to queue notification from @{author_handle}")
+        else:
+            recovered += 1
+    
+    # Summary
+    logger.info(f"Recovery summary:")
+    logger.info(f"  • Total found: {len(all_notifications)}")
+    logger.info(f"  • Skipped (likes): {skipped_likes}")
+    logger.info(f"  • Already processed: {already_processed}")
+    logger.info(f"  • {'Would recover' if dry_run else 'Recovered'}: {recovered}")
+    
+    if dry_run and recovered > 0:
+        logger.info("Run with --execute to actually recover these notifications")
+    
+    return recovered
+
+
+def check_database_health():
+    """Check the health of the notification database."""
+    db = NotificationDB()
+    stats = db.get_stats()
+    
+    logger.info("Database Statistics:")
+    logger.info(f"  • Total notifications: {stats.get('total', 0)}")
+    logger.info(f"  • Pending: {stats.get('status_pending', 0)}")
+    logger.info(f"  • Processed: {stats.get('status_processed', 0)}")
+    logger.info(f"  • Ignored: {stats.get('status_ignored', 0)}")
+    logger.info(f"  • No reply: {stats.get('status_no_reply', 0)}")
+    logger.info(f"  • Errors: {stats.get('status_error', 0)}")
+    logger.info(f"  • Recent (24h): {stats.get('recent_24h', 0)}")
+    
+    # Check for issues
+    if stats.get('status_pending', 0) > 100:
+        logger.warning(f"⚠️ High number of pending notifications: {stats.get('status_pending', 0)}")
+    
+    if stats.get('status_error', 0) > 50:
+        logger.warning(f"⚠️ High number of error notifications: {stats.get('status_error', 0)}")
+    
+    return stats
+
+
+def reset_notification_status(hours=1, dry_run=True):
+    """
+    Reset notifications from error/no_reply status back to pending.
+    
+    Args:
+        hours: Reset notifications from the last N hours
+        dry_run: If True, only show what would be reset
+    """
+    db = NotificationDB()
+    cutoff_time = datetime.now() - timedelta(hours=hours)
+    cutoff_iso = cutoff_time.isoformat()
+    
+    # Get notifications to reset
+    cursor = db.conn.execute("""
+        SELECT uri, status, indexed_at, author_handle 
+        FROM notifications 
+        WHERE status IN ('error', 'no_reply')
+        AND indexed_at > ?
+        ORDER BY indexed_at DESC
+    """, (cutoff_iso,))
+    
+    notifications_to_reset = cursor.fetchall()
+    
+    if not notifications_to_reset:
+        logger.info(f"No notifications to reset from the last {hours} hours")
+        return 0
+    
+    logger.info(f"Found {len(notifications_to_reset)} notifications to reset")
+    
+    for notif in notifications_to_reset:
+        logger.info(f"Would reset: {notif['status']} -> pending for @{notif['author_handle']} at {notif['indexed_at']}")
+    
+    if not dry_run:
+        reset_count = db.conn.execute("""
+            UPDATE notifications 
+            SET status = 'pending', processed_at = NULL, error = NULL
+            WHERE status IN ('error', 'no_reply')
+            AND indexed_at > ?
+        """, (cutoff_iso,)).rowcount
+        
+        db.conn.commit()
+        logger.info(f"Reset {reset_count} notifications to pending status")
+        return reset_count
+    else:
+        logger.info("Run with --execute to actually reset these notifications")
+        return len(notifications_to_reset)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Notification recovery and management tools")
+    
+    subparsers = parser.add_subparsers(dest='command', help='Command to run')
+    
+    # Recover command
+    recover_parser = subparsers.add_parser('recover', help='Recover missed notifications')
+    recover_parser.add_argument('--hours', type=int, default=24, 
+                              help='Number of hours back to check (default: 24)')
+    recover_parser.add_argument('--execute', action='store_true',
+                              help='Actually recover notifications (default is dry run)')
+    
+    # Health check command
+    health_parser = subparsers.add_parser('health', help='Check database health')
+    
+    # Reset command
+    reset_parser = subparsers.add_parser('reset', help='Reset error notifications to pending')
+    reset_parser.add_argument('--hours', type=int, default=1,
+                            help='Reset notifications from last N hours (default: 1)')
+    reset_parser.add_argument('--execute', action='store_true',
+                            help='Actually reset notifications (default is dry run)')
+    
+    args = parser.parse_args()
+    
+    if args.command == 'recover':
+        recover_notifications(hours=args.hours, dry_run=not args.execute)
+    elif args.command == 'health':
+        check_database_health()
+    elif args.command == 'reset':
+        reset_notification_status(hours=args.hours, dry_run=not args.execute)
+    else:
+        parser.print_help()
+
+
+if __name__ == "__main__":
+    main()