update to v2

entrypoint.sh

@@ -1,8 +1,29 @@
 #!/bin/bash
-# Webapp entrypoint script - runs database migrations before starting Gunicorn
+# Webapp entrypoint script - restores from S3 and runs database migrations before starting Gunicorn
 
 set -e
 
+echo "[ENTRYPOINT] Checking for S3 restore..."
+
+# Restore database from S3 if enabled (Feature 015)
+python3 <<'RESTORE_PYEOF'
+import os
+import sys
+
+# Add src directory to Python path
+sys.path.insert(0, '/app')
+
+try:
+    from src.services.s3_restore import restore_on_startup
+    
+    result = restore_on_startup()
+    print(f"[S3_RESTORE] Result: {result.value}")
+except Exception as e:
+    print(f"[S3_RESTORE] Failed (non-fatal): {e}")
+    # Continue startup even if restore fails
+
+RESTORE_PYEOF
+
 echo "[ENTRYPOINT] Running database migrations..."
 
 # Run Python migration script
@@ -42,6 +63,35 @@ if not table_exists:
 cursor.execute("PRAGMA table_info(contact_sessions)")
 columns = [row[1] for row in cursor.fetchall()]
 
+# Step 2a: Add contact_id column if missing (v2 API alignment)
+if 'contact_id' not in columns:
+    print("[MIGRATION] Applying migration: Add contact_id column")
+    
+    try:
+        cursor.execute("ALTER TABLE contact_sessions ADD COLUMN contact_id VARCHAR(255) NOT NULL DEFAULT ''")
+        print("  ✓ Added contact_id column")
+    except Exception as e:
+        print(f"  ⚠ contact_id: {e}")
+    
+    try:
+        cursor.execute("UPDATE contact_sessions SET contact_id = session_id WHERE contact_id = ''")
+        print(f"  ✓ Backfilled {cursor.rowcount} existing contact_id values")
+    except Exception as e:
+        print(f"  ⚠ backfill contact_id: {e}")
+    
+    try:
+        cursor.execute("CREATE INDEX IF NOT EXISTS idx_contact_sessions_contact_id ON contact_sessions(contact_id)")
+        print("  ✓ Created index idx_contact_sessions_contact_id")
+    except Exception as e:
+        print(f"  ⚠ index: {e}")
+    
+    conn.commit()
+    print("[MIGRATION] contact_id migration complete")
+
+# Refresh columns list after contact_id migration
+cursor.execute("PRAGMA table_info(contact_sessions)")
+columns = [row[1] for row in cursor.fetchall()]
+
 if 'normalized_name' not in columns:
     print("[MIGRATION] Applying migration: Add producer-related columns")
     

migrations/003_add_contact_id.sql

@@ -0,0 +1,12 @@
+-- Migration: Add contact_id to contact_sessions table
+-- This aligns with v2 API where contact_id maps to project_id in backend
+
+-- Add contact_id column (default to session_id for existing records)
+ALTER TABLE contact_sessions ADD COLUMN contact_id VARCHAR(255) NOT NULL DEFAULT '';
+
+-- Update existing records: set contact_id from session_id
+-- Format is typically user_id_uuid, we'll use the full session_id as contact_id
+UPDATE contact_sessions SET contact_id = session_id WHERE contact_id = '';
+
+-- Create index for contact_id lookups
+CREATE INDEX IF NOT EXISTS idx_contact_sessions_contact_id ON contact_sessions(contact_id);

requirements.txt

@@ -18,6 +18,9 @@ opentelemetry-instrumentation-requests==0.42b0
 # Database
 SQLAlchemy==2.0.23
 
+# S3 backup/restore
+boto3>=1.34.0
+
 # Configuration
 python-dotenv==1.0.0
 

src/models/__init__.py

@@ -26,6 +26,7 @@ class ContactSession:
 
     session_id: str  # Format: {user_id}_{UUID_v4}
     user_id: str
+    contact_id: str  # Contact identifier (maps to project_id in v2 backend)
     contact_name: str
     contact_description: Optional[str]
     is_reference: bool

src/routes/auth.py

@@ -108,8 +108,10 @@ def callback():
             # New user - create backend session first, then store with returned session_id
             try:
                 backend_response = backend_client.create_session(
-                    title=f"{display_name}'s Profile",
-                    user_id=user_id
+                    user_id=user_id,
+                    contact_id="user-profile",
+                    description=f"{display_name}'s Profile",
+                    is_reference=False
                 )
                 backend_session_id = backend_response.get("session_id")
                 

src/routes/contacts.py

@@ -137,9 +137,13 @@ def create_contact():
     
     try:
         # Phase 1: Create backend session in PostgreSQL
-        title = f"Contact: {contact_name}"
         logger.info(f"[CONTACT_CREATE] Phase 1: Creating backend session for user {user_id}, contact '{contact_name}'")
-        backend_response = backend_api.create_session(title=title, user_id=user_id)
+        backend_response = backend_api.create_session(
+            user_id=user_id,
+            contact_id=contact_name,  # Use contact name as contact_id initially
+            description=f"Contact: {contact_name}",
+            is_reference=False
+        )
         session_id = backend_response.get("session_id")
         
         if not session_id:

src/services/backend_client.py

@@ -212,23 +212,35 @@ class BackendAPIClient:
             finally:
                 self._track_latency(start_time)
 
-    def create_session(self, title: str, user_id: str) -> Dict[str, Any]:
+    def create_session(
+        self, 
+        user_id: str,
+        contact_id: str,
+        description: str = "",
+        is_reference: bool = False
+    ) -> Dict[str, Any]:
         """
-        Create a new session (profile or contact).
+        Create a new session in the backend API (v2 format).
 
         Args:
-            title: Title for the session
-            user_id: User ID who owns the session
+            user_id: User ID (required for API authentication)
+            contact_id: Contact identifier (maps to project_id in v2)
+            description: Session description (optional)
+            is_reference: Whether this is a reference session
 
         Returns:
-            Created session dict with keys: session_id, title
+            Response dict with session_id, contact_id, description, org_id, project_id
 
         Raises:
             BackendAPIError: If API request fails
         """
         url = f"{self.base_url}/sessions"
 
-        payload = {"title": title}
+        payload = {
+            "contact_id": contact_id,
+            "description": description,
+            "is_reference": is_reference
+        }
         
         # Create span for backend call with OpenTelemetry
         tracer = trace.get_tracer(__name__)
@@ -242,7 +254,9 @@ class BackendAPIClient:
             span.set_attribute("http.method", "POST")
             span.set_attribute("http.url", url)
             span.set_attribute("user_id", user_id)
-            span.set_attribute("title", title)
+            span.set_attribute("contact_id", contact_id)
+            span.set_attribute("description", description)
+            span.set_attribute("is_reference", is_reference)
             
             try:
                 response = requests.post(url, json=payload, headers=headers, timeout=self.timeout)
@@ -326,7 +340,16 @@ class BackendAPIClient:
             user_id: User ID to list sessions for
 
         Returns:
-            List of session dicts
+            List of session dicts with fields:
+            - id: Session ID
+            - contact_id: Contact identifier (project_id in v2)
+            - title: Session title (legacy, same as description)
+            - description: Session description
+            - user_id: User ID
+            - created_at: Unix timestamp
+            - last_interaction: Unix timestamp
+            - message_count: Number of messages
+            - is_reference: Reference flag
 
         Raises:
             BackendAPIError: If API request fails

src/services/s3_backup.py

@@ -0,0 +1,434 @@
+"""
+S3 Backup Manager
+
+Handles automatic backup of SQLite database to S3-compatible storage.
+Provides debouncing to prevent excessive S3 uploads and background
+threading for non-blocking backup operations.
+"""
+
+import os
+import sqlite3
+import hashlib
+import threading
+import time
+import logging
+from datetime import datetime
+from typing import Optional
+from dataclasses import dataclass
+
+import boto3
+from botocore.exceptions import ClientError
+
+from ..utils import s3_logger
+from .s3_config import (
+    S3Config,
+    S3BackupError,
+    S3CredentialsError,
+    S3BucketNotFoundError,
+    S3ConnectionError,
+    DatabaseCorruptedError
+)
+
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class BackupMetadata:
+    """Metadata for a backup file in S3."""
+    s3_key: str
+    last_modified: datetime
+    size_bytes: int
+    checksum_sha256: Optional[str] = None
+
+
+class BackupManager:
+    """
+    Manages automatic backup of SQLite database to S3.
+    
+    Features:
+    - Non-blocking backup requests with debouncing
+    - Background thread execution
+    - Retry logic with exponential backoff
+    - Checksum validation
+    - Graceful error handling
+    """
+    
+    def __init__(self, config: S3Config, db_path: str):
+        """
+        Initialize the backup manager.
+        
+        Args:
+            config: S3 configuration object
+            db_path: Absolute path to SQLite database file
+        
+        Raises:
+            ValueError: If db_path does not exist or is not readable
+        """
+        if not os.path.exists(db_path):
+            raise ValueError(f"Database file does not exist: {db_path}")
+        
+        if not os.access(db_path, os.R_OK):
+            raise ValueError(f"Database file is not readable: {db_path}")
+        
+        self.config = config
+        self.db_path = db_path
+        self.last_backup_request = None
+        self.backup_lock = threading.Lock()
+        self._debounce_thread = None
+        
+        if config.enabled:
+            self.s3_client = config.create_s3_client()
+            logger.info(f"BackupManager initialized for {db_path}")
+        else:
+            self.s3_client = None
+            logger.info("BackupManager initialized but S3 is disabled")
+    
+    def request_backup(self) -> None:
+        """
+        Non-blocking method to request a database backup.
+        
+        Uses debouncing to prevent excessive S3 uploads. Multiple requests
+        within the debounce period are collapsed into a single backup.
+        
+        Side Effects:
+            - Starts background thread if not already running
+            - Updates last_backup_request timestamp
+        """
+        if not self.config.enabled:
+            s3_logger.backup_skip_reason('s3_not_enabled', 'backup_requested')
+            return
+        
+        with self.backup_lock:
+            self.last_backup_request = time.time()
+            
+            # Start debounce thread if not already running
+            if self._debounce_thread is None or not self._debounce_thread.is_alive():
+                self._debounce_thread = threading.Thread(
+                    target=self._debounced_backup,
+                    daemon=True
+                )
+                self._debounce_thread.start()
+                logger.debug("Started debounce thread for backup")
+    
+    def _debounced_backup(self) -> None:
+        """
+        Wait for debounce period, then execute backup.
+        
+        This runs in a background thread and waits for the debounce period
+        to ensure no more backup requests arrive before executing.
+        """
+        debounce_seconds = self.config.debounce_seconds
+        
+        # Wait for debounce period
+        time.sleep(debounce_seconds)
+        
+        # Check if another request came in during debounce
+        with self.backup_lock:
+            time_since_last_request = time.time() - self.last_backup_request
+            if time_since_last_request < debounce_seconds:
+                # Another request came in, let it handle the backup
+                logger.debug("Backup skipped - newer request pending")
+                return
+        
+        # Execute the backup
+        self._execute_backup()
+    
+    def execute_backup_now(self) -> bool:
+        """
+        Synchronous method to execute backup immediately, bypassing debounce.
+        
+        Returns:
+            True if backup succeeded, False if it failed
+        
+        Raises:
+            S3CredentialsError: Invalid S3 credentials
+            S3BucketNotFoundError: Bucket does not exist
+            DatabaseCorruptedError: Source database failed integrity check
+        """
+        if not self.config.enabled:
+            logger.warning("Backup requested but S3 is disabled")
+            return False
+        
+        return self._execute_backup()
+    
+    def _execute_backup(self) -> bool:
+        """
+        Execute the actual backup operation.
+        
+        Process:
+        1. Check database integrity
+        2. Create hot backup using sqlite3.backup()
+        3. Calculate checksum
+        4. Upload to S3 with retries
+        5. Clean up temp files
+        
+        Returns:
+            True if backup succeeded, False otherwise
+        """
+        start_time = time.time()
+        db_size = os.path.getsize(self.db_path)
+        
+        s3_logger.backup_started(self.db_path, db_size)
+        logger.info(f"Starting backup of {self.db_path} ({db_size} bytes)")
+        
+        temp_path = None
+        
+        try:
+            # Validate source database integrity
+            if not self._validate_database(self.db_path):
+                raise DatabaseCorruptedError("Source database failed integrity check")
+            
+            # Create backup using sqlite3.backup() API
+            temp_path = f"{self.db_path}.backup"
+            self._create_hot_backup(temp_path)
+            
+            # Calculate checksum
+            checksum = self._calculate_checksum(temp_path)
+            
+            # Upload to S3 with timestamp
+            timestamp = datetime.utcnow().strftime('%Y-%m-%d-%H-%M-%S')
+            s3_key = f"contacts-{timestamp}.db"
+            
+            if not self._upload_to_s3(temp_path, s3_key, checksum):
+                return False
+            
+            # Success
+            duration = time.time() - start_time
+            upload_size = os.path.getsize(temp_path)
+            s3_logger.backup_completed(duration, s3_key, upload_size)
+            logger.info(f"Backup completed successfully: {s3_key} ({duration:.2f}s)")
+            
+            return True
+            
+        except DatabaseCorruptedError as e:
+            s3_logger.backup_failed(str(e))
+            logger.error(f"Backup failed - database corrupted: {e}")
+            raise
+            
+        except Exception as e:
+            s3_logger.backup_failed(str(e))
+            logger.error(f"Backup failed: {e}", exc_info=True)
+            return False
+            
+        finally:
+            # Clean up temp file
+            if temp_path and os.path.exists(temp_path):
+                try:
+                    os.remove(temp_path)
+                    logger.debug(f"Cleaned up temp file: {temp_path}")
+                except OSError as e:
+                    logger.warning(f"Failed to clean up temp file: {e}")
+    
+    def _create_hot_backup(self, dest_path: str) -> None:
+        """
+        Create a hot backup of the database using sqlite3.backup() API.
+        
+        This method is safe to use while the database is being written to,
+        as sqlite3.backup() handles concurrent access properly.
+        
+        Args:
+            dest_path: Path where backup should be created
+        """
+        logger.debug(f"Creating hot backup to {dest_path}")
+        
+        # Connect to source database
+        source_conn = sqlite3.connect(self.db_path)
+        
+        try:
+            # Create destination connection
+            dest_conn = sqlite3.connect(dest_path)
+            
+            try:
+                # Execute hot backup
+                source_conn.backup(dest_conn)
+                logger.debug("Hot backup completed")
+            finally:
+                dest_conn.close()
+        finally:
+            source_conn.close()
+    
+    def _calculate_checksum(self, file_path: str) -> str:
+        """
+        Calculate SHA-256 checksum of a file.
+        
+        Args:
+            file_path: Path to file
+        
+        Returns:
+            Hexadecimal SHA-256 checksum string
+        """
+        sha256_hash = hashlib.sha256()
+        
+        with open(file_path, 'rb') as f:
+            # Read in chunks for memory efficiency
+            for chunk in iter(lambda: f.read(8192), b''):
+                sha256_hash.update(chunk)
+        
+        checksum = sha256_hash.hexdigest()
+        logger.debug(f"Calculated checksum: {checksum}")
+        return checksum
+    
+    def _upload_to_s3(
+        self,
+        file_path: str,
+        s3_key: str,
+        checksum: str,
+        max_retries: int = 3
+    ) -> bool:
+        """
+        Upload file to S3 with retry logic and exponential backoff.
+        
+        Args:
+            file_path: Path to file to upload
+            s3_key: S3 object key
+            checksum: SHA-256 checksum to store in metadata
+            max_retries: Maximum number of retry attempts
+        
+        Returns:
+            True if upload succeeded, False otherwise
+        """
+        file_size = os.path.getsize(file_path)
+        
+        for attempt in range(max_retries):
+            try:
+                logger.debug(f"Uploading to S3: {s3_key} (attempt {attempt + 1}/{max_retries})")
+                
+                # Upload with metadata
+                with open(file_path, 'rb') as f:
+                    self.s3_client.upload_fileobj(
+                        f,
+                        self.config.bucket,
+                        s3_key,
+                        ExtraArgs={
+                            'Metadata': {
+                                'sha256': checksum,
+                                'source_host': os.uname().nodename,
+                                'db_version': str(sqlite3.sqlite_version)
+                            }
+                        }
+                    )
+                
+                logger.info(f"Upload successful: {s3_key} ({file_size} bytes)")
+                return True
+                
+            except ClientError as e:
+                error_code = e.response['Error']['Code']
+                
+                # Permanent errors - don't retry
+                if error_code in ['NoSuchBucket', 'AccessDenied', 'InvalidAccessKeyId']:
+                    s3_logger.backup_failed(error_code, attempt + 1, max_retries)
+                    logger.error(f"Permanent S3 error: {error_code}")
+                    
+                    if error_code == 'NoSuchBucket':
+                        raise S3BucketNotFoundError(f"Bucket not found: {self.config.bucket}")
+                    elif error_code in ['AccessDenied', 'InvalidAccessKeyId']:
+                        raise S3CredentialsError(f"Invalid credentials: {error_code}")
+                    
+                    return False
+                
+                # Transient errors - retry with backoff
+                if attempt < max_retries - 1:
+                    backoff = 2 ** attempt  # 1s, 2s, 4s
+                    logger.warning(
+                        f"S3 upload failed (attempt {attempt + 1}/{max_retries}): {error_code}, "
+                        f"retrying in {backoff}s"
+                    )
+                    time.sleep(backoff)
+                else:
+                    s3_logger.backup_failed(error_code, attempt + 1, max_retries)
+                    logger.error(f"S3 upload failed after {max_retries} attempts: {error_code}")
+                    return False
+                    
+            except Exception as e:
+                s3_logger.backup_failed(str(e), attempt + 1, max_retries)
+                logger.error(f"Unexpected error during S3 upload: {e}", exc_info=True)
+                return False
+        
+        return False
+    
+    def _validate_database(self, db_path: str) -> bool:
+        """
+        Validate SQLite database integrity.
+        
+        Args:
+            db_path: Path to database file
+        
+        Returns:
+            True if database passes integrity check, False otherwise
+        """
+        try:
+            conn = sqlite3.connect(db_path)
+            cursor = conn.cursor()
+            cursor.execute("PRAGMA integrity_check")
+            result = cursor.fetchone()[0]
+            conn.close()
+            
+            if result == 'ok':
+                logger.debug(f"Database integrity check passed: {db_path}")
+                return True
+            else:
+                logger.error(f"Database integrity check failed: {result}")
+                return False
+                
+        except Exception as e:
+            logger.error(f"Database validation failed: {e}")
+            return False
+    
+    def get_latest_backup(self) -> Optional[BackupMetadata]:
+        """
+        Query S3 for the latest backup file.
+        
+        Returns:
+            BackupMetadata object with latest backup info, or None if no backups found
+        
+        Raises:
+            S3ConnectionError: Network or S3 service error
+        """
+        if not self.config.enabled:
+            return None
+        
+        try:
+            logger.debug(f"Querying S3 for latest backup in bucket: {self.config.bucket}")
+            
+            response = self.s3_client.list_objects_v2(
+                Bucket=self.config.bucket,
+                Prefix='contacts-'
+            )
+            
+            if 'Contents' not in response or len(response['Contents']) == 0:
+                logger.info("No backups found in S3")
+                return None
+            
+            # Find latest by LastModified
+            latest = max(response['Contents'], key=lambda x: x['LastModified'])
+            
+            # Get metadata if available
+            try:
+                head_response = self.s3_client.head_object(
+                    Bucket=self.config.bucket,
+                    Key=latest['Key']
+                )
+                checksum = head_response.get('Metadata', {}).get('sha256')
+            except Exception as e:
+                logger.warning(f"Failed to get metadata for {latest['Key']}: {e}")
+                checksum = None
+            
+            metadata = BackupMetadata(
+                s3_key=latest['Key'],
+                last_modified=latest['LastModified'],
+                size_bytes=latest['Size'],
+                checksum_sha256=checksum
+            )
+            
+            logger.info(f"Latest backup: {metadata.s3_key} ({metadata.last_modified})")
+            return metadata
+            
+        except ClientError as e:
+            error_code = e.response['Error']['Code']
+            logger.error(f"S3 error querying backups: {error_code}")
+            raise S3ConnectionError(f"S3 error: {error_code}") from e
+            
+        except Exception as e:
+            logger.error(f"Unexpected error querying backups: {e}")
+            raise S3ConnectionError(f"Error querying backups: {e}") from e

src/services/s3_config.py

@@ -0,0 +1,226 @@
+"""
+S3 Configuration Module
+
+Manages S3 backup/restore configuration loaded from environment variables.
+Provides validation and credential checking for S3-compatible storage
+(AWS S3, MinIO, DigitalOcean Spaces, etc.).
+"""
+
+import os
+import logging
+from dataclasses import dataclass
+from typing import Optional
+
+import boto3
+from botocore.exceptions import ClientError
+
+
+logger = logging.getLogger(__name__)
+
+
+# Custom Exceptions
+class S3BackupError(Exception):
+    """Base exception for S3 backup operations"""
+    pass
+
+
+class S3CredentialsError(S3BackupError):
+    """Invalid S3 credentials"""
+    pass
+
+
+class S3BucketNotFoundError(S3BackupError):
+    """S3 bucket does not exist"""
+    pass
+
+
+class DatabaseCorruptedError(S3BackupError):
+    """SQLite database failed integrity check"""
+    pass
+
+
+class S3ConnectionError(S3BackupError):
+    """Network error connecting to S3"""
+    pass
+
+
+class RestoreError(S3BackupError):
+    """Critical error during restore that prevents startup"""
+    pass
+
+
+@dataclass
+class S3Config:
+    """
+    S3 configuration for backup/restore operations.
+    
+    Attributes:
+        enabled: Whether S3 backup/restore is enabled
+        bucket: S3 bucket name
+        access_key: AWS access key ID
+        secret_key: AWS secret access key
+        region: AWS region (default: us-east-1)
+        endpoint_url: Custom S3 endpoint for MinIO/DigitalOcean Spaces
+        upload_timeout: Upload timeout in seconds
+        download_timeout: Download timeout in seconds
+        debounce_seconds: Debounce period for backup requests
+    """
+    enabled: bool
+    bucket: Optional[str] = None
+    access_key: Optional[str] = None
+    secret_key: Optional[str] = None
+    region: str = "us-east-1"
+    endpoint_url: Optional[str] = None
+    upload_timeout: int = 60
+    download_timeout: int = 30
+    debounce_seconds: int = 300
+    
+    @staticmethod
+    def from_env() -> 'S3Config':
+        """
+        Factory method to create S3Config from environment variables.
+        
+        Environment Variables:
+            S3_BACKUP_ENABLED: "true" or "false" (default: "false")
+            S3_BUCKET_NAME: S3 bucket name (required if enabled)
+            S3_ACCESS_KEY: AWS access key ID (required if enabled)
+            S3_SECRET_KEY: AWS secret access key (required if enabled)
+            S3_REGION: AWS region (default: "us-east-1")
+            S3_ENDPOINT_URL: Custom S3 endpoint (optional)
+            S3_UPLOAD_TIMEOUT: Upload timeout in seconds (default: 60)
+            S3_DOWNLOAD_TIMEOUT: Download timeout in seconds (default: 30)
+            S3_DEBOUNCE_SECONDS: Debounce period in seconds (default: 300)
+        
+        Returns:
+            S3Config instance with enabled=False if configuration is incomplete
+        """
+        enabled = os.getenv('S3_BACKUP_ENABLED', 'false').lower() == 'true'
+        
+        if not enabled:
+            logger.info("S3 backup/restore disabled (S3_BACKUP_ENABLED not set)")
+            return S3Config(enabled=False)
+        
+        # Load required configuration
+        bucket = os.getenv('S3_BUCKET_NAME')
+        access_key = os.getenv('S3_ACCESS_KEY')
+        secret_key = os.getenv('S3_SECRET_KEY')
+        
+        # Validate required fields
+        missing = []
+        if not bucket:
+            missing.append('S3_BUCKET_NAME')
+        if not access_key:
+            missing.append('S3_ACCESS_KEY')
+        if not secret_key:
+            missing.append('S3_SECRET_KEY')
+        
+        if missing:
+            logger.warning(
+                f"S3 backup disabled - missing required configuration: {', '.join(missing)}"
+            )
+            return S3Config(enabled=False)
+        
+        # Load optional configuration with defaults
+        region = os.getenv('S3_REGION', 'us-east-1')
+        endpoint_url = os.getenv('S3_ENDPOINT_URL')  # None for AWS S3
+        
+        try:
+            upload_timeout = int(os.getenv('S3_UPLOAD_TIMEOUT', '60'))
+            download_timeout = int(os.getenv('S3_DOWNLOAD_TIMEOUT', '30'))
+            debounce_seconds = int(os.getenv('S3_DEBOUNCE_SECONDS', '300'))
+        except ValueError as e:
+            logger.warning(f"Invalid timeout configuration: {e}, using defaults")
+            upload_timeout = 60
+            download_timeout = 30
+            debounce_seconds = 300
+        
+        config = S3Config(
+            enabled=True,
+            bucket=bucket,
+            access_key=access_key,
+            secret_key=secret_key,
+            region=region,
+            endpoint_url=endpoint_url,
+            upload_timeout=upload_timeout,
+            download_timeout=download_timeout,
+            debounce_seconds=debounce_seconds
+        )
+        
+        logger.info(
+            f"S3 backup enabled - bucket: {bucket}, region: {region}, "
+            f"endpoint: {endpoint_url or 'AWS S3'}"
+        )
+        
+        return config
+    
+    def validate_credentials(self) -> bool:
+        """
+        Test S3 credentials by performing a HeadBucket operation.
+        
+        Returns:
+            True if credentials are valid and bucket is accessible
+            False if credentials are invalid or bucket not found
+        
+        Raises:
+            S3ConnectionError: Network or S3 service error
+        """
+        if not self.enabled:
+            return False
+        
+        try:
+            s3_client = boto3.client(
+                's3',
+                endpoint_url=self.endpoint_url,
+                aws_access_key_id=self.access_key,
+                aws_secret_access_key=self.secret_key,
+                region_name=self.region
+            )
+            
+            # HeadBucket validates both credentials and bucket existence
+            s3_client.head_bucket(Bucket=self.bucket)
+            logger.info(f"S3 credentials validated - bucket '{self.bucket}' is accessible")
+            return True
+            
+        except ClientError as e:
+            error_code = e.response['Error']['Code']
+            
+            if error_code == '404':
+                logger.error(f"S3 bucket not found: {self.bucket}")
+                return False
+            elif error_code == '403':
+                logger.error("S3 credentials invalid or insufficient permissions")
+                return False
+            else:
+                logger.error(f"S3 error during credential validation: {error_code}")
+                raise S3ConnectionError(f"S3 error: {error_code}") from e
+                
+        except Exception as e:
+            logger.error(f"Unexpected error during S3 credential validation: {e}")
+            raise S3ConnectionError(f"S3 connection error: {e}") from e
+    
+    def create_s3_client(self):
+        """
+        Create a boto3 S3 client with this configuration.
+        
+        Returns:
+            boto3.client instance configured for S3
+        """
+        if not self.enabled:
+            raise S3CredentialsError("S3 backup is not enabled")
+        
+        from botocore.config import Config
+        
+        boto_config = Config(
+            connect_timeout=5,
+            read_timeout=self.download_timeout,
+            retries={'max_attempts': 0}  # We handle retries manually
+        )
+        
+        return boto3.client(
+            's3',
+            endpoint_url=self.endpoint_url,
+            aws_access_key_id=self.access_key,
+            aws_secret_access_key=self.secret_key,
+            region_name=self.region,
+            config=boto_config
+        )

src/services/s3_restore.py

@@ -0,0 +1,406 @@
+"""
+S3 Restore Manager
+
+Handles automatic restore of SQLite database from S3-compatible storage
+during webapp startup. Provides validation and fallback logic to ensure
+reliable database restoration.
+"""
+
+import os
+import sqlite3
+import hashlib
+import logging
+from datetime import datetime
+from typing import Optional
+from enum import Enum
+
+import boto3
+from botocore.exceptions import ClientError
+
+from ..utils import s3_logger
+from .s3_config import (
+    S3Config,
+    S3ConnectionError,
+    RestoreError,
+    DatabaseCorruptedError
+)
+
+
+logger = logging.getLogger(__name__)
+
+
+class RestoreResult(Enum):
+    """Result of restore operation."""
+    RESTORED_FROM_S3 = "restored_from_s3"
+    LOCAL_NEWER = "local_newer"
+    NO_BACKUP_FOUND = "no_backup_found"
+    VALIDATION_FAILED = "validation_failed"
+    NETWORK_ERROR = "network_error"
+
+
+class RestoreManager:
+    """
+    Manages automatic restore of SQLite database from S3.
+    
+    Features:
+    - Startup restore with validation
+    - Timestamp comparison (local vs S3)
+    - Checksum verification
+    - SQLite integrity check
+    - Atomic file replacement
+    - Fallback chain for reliability
+    """
+    
+    def __init__(self, config: S3Config, db_path: str):
+        """
+        Initialize the restore manager.
+        
+        Args:
+            config: S3 configuration object
+            db_path: Absolute path where database should be restored
+        """
+        self.config = config
+        self.db_path = db_path
+        
+        if config.enabled:
+            self.s3_client = config.create_s3_client()
+            logger.info(f"RestoreManager initialized for {db_path}")
+        else:
+            self.s3_client = None
+            logger.info("RestoreManager initialized but S3 is disabled")
+    
+    def restore_from_s3(self) -> RestoreResult:
+        """
+        Restore database from S3 with validation and fallback logic.
+        
+        Process:
+        1. List backups in S3
+        2. Find latest by LastModified
+        3. Compare timestamps (local vs S3)
+        4. Download if S3 is newer
+        5. Validate checksum and integrity
+        6. Atomic replace local file
+        
+        Returns:
+            RestoreResult enum indicating the outcome
+        
+        Side Effects:
+            - May replace local database file atomically
+            - Creates temp files during download (cleaned up automatically)
+            - Logs all operations with structured logging
+        """
+        start_time = datetime.now()
+        s3_logger.restore_started()
+        
+        if not self.config.enabled:
+            duration = (datetime.now() - start_time).total_seconds()
+            s3_logger.restore_completed(duration, None, None, RestoreResult.NO_BACKUP_FOUND.value)
+            logger.info("Restore skipped - S3 disabled")
+            return RestoreResult.NO_BACKUP_FOUND
+        
+        try:
+            # List backups from S3
+            backups = self._list_backups()
+            
+            if not backups or len(backups) == 0:
+                duration = (datetime.now() - start_time).total_seconds()
+                s3_logger.restore_completed(duration, None, None, RestoreResult.NO_BACKUP_FOUND.value)
+                logger.info("No backups found in S3")
+                return RestoreResult.NO_BACKUP_FOUND
+            
+            # Find latest backup
+            latest_backup = max(backups, key=lambda x: x['LastModified'])
+            s3_key = latest_backup['Key']
+            s3_timestamp = latest_backup['LastModified']
+            
+            logger.info(f"Latest S3 backup: {s3_key} ({s3_timestamp})")
+            
+            # Compare with local file timestamp
+            if os.path.exists(self.db_path):
+                local_mtime = datetime.fromtimestamp(os.path.getmtime(self.db_path))
+                local_mtime = local_mtime.replace(tzinfo=s3_timestamp.tzinfo)  # Make timezone-aware
+                
+                if local_mtime >= s3_timestamp:
+                    duration = (datetime.now() - start_time).total_seconds()
+                    s3_logger.restore_completed(duration, s3_key, None, RestoreResult.LOCAL_NEWER.value)
+                    logger.info(f"Local database is newer ({local_mtime} >= {s3_timestamp}), skipping restore")
+                    return RestoreResult.LOCAL_NEWER
+            
+            # Download from S3
+            temp_path = f"{self.db_path}.restore"
+            download_size = self._download_backup(s3_key, temp_path)
+            
+            # Validate backup
+            if not self.validate_backup(temp_path, s3_key):
+                # Validation failed - use local fallback
+                if os.path.exists(temp_path):
+                    os.remove(temp_path)
+                
+                duration = (datetime.now() - start_time).total_seconds()
+                s3_logger.restore_fallback("validation_failed", "using_local_database")
+                s3_logger.restore_completed(duration, s3_key, None, RestoreResult.VALIDATION_FAILED.value)
+                logger.warning("Backup validation failed, using local database")
+                return RestoreResult.VALIDATION_FAILED
+            
+            # Atomic replace
+            self._atomic_replace(temp_path)
+            
+            duration = (datetime.now() - start_time).total_seconds()
+            s3_logger.restore_completed(duration, s3_key, download_size, RestoreResult.RESTORED_FROM_S3.value)
+            logger.info(f"Restore completed successfully from {s3_key} ({duration:.2f}s)")
+            
+            return RestoreResult.RESTORED_FROM_S3
+            
+        except S3ConnectionError as e:
+            duration = (datetime.now() - start_time).total_seconds()
+            s3_logger.restore_fallback("network_error", "using_local_database")
+            s3_logger.restore_completed(duration, None, None, RestoreResult.NETWORK_ERROR.value)
+            logger.error(f"Network error during restore: {e}")
+            return RestoreResult.NETWORK_ERROR
+            
+        except Exception as e:
+            duration = (datetime.now() - start_time).total_seconds()
+            s3_logger.restore_fallback(str(e), "using_local_database")
+            s3_logger.restore_completed(duration, None, None, RestoreResult.NETWORK_ERROR.value)
+            logger.error(f"Unexpected error during restore: {e}", exc_info=True)
+            return RestoreResult.NETWORK_ERROR
+    
+    def _list_backups(self) -> list:
+        """
+        List all backup files in S3.
+        
+        Returns:
+            List of S3 object dictionaries with Key, LastModified, Size
+        
+        Raises:
+            S3ConnectionError: Network or S3 service error
+        """
+        try:
+            logger.debug(f"Listing backups in bucket: {self.config.bucket}")
+            
+            response = self.s3_client.list_objects_v2(
+                Bucket=self.config.bucket,
+                Prefix='contacts-'
+            )
+            
+            if 'Contents' not in response:
+                return []
+            
+            return response['Contents']
+            
+        except ClientError as e:
+            error_code = e.response['Error']['Code']
+            logger.error(f"S3 error listing backups: {error_code}")
+            raise S3ConnectionError(f"S3 error: {error_code}") from e
+            
+        except Exception as e:
+            logger.error(f"Unexpected error listing backups: {e}")
+            raise S3ConnectionError(f"Error listing backups: {e}") from e
+    
+    def _compare_timestamps(self, local_path: str, s3_timestamp: datetime) -> bool:
+        """
+        Compare local file timestamp with S3 backup timestamp.
+        
+        Args:
+            local_path: Path to local database file
+            s3_timestamp: LastModified timestamp from S3
+        
+        Returns:
+            True if S3 backup is newer, False if local is newer or equal
+        """
+        if not os.path.exists(local_path):
+            return True  # No local file, S3 is "newer"
+        
+        local_mtime = datetime.fromtimestamp(os.path.getmtime(local_path))
+        local_mtime = local_mtime.replace(tzinfo=s3_timestamp.tzinfo)
+        
+        logger.debug(f"Timestamp comparison - Local: {local_mtime}, S3: {s3_timestamp}")
+        return s3_timestamp > local_mtime
+    
+    def _download_backup(self, s3_key: str, dest_path: str) -> int:
+        """
+        Download backup file from S3.
+        
+        Args:
+            s3_key: S3 object key
+            dest_path: Local path where file should be saved
+        
+        Returns:
+            Size of downloaded file in bytes
+        
+        Raises:
+            S3ConnectionError: Network or S3 service error
+        """
+        try:
+            logger.info(f"Downloading backup from S3: {s3_key}")
+            
+            self.s3_client.download_file(
+                self.config.bucket,
+                s3_key,
+                dest_path
+            )
+            
+            file_size = os.path.getsize(dest_path)
+            logger.info(f"Download completed: {file_size} bytes")
+            return file_size
+            
+        except ClientError as e:
+            error_code = e.response['Error']['Code']
+            logger.error(f"S3 error downloading backup: {error_code}")
+            raise S3ConnectionError(f"S3 error: {error_code}") from e
+            
+        except Exception as e:
+            logger.error(f"Unexpected error downloading backup: {e}")
+            raise S3ConnectionError(f"Error downloading backup: {e}") from e
+    
+    def _validate_checksum(self, file_path: str, s3_key: str) -> bool:
+        """
+        Validate file checksum against S3 metadata.
+        
+        Args:
+            file_path: Path to downloaded file
+            s3_key: S3 object key
+        
+        Returns:
+            True if checksum matches or no checksum in metadata, False if mismatch
+        """
+        try:
+            # Get S3 object metadata
+            response = self.s3_client.head_object(
+                Bucket=self.config.bucket,
+                Key=s3_key
+            )
+            
+            remote_checksum = response.get('Metadata', {}).get('sha256')
+            
+            if not remote_checksum:
+                logger.debug("No checksum in S3 metadata, skipping validation")
+                return True
+            
+            # Calculate local checksum
+            sha256_hash = hashlib.sha256()
+            with open(file_path, 'rb') as f:
+                for chunk in iter(lambda: f.read(8192), b''):
+                    sha256_hash.update(chunk)
+            
+            local_checksum = sha256_hash.hexdigest()
+            
+            if local_checksum == remote_checksum:
+                logger.debug("Checksum validation passed")
+                return True
+            else:
+                logger.error(f"Checksum mismatch - Local: {local_checksum}, Remote: {remote_checksum}")
+                return False
+                
+        except Exception as e:
+            logger.warning(f"Checksum validation failed: {e}")
+            return False
+    
+    def _validate_sqlite_integrity(self, file_path: str) -> bool:
+        """
+        Validate SQLite database integrity using PRAGMA integrity_check.
+        
+        Args:
+            file_path: Path to database file
+        
+        Returns:
+            True if database passes integrity check, False otherwise
+        """
+        try:
+            conn = sqlite3.connect(file_path)
+            cursor = conn.cursor()
+            cursor.execute("PRAGMA integrity_check")
+            result = cursor.fetchone()[0]
+            conn.close()
+            
+            if result == 'ok':
+                logger.debug("SQLite integrity check passed")
+                return True
+            else:
+                logger.error(f"SQLite integrity check failed: {result}")
+                return False
+                
+        except Exception as e:
+            logger.error(f"SQLite integrity check failed: {e}")
+            return False
+    
+    def validate_backup(self, file_path: str, s3_key: Optional[str] = None) -> bool:
+        """
+        Validate a backup file's integrity.
+        
+        Performs both checksum validation (if S3 metadata available) and
+        SQLite integrity check.
+        
+        Args:
+            file_path: Path to backup file to validate
+            s3_key: Optional S3 key for checksum validation
+        
+        Returns:
+            True if file is valid, False otherwise
+        """
+        logger.info(f"Validating backup: {file_path}")
+        
+        # Validate checksum if S3 key provided
+        if s3_key and not self._validate_checksum(file_path, s3_key):
+            return False
+        
+        # Validate SQLite integrity
+        if not self._validate_sqlite_integrity(file_path):
+            return False
+        
+        logger.info("Backup validation passed")
+        return True
+    
+    def _atomic_replace(self, temp_path: str) -> None:
+        """
+        Atomically replace local database file with validated backup.
+        
+        Uses os.replace() which is atomic on POSIX systems.
+        
+        Args:
+            temp_path: Path to validated backup file
+        
+        Raises:
+            OSError: If atomic replace fails
+        """
+        try:
+            logger.info(f"Replacing database: {self.db_path}")
+            
+            # Create backup directory if it doesn't exist
+            os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
+            
+            # Atomic replace
+            os.replace(temp_path, self.db_path)
+            
+            logger.info("Database replaced successfully")
+            
+        except OSError as e:
+            logger.error(f"Failed to replace database: {e}")
+            raise
+
+
+def restore_on_startup() -> RestoreResult:
+    """
+    Convenience function to restore database on webapp startup.
+    
+    This function is called from entrypoint.sh before Flask starts.
+    
+    Returns:
+        RestoreResult enum indicating the outcome
+    """
+    try:
+        from .s3_config import S3Config
+        
+        config = S3Config.from_env()
+        db_path = os.getenv("DATABASE_PATH", "/app/data/contacts.db")
+        
+        restore_manager = RestoreManager(config, db_path)
+        result = restore_manager.restore_from_s3()
+        
+        logger.info(f"Startup restore completed: {result.value}")
+        return result
+        
+    except Exception as e:
+        logger.error(f"Startup restore failed: {e}", exc_info=True)
+        s3_logger.restore_fallback(str(e), "using_local_or_empty_database")
+        return RestoreResult.NETWORK_ERROR

src/services/storage_service.py

@@ -2,6 +2,20 @@
 SQLite storage service for user profiles and contact sessions.
 Feature: 012-profile-contact-ui
 Feature: 001-refine-memory-producer-logic (producer_id generation)
+Feature: 015-sqlite-s3-backup (S3 backup triggers)
+
+S3 Backup Trigger Policy:
+- Backup IS triggered for:
+  - New user profile creation (create_or_update_user)
+  - New contact session creation (create_contact_session_with_id)
+  - Contact metadata changes (update_contact_session: name/description)
+  
+- Backup is NOT triggered for:
+  - User login/last_login updates (create_or_update_user on existing user)
+  - last_interaction timestamp updates (update_contact_last_interaction)
+  - Message count increments or other ephemeral data
+  
+This policy optimizes S3 costs by only backing up when critical data changes.
 """
 
 import os
@@ -20,6 +34,37 @@ class NotFoundError(Exception):
     pass
 
 
+# Global S3 backup manager (initialized on first use)
+_backup_manager = None
+
+
+def _get_backup_manager():
+    """
+    Lazy initialization of BackupManager.
+    
+    Returns:
+        BackupManager instance or None if S3 is disabled
+    """
+    global _backup_manager
+    
+    if _backup_manager is None:
+        try:
+            from .s3_config import S3Config
+            from .s3_backup import BackupManager
+            
+            config = S3Config.from_env()
+            db_path = os.getenv("DATABASE_PATH", "data/contacts.db")
+            _backup_manager = BackupManager(config, db_path)
+        except Exception as e:
+            # Log but don't crash if backup initialization fails
+            import logging
+            logger = logging.getLogger(__name__)
+            logger.warning(f"Failed to initialize S3 backup: {e}")
+            _backup_manager = None
+    
+    return _backup_manager
+
+
 def get_db_connection() -> sqlite3.Connection:
     """Get SQLite database connection."""
     db_path = os.getenv("DATABASE_PATH", "data/contacts.db")
@@ -136,6 +181,11 @@ def create_or_update_user(
         # Fetch created profile
         cursor.execute("SELECT * FROM user_profiles WHERE user_id = ?", (user_id,))
         row = cursor.fetchone()
+        
+        # Trigger S3 backup for new user creation (Feature 015)
+        backup_manager = _get_backup_manager()
+        if backup_manager:
+            backup_manager.request_backup()
 
     conn.close()
 
@@ -235,13 +285,14 @@ def create_contact_session(
             cursor.execute(
                 """
                 INSERT INTO contact_sessions 
-                (session_id, user_id, contact_name, contact_description, is_reference, 
+                (session_id, user_id, contact_id, contact_name, contact_description, is_reference, 
                  created_at, last_interaction, normalized_name, sequence_number, producer_id)
-                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                 """,
                 (
                     session_id,
                     user_id,
+                    session_id,  # contact_id defaults to session_id
                     contact_name,
                     contact_description,
                     is_reference,
@@ -262,6 +313,7 @@ def create_contact_session(
             return ContactSession(
                 session_id=row["session_id"],
                 user_id=row["user_id"],
+                contact_id=row.get("contact_id") or row["session_id"],
                 contact_name=row["contact_name"],
                 contact_description=row["contact_description"],
                 is_reference=bool(row["is_reference"]),
@@ -343,13 +395,14 @@ def create_contact_session_with_id(
         cursor.execute(
             """
             INSERT INTO contact_sessions 
-            (session_id, user_id, contact_name, contact_description, is_reference, 
+            (session_id, user_id, contact_id, contact_name, contact_description, is_reference, 
              created_at, last_interaction, normalized_name, sequence_number, producer_id)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             """,
             (
                 session_id,
                 user_id,
+                session_id,  # contact_id defaults to session_id
                 contact_name,
                 contact_description,
                 is_reference,
@@ -366,10 +419,16 @@ def create_contact_session_with_id(
         cursor.execute("SELECT * FROM contact_sessions WHERE session_id = ?", (session_id,))
         row = cursor.fetchone()
         conn.close()
+        
+        # Trigger S3 backup for new contact creation (Feature 015)
+        backup_manager = _get_backup_manager()
+        if backup_manager:
+            backup_manager.request_backup()
 
         return ContactSession(
             session_id=row["session_id"],
             user_id=row["user_id"],
+            contact_id=row["contact_id"] if row["contact_id"] else row["session_id"],
             contact_name=row["contact_name"],
             contact_description=row["contact_description"],
             is_reference=bool(row["is_reference"]),
@@ -438,6 +497,7 @@ def list_contact_sessions(
             ContactSession(
                 session_id=row["session_id"],
                 user_id=row["user_id"],
+                contact_id=safe_row_access(row, "contact_id") or row["session_id"],
                 contact_name=row["contact_name"],
                 contact_description=row["contact_description"],
                 is_reference=bool(row["is_reference"]),
@@ -483,6 +543,7 @@ def get_contact_session(session_id: str) -> Optional[ContactSession]:
     return ContactSession(
         session_id=row["session_id"],
         user_id=row["user_id"],
+        contact_id=row["contact_id"] if "contact_id" in columns else row["session_id"],
         contact_name=row["contact_name"],
         contact_description=row["contact_description"],
         is_reference=bool(row["is_reference"]),
@@ -551,6 +612,12 @@ def update_contact_session(
         query = f"UPDATE contact_sessions SET {', '.join(updates)} WHERE session_id = ?"
         cursor.execute(query, params)
         conn.commit()
+        
+        # Trigger S3 backup for critical metadata changes (Feature 015)
+        # Only when name or description changes (not for last_interaction updates)
+        backup_manager = _get_backup_manager()
+        if backup_manager:
+            backup_manager.request_backup()
 
     # Fetch updated session
     cursor.execute("SELECT * FROM contact_sessions WHERE session_id = ?", (session_id,))

src/utils/s3_logger.py

@@ -0,0 +1,206 @@
+"""
+Structured JSON Logging for S3 Backup/Restore Operations
+
+Provides consistent logging format for observability and debugging.
+All logs are emitted as JSON for easy parsing by log aggregation systems.
+"""
+
+import json
+import logging
+from datetime import datetime
+from typing import Optional, Dict, Any
+
+
+logger = logging.getLogger(__name__)
+
+
+def _log_event(event: str, **kwargs) -> None:
+    """
+    Internal helper to emit structured JSON log messages.
+    
+    Args:
+        event: Event name (e.g., "backup_started", "restore_completed")
+        **kwargs: Additional event-specific fields
+    """
+    log_data = {
+        'event': event,
+        'timestamp': datetime.utcnow().isoformat() + 'Z',
+        **kwargs
+    }
+    
+    # Mask sensitive data
+    if 's3_key' in log_data:
+        # S3 keys are safe to log
+        pass
+    if 'error' in log_data and isinstance(log_data['error'], Exception):
+        # Convert exception to string
+        log_data['error'] = str(log_data['error'])
+    
+    logger.info(json.dumps(log_data))
+
+
+def backup_started(db_path: str, db_size_bytes: int) -> None:
+    """
+    Log that a backup operation has started.
+    
+    Args:
+        db_path: Path to the database file being backed up
+        db_size_bytes: Size of the database in bytes
+    """
+    _log_event(
+        'backup_started',
+        db_path=db_path,
+        db_size_bytes=db_size_bytes
+    )
+
+
+def backup_completed(
+    duration_seconds: float,
+    s3_key: str,
+    upload_size_bytes: int
+) -> None:
+    """
+    Log that a backup operation completed successfully.
+    
+    Args:
+        duration_seconds: Time taken for the backup operation
+        s3_key: S3 object key where backup was stored
+        upload_size_bytes: Size of uploaded file in bytes
+    """
+    _log_event(
+        'backup_completed',
+        duration_seconds=round(duration_seconds, 2),
+        s3_key=s3_key,
+        upload_size_bytes=upload_size_bytes
+    )
+
+
+def backup_failed(
+    error: str,
+    retry_attempt: Optional[int] = None,
+    max_retries: Optional[int] = None
+) -> None:
+    """
+    Log that a backup operation failed.
+    
+    Args:
+        error: Error message or exception
+        retry_attempt: Current retry attempt number (if retrying)
+        max_retries: Maximum number of retries (if retrying)
+    """
+    log_data = {
+        'error': str(error)
+    }
+    
+    if retry_attempt is not None:
+        log_data['retry_attempt'] = retry_attempt
+    if max_retries is not None:
+        log_data['max_retries'] = max_retries
+    
+    _log_event('backup_failed', **log_data)
+
+
+def restore_started() -> None:
+    """Log that a restore operation has started."""
+    _log_event('restore_started')
+
+
+def restore_completed(
+    duration_seconds: float,
+    s3_key: Optional[str],
+    download_size_bytes: Optional[int],
+    result: str
+) -> None:
+    """
+    Log that a restore operation completed.
+    
+    Args:
+        duration_seconds: Time taken for the restore operation
+        s3_key: S3 object key that was restored (None if no restore needed)
+        download_size_bytes: Size of downloaded file (None if no download)
+        result: Restore result enum value (e.g., "restored_from_s3", "local_newer")
+    """
+    log_data = {
+        'duration_seconds': round(duration_seconds, 2),
+        'result': result
+    }
+    
+    if s3_key:
+        log_data['s3_key'] = s3_key
+    if download_size_bytes:
+        log_data['download_size_bytes'] = download_size_bytes
+    
+    _log_event('restore_completed', **log_data)
+
+
+def restore_fallback(reason: str, fallback_action: str) -> None:
+    """
+    Log that a restore operation fell back to an alternative.
+    
+    Args:
+        reason: Reason for fallback (e.g., "validation_failed", "network_error")
+        fallback_action: Action taken (e.g., "using_local_database", "initializing_empty")
+    """
+    _log_event(
+        'restore_fallback',
+        reason=reason,
+        fallback_action=fallback_action
+    )
+
+
+def backup_debounced(pending_requests: int, debounce_seconds: int) -> None:
+    """
+    Log that backup requests are being debounced.
+    
+    Args:
+        pending_requests: Number of backup requests collapsed into one
+        debounce_seconds: Debounce period in seconds
+    """
+    _log_event(
+        'backup_debounced',
+        pending_requests=pending_requests,
+        debounce_seconds=debounce_seconds
+    )
+
+
+def s3_credentials_validated(bucket: str, endpoint: Optional[str]) -> None:
+    """
+    Log that S3 credentials were successfully validated.
+    
+    Args:
+        bucket: S3 bucket name
+        endpoint: S3 endpoint URL (None for AWS S3)
+    """
+    _log_event(
+        's3_credentials_validated',
+        bucket=bucket,
+        endpoint=endpoint or 'AWS S3'
+    )
+
+
+def s3_credentials_invalid(error: str) -> None:
+    """
+    Log that S3 credentials validation failed.
+    
+    Args:
+        error: Error message
+    """
+    _log_event(
+        's3_credentials_invalid',
+        error=str(error)
+    )
+
+
+def backup_skip_reason(reason: str, operation: str) -> None:
+    """
+    Log why a backup was skipped.
+    
+    Args:
+        reason: Reason for skipping (e.g., "s3_not_enabled", "non_critical_update")
+        operation: Operation that would have triggered backup
+    """
+    _log_event(
+        'backup_skip',
+        reason=reason,
+        operation=operation
+    )

tests/integration/test_s3_minio.py

@@ -0,0 +1,214 @@
+"""
+Integration tests for S3 backup/restore with MinIO.
+
+Feature: 015-sqlite-s3-backup
+
+These tests require a running MinIO instance. They can be run with:
+    docker compose up -d minio
+    pytest webapp/tests/integration/test_s3_minio.py
+"""
+
+import os
+import time
+import tempfile
+import sqlite3
+import pytest
+
+from src.services.s3_config import S3Config
+from src.services.s3_backup import BackupManager
+from src.services.s3_restore import RestoreManager, RestoreResult
+
+
+# Check if MinIO is available
+MINIO_AVAILABLE = os.getenv('MINIO_TEST_ENABLED', 'false').lower() == 'true'
+
+skip_if_no_minio = pytest.mark.skipif(
+    not MINIO_AVAILABLE,
+    reason="MinIO not available - set MINIO_TEST_ENABLED=true to run"
+)
+
+
+@pytest.fixture
+def minio_config():
+    """Create S3Config for local MinIO instance."""
+    return S3Config(
+        enabled=True,
+        bucket='test-prepmate-backups',
+        access_key='minioadmin',
+        secret_key='minioadmin',
+        region='us-east-1',
+        endpoint_url='http://localhost:9000',
+        debounce_seconds=5  # Shorter for tests
+    )
+
+
+@pytest.fixture
+def test_db():
+    """Create a temporary test database."""
+    with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
+        db_path = tmp.name
+    
+    # Initialize database
+    conn = sqlite3.connect(db_path)
+    conn.execute("CREATE TABLE contacts (id INTEGER PRIMARY KEY, name TEXT)")
+    conn.execute("INSERT INTO contacts VALUES (1, 'Alice')")
+    conn.execute("INSERT INTO contacts VALUES (2, 'Bob')")
+    conn.commit()
+    conn.close()
+    
+    yield db_path
+    
+    # Cleanup
+    if os.path.exists(db_path):
+        os.unlink(db_path)
+
+
+@skip_if_no_minio
+class TestMinIOIntegration:
+    """Integration tests with MinIO S3-compatible storage."""
+    
+    def test_backup_restore_roundtrip(self, minio_config, test_db):
+        """Test complete backup and restore cycle."""
+        # Create backup manager and execute backup
+        backup_manager = BackupManager(minio_config, test_db)
+        success = backup_manager.execute_backup_now()
+        
+        assert success is True
+        
+        # Verify backup exists in S3
+        metadata = backup_manager.get_latest_backup()
+        assert metadata is not None
+        assert metadata.s3_key.startswith('contacts-')
+        assert metadata.size_bytes > 0
+        
+        # Delete local database
+        os.unlink(test_db)
+        assert not os.path.exists(test_db)
+        
+        # Restore from S3
+        restore_manager = RestoreManager(minio_config, test_db)
+        result = restore_manager.restore_from_s3()
+        
+        assert result == RestoreResult.RESTORED_FROM_S3
+        assert os.path.exists(test_db)
+        
+        # Verify restored data
+        conn = sqlite3.connect(test_db)
+        cursor = conn.cursor()
+        cursor.execute("SELECT * FROM contacts ORDER BY id")
+        rows = cursor.fetchall()
+        conn.close()
+        
+        assert len(rows) == 2
+        assert rows[0] == (1, 'Alice')
+        assert rows[1] == (2, 'Bob')
+    
+    def test_multiple_rapid_creates(self, minio_config, test_db):
+        """Test debouncing with 10 rapid backup requests."""
+        backup_manager = BackupManager(minio_config, test_db)
+        
+        # Get initial backup count
+        initial_metadata = backup_manager.get_latest_backup()
+        initial_count = 0 if initial_metadata is None else 1
+        
+        # Request 10 rapid backups
+        for i in range(10):
+            backup_manager.request_backup()
+            time.sleep(0.1)  # Small delay between requests
+        
+        # Wait for debounce period + execution
+        time.sleep(minio_config.debounce_seconds + 5)
+        
+        # Should have only 1-2 new backups due to debouncing
+        # (1 if all collapsed, 2 if one more request came during execution)
+        final_metadata = backup_manager.get_latest_backup()
+        # Just verify at least one backup succeeded
+        assert final_metadata is not None
+    
+    def test_restore_on_webapp_startup(self, minio_config, test_db):
+        """Test restore during simulated webapp startup."""
+        # Create a backup
+        backup_manager = BackupManager(minio_config, test_db)
+        backup_manager.execute_backup_now()
+        
+        # Modify local database
+        conn = sqlite3.connect(test_db)
+        conn.execute("INSERT INTO contacts VALUES (3, 'Charlie')")
+        conn.commit()
+        conn.close()
+        
+        # Simulate startup restore (should keep newer local)
+        restore_manager = RestoreManager(minio_config, test_db)
+        result = restore_manager.restore_from_s3()
+        
+        # Local is newer, should not restore
+        assert result == RestoreResult.LOCAL_NEWER
+        
+        # Verify Charlie still exists
+        conn = sqlite3.connect(test_db)
+        cursor = conn.cursor()
+        cursor.execute("SELECT COUNT(*) FROM contacts")
+        count = cursor.fetchone()[0]
+        conn.close()
+        
+        assert count == 3
+    
+    def test_graceful_degradation(self, test_db):
+        """Test webapp continues when S3 is unavailable."""
+        # Create config with wrong endpoint
+        bad_config = S3Config(
+            enabled=True,
+            bucket='test-bucket',
+            access_key='test',
+            secret_key='test',
+            endpoint_url='http://nonexistent:9999'
+        )
+        
+        # Backup should fail gracefully
+        backup_manager = BackupManager(bad_config, test_db)
+        success = backup_manager.execute_backup_now()
+        
+        assert success is False  # Failed but didn't crash
+        
+        # Restore should fall back gracefully
+        restore_manager = RestoreManager(bad_config, test_db)
+        result = restore_manager.restore_from_s3()
+        
+        assert result == RestoreResult.NETWORK_ERROR
+    
+    def test_performance_budgets(self, minio_config, test_db):
+        """Test backup/restore performance meets targets."""
+        # Backup performance: <10s for 50MB database
+        # (test DB is smaller, so should be much faster)
+        backup_manager = BackupManager(minio_config, test_db)
+        
+        start = time.time()
+        success = backup_manager.execute_backup_now()
+        backup_duration = time.time() - start
+        
+        assert success is True
+        assert backup_duration < 10, f"Backup took {backup_duration}s (target: <10s)"
+        
+        # Restore performance: <15s
+        restore_manager = RestoreManager(minio_config, test_db)
+        
+        start = time.time()
+        result = restore_manager.restore_from_s3()
+        restore_duration = time.time() - start
+        
+        assert result in [RestoreResult.RESTORED_FROM_S3, RestoreResult.LOCAL_NEWER]
+        assert restore_duration < 15, f"Restore took {restore_duration}s (target: <15s)"
+
+
+if __name__ == '__main__':
+    """
+    Run integration tests with MinIO.
+    
+    Usage:
+        # Start MinIO
+        docker compose up -d minio
+        
+        # Run tests
+        MINIO_TEST_ENABLED=true pytest webapp/tests/integration/test_s3_minio.py -v
+    """
+    pytest.main([__file__, '-v'])

tests/unit/test_s3_backup.py

@@ -0,0 +1,241 @@
+"""
+Unit tests for BackupManager module.
+
+Feature: 015-sqlite-s3-backup
+"""
+
+import os
+import time
+import tempfile
+import sqlite3
+import pytest
+from unittest.mock import patch, MagicMock, call
+from datetime import datetime
+
+from src.services.s3_config import S3Config
+from src.services.s3_backup import BackupManager, BackupMetadata
+
+
+class TestBackupManagerInit:
+    """Tests for BackupManager initialization."""
+    
+    def test_backup_manager_init_enabled(self):
+        """Test BackupManager initialization with S3 enabled."""
+        config = S3Config(
+            enabled=True,
+            bucket='test-bucket',
+            access_key='test-key',
+            secret_key='test-secret'
+        )
+        
+        # Create temp database for testing
+        with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
+            db_path = tmp.name
+        
+        try:
+            # Initialize database
+            conn = sqlite3.connect(db_path)
+            conn.execute("CREATE TABLE test (id INTEGER)")
+            conn.close()
+            
+            # Create manager
+            with patch.object(config, 'create_s3_client'):
+                manager = BackupManager(config, db_path)
+            
+            assert manager.config == config
+            assert manager.db_path == db_path
+            assert manager.last_backup_request is None
+        finally:
+            os.unlink(db_path)
+    
+    def test_backup_manager_init_disabled(self):
+        """Test BackupManager initialization with S3 disabled."""
+        config = S3Config(enabled=False)
+        
+        with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
+            db_path = tmp.name
+        
+        try:
+            manager = BackupManager(config, db_path)
+            
+            assert manager.config == config
+            assert manager.s3_client is None
+        finally:
+            os.unlink(db_path)
+    
+    def test_backup_manager_init_missing_db(self):
+        """Test BackupManager initialization with non-existent database."""
+        config = S3Config(enabled=True)
+        
+        with pytest.raises(ValueError, match="does not exist"):
+            BackupManager(config, "/nonexistent/path.db")
+
+
+class TestBackupManagerRequestBackup:
+    """Tests for BackupManager.request_backup() method."""
+    
+    @patch('threading.Thread')
+    def test_request_backup_debouncing(self, mock_thread):
+        """Test that multiple backup requests are debounced."""
+        config = S3Config(enabled=True, debounce_seconds=5)
+        
+        with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
+            db_path = tmp.name
+        
+        try:
+            conn = sqlite3.connect(db_path)
+            conn.execute("CREATE TABLE test (id INTEGER)")
+            conn.close()
+            
+            with patch.object(config, 'create_s3_client'):
+                manager = BackupManager(config, db_path)
+            
+            # Request multiple backups
+            manager.request_backup()
+            time1 = manager.last_backup_request
+            
+            time.sleep(0.1)
+            
+            manager.request_backup()
+            time2 = manager.last_backup_request
+            
+            # Second request should update timestamp
+            assert time2 > time1
+            
+            # Thread should be started
+            assert mock_thread.called
+        finally:
+            os.unlink(db_path)
+
+
+class TestBackupManagerExecuteBackup:
+    """Tests for BackupManager._execute_backup() method."""
+    
+    @patch('boto3.client')
+    def test_execute_backup_success(self, mock_boto_client):
+        """Test successful backup execution."""
+        # Setup mocks
+        mock_s3 = MagicMock()
+        mock_boto_client.return_value = mock_s3
+        mock_s3.upload_fileobj.return_value = None
+        
+        config = S3Config(
+            enabled=True,
+            bucket='test-bucket',
+            access_key='test-key',
+            secret_key='test-secret'
+        )
+        
+        with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
+            db_path = tmp.name
+        
+        try:
+            # Create test database
+            conn = sqlite3.connect(db_path)
+            conn.execute("CREATE TABLE test (id INTEGER)")
+            conn.execute("INSERT INTO test VALUES (1)")
+            conn.commit()
+            conn.close()
+            
+            # Create manager and execute backup
+            with patch.object(config, 'create_s3_client', return_value=mock_s3):
+                manager = BackupManager(config, db_path)
+                result = manager.execute_backup_now()
+            
+            assert result is True
+            # Verify S3 upload was called
+            assert mock_s3.upload_fileobj.called
+        finally:
+            os.unlink(db_path)
+    
+    @patch('boto3.client')
+    def test_execute_backup_s3_failure(self, mock_boto_client):
+        """Test backup with S3 upload failure."""
+        # Setup mocks
+        mock_s3 = MagicMock()
+        mock_boto_client.return_value = mock_s3
+        
+        from botocore.exceptions import ClientError
+        mock_s3.upload_fileobj.side_effect = ClientError(
+            {'Error': {'Code': 'ServiceUnavailable'}},
+            'UploadFileobj'
+        )
+        
+        config = S3Config(
+            enabled=True,
+            bucket='test-bucket',
+            access_key='test-key',
+            secret_key='test-secret'
+        )
+        
+        with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
+            db_path = tmp.name
+        
+        try:
+            conn = sqlite3.connect(db_path)
+            conn.execute("CREATE TABLE test (id INTEGER)")
+            conn.close()
+            
+            with patch.object(config, 'create_s3_client', return_value=mock_s3):
+                manager = BackupManager(config, db_path)
+                result = manager.execute_backup_now()
+            
+            # Backup should fail gracefully
+            assert result is False
+        finally:
+            os.unlink(db_path)
+
+
+class TestBackupManagerGetLatestBackup:
+    """Tests for BackupManager.get_latest_backup() method."""
+    
+    @patch('boto3.client')
+    def test_get_latest_backup_success(self, mock_boto_client):
+        """Test querying latest backup from S3."""
+        mock_s3 = MagicMock()
+        mock_boto_client.return_value = mock_s3
+        
+        # Mock S3 response
+        mock_s3.list_objects_v2.return_value = {
+            'Contents': [
+                {
+                    'Key': 'contacts-2025-11-27-10-00-00.db',
+                    'LastModified': datetime(2025, 11, 27, 10, 0, 0),
+                    'Size': 1048576
+                },
+                {
+                    'Key': 'contacts-2025-11-27-11-00-00.db',
+                    'LastModified': datetime(2025, 11, 27, 11, 0, 0),
+                    'Size': 1048576
+                }
+            ]
+        }
+        
+        mock_s3.head_object.return_value = {
+            'Metadata': {'sha256': 'abc123'}
+        }
+        
+        config = S3Config(
+            enabled=True,
+            bucket='test-bucket',
+            access_key='test-key',
+            secret_key='test-secret'
+        )
+        
+        with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
+            db_path = tmp.name
+        
+        try:
+            conn = sqlite3.connect(db_path)
+            conn.execute("CREATE TABLE test (id INTEGER)")
+            conn.close()
+            
+            with patch.object(config, 'create_s3_client', return_value=mock_s3):
+                manager = BackupManager(config, db_path)
+                metadata = manager.get_latest_backup()
+            
+            assert metadata is not None
+            assert metadata.s3_key == 'contacts-2025-11-27-11-00-00.db'
+            assert metadata.checksum_sha256 == 'abc123'
+        finally:
+            os.unlink(db_path)

tests/unit/test_s3_config.py

@@ -0,0 +1,136 @@
+"""
+Unit tests for S3Config module.
+
+Feature: 015-sqlite-s3-backup
+"""
+
+import os
+import pytest
+from unittest.mock import patch, MagicMock
+
+from src.services.s3_config import (
+    S3Config,
+    S3CredentialsError,
+    S3BucketNotFoundError,
+    S3ConnectionError
+)
+
+
+class TestS3ConfigFromEnv:
+    """Tests for S3Config.from_env() factory method."""
+    
+    @patch.dict(os.environ, {
+        'S3_BACKUP_ENABLED': 'true',
+        'S3_BUCKET_NAME': 'test-bucket',
+        'S3_ACCESS_KEY': 'test-access-key',
+        'S3_SECRET_KEY': 'test-secret-key',
+        'S3_REGION': 'us-west-2'
+    })
+    def test_from_env_valid_config(self):
+        """Test loading valid S3 configuration from environment."""
+        config = S3Config.from_env()
+        
+        assert config.enabled is True
+        assert config.bucket == 'test-bucket'
+        assert config.access_key == 'test-access-key'
+        assert config.secret_key == 'test-secret-key'
+        assert config.region == 'us-west-2'
+        assert config.upload_timeout == 60
+        assert config.download_timeout == 30
+        assert config.debounce_seconds == 300
+    
+    @patch.dict(os.environ, {}, clear=True)
+    def test_from_env_missing_config(self):
+        """Test that missing S3_BACKUP_ENABLED defaults to disabled."""
+        config = S3Config.from_env()
+        
+        assert config.enabled is False
+    
+    @patch.dict(os.environ, {
+        'S3_BACKUP_ENABLED': 'true',
+        'S3_BUCKET_NAME': 'test-bucket',
+        # Missing S3_ACCESS_KEY and S3_SECRET_KEY
+    }, clear=True)
+    def test_from_env_partial_config(self):
+        """Test that partial configuration disables the feature with warning."""
+        config = S3Config.from_env()
+        
+        assert config.enabled is False
+
+
+class TestS3ConfigValidateCredentials:
+    """Tests for S3Config.validate_credentials() method."""
+    
+    @patch('boto3.client')
+    def test_validate_credentials_success(self, mock_boto_client):
+        """Test successful credential validation."""
+        # Setup mock
+        mock_s3 = MagicMock()
+        mock_boto_client.return_value = mock_s3
+        mock_s3.head_bucket.return_value = {}
+        
+        # Create config
+        config = S3Config(
+            enabled=True,
+            bucket='test-bucket',
+            access_key='test-key',
+            secret_key='test-secret'
+        )
+        
+        # Validate
+        result = config.validate_credentials()
+        
+        assert result is True
+        mock_s3.head_bucket.assert_called_once_with(Bucket='test-bucket')
+    
+    @patch('boto3.client')
+    def test_validate_credentials_bucket_not_found(self, mock_boto_client):
+        """Test validation with non-existent bucket."""
+        # Setup mock
+        mock_s3 = MagicMock()
+        mock_boto_client.return_value = mock_s3
+        
+        from botocore.exceptions import ClientError
+        mock_s3.head_bucket.side_effect = ClientError(
+            {'Error': {'Code': '404'}},
+            'HeadBucket'
+        )
+        
+        # Create config
+        config = S3Config(
+            enabled=True,
+            bucket='nonexistent-bucket',
+            access_key='test-key',
+            secret_key='test-secret'
+        )
+        
+        # Validate
+        result = config.validate_credentials()
+        
+        assert result is False
+    
+    @patch('boto3.client')
+    def test_validate_credentials_access_denied(self, mock_boto_client):
+        """Test validation with invalid credentials."""
+        # Setup mock
+        mock_s3 = MagicMock()
+        mock_boto_client.return_value = mock_s3
+        
+        from botocore.exceptions import ClientError
+        mock_s3.head_bucket.side_effect = ClientError(
+            {'Error': {'Code': '403'}},
+            'HeadBucket'
+        )
+        
+        # Create config
+        config = S3Config(
+            enabled=True,
+            bucket='test-bucket',
+            access_key='invalid-key',
+            secret_key='invalid-secret'
+        )
+        
+        # Validate
+        result = config.validate_credentials()
+        
+        assert result is False

tests/unit/test_s3_restore.py

@@ -0,0 +1,205 @@
+"""
+Unit tests for RestoreManager module.
+
+Feature: 015-sqlite-s3-backup
+"""
+
+import os
+import tempfile
+import sqlite3
+import pytest
+from unittest.mock import patch, MagicMock
+from datetime import datetime, timezone
+
+from src.services.s3_config import S3Config
+from src.services.s3_restore import RestoreManager, RestoreResult
+
+
+class TestRestoreManagerRestoreFromS3:
+    """Tests for RestoreManager.restore_from_s3() method."""
+    
+    @patch('boto3.client')
+    def test_restore_from_s3_success(self, mock_boto_client):
+        """Test successful restore from S3."""
+        mock_s3 = MagicMock()
+        mock_boto_client.return_value = mock_s3
+        
+        # Mock S3 responses
+        mock_s3.list_objects_v2.return_value = {
+            'Contents': [{
+                'Key': 'contacts-2025-11-27-12-00-00.db',
+                'LastModified': datetime(2025, 11, 27, 12, 0, 0, tzinfo=timezone.utc),
+                'Size': 1048576
+            }]
+        }
+        
+        mock_s3.head_object.return_value = {
+            'Metadata': {'sha256': 'abc123'}
+        }
+        
+        # Create temp files
+        with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
+            db_path = tmp.name
+        
+        try:
+            # Create a valid SQLite database for download
+            with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as backup_tmp:
+                backup_path = backup_tmp.name
+            
+            conn = sqlite3.connect(backup_path)
+            conn.execute("CREATE TABLE test (id INTEGER)")
+            conn.execute("INSERT INTO test VALUES (1)")
+            conn.commit()
+            conn.close()
+            
+            # Mock download to copy the backup file
+            def mock_download(bucket, key, dest):
+                import shutil
+                shutil.copy(backup_path, dest)
+            
+            mock_s3.download_file.side_effect = mock_download
+            
+            config = S3Config(
+                enabled=True,
+                bucket='test-bucket',
+                access_key='test-key',
+                secret_key='test-secret'
+            )
+            
+            with patch.object(config, 'create_s3_client', return_value=mock_s3):
+                manager = RestoreManager(config, db_path)
+                result = manager.restore_from_s3()
+            
+            # Should restore successfully
+            assert result == RestoreResult.RESTORED_FROM_S3
+            assert os.path.exists(db_path)
+            
+            # Verify database was restored correctly
+            conn = sqlite3.connect(db_path)
+            cursor = conn.cursor()
+            cursor.execute("SELECT * FROM test")
+            rows = cursor.fetchall()
+            conn.close()
+            
+            assert len(rows) == 1
+            assert rows[0][0] == 1
+            
+            os.unlink(backup_path)
+        finally:
+            if os.path.exists(db_path):
+                os.unlink(db_path)
+    
+    @patch('boto3.client')
+    def test_restore_validation_failure(self, mock_boto_client):
+        """Test restore with corrupted backup file."""
+        mock_s3 = MagicMock()
+        mock_boto_client.return_value = mock_s3
+        
+        mock_s3.list_objects_v2.return_value = {
+            'Contents': [{
+                'Key': 'contacts-2025-11-27-12-00-00.db',
+                'LastModified': datetime(2025, 11, 27, 12, 0, 0, tzinfo=timezone.utc),
+                'Size': 1048576
+            }]
+        }
+        
+        # Mock download to create invalid file
+        def mock_download(bucket, key, dest):
+            with open(dest, 'wb') as f:
+                f.write(b'invalid sqlite data')
+        
+        mock_s3.download_file.side_effect = mock_download
+        
+        config = S3Config(enabled=True, bucket='test-bucket')
+        
+        with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
+            db_path = tmp.name
+        
+        try:
+            with patch.object(config, 'create_s3_client', return_value=mock_s3):
+                manager = RestoreManager(config, db_path)
+                result = manager.restore_from_s3()
+            
+            # Should fail validation and fall back
+            assert result == RestoreResult.VALIDATION_FAILED
+        finally:
+            if os.path.exists(db_path):
+                os.unlink(db_path)
+    
+    @patch('boto3.client')
+    def test_restore_no_backup_found(self, mock_boto_client):
+        """Test restore when no backups exist in S3."""
+        mock_s3 = MagicMock()
+        mock_boto_client.return_value = mock_s3
+        
+        # No backups in S3
+        mock_s3.list_objects_v2.return_value = {}
+        
+        config = S3Config(enabled=True, bucket='test-bucket')
+        
+        with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
+            db_path = tmp.name
+        
+        try:
+            with patch.object(config, 'create_s3_client', return_value=mock_s3):
+                manager = RestoreManager(config, db_path)
+                result = manager.restore_from_s3()
+            
+            assert result == RestoreResult.NO_BACKUP_FOUND
+        finally:
+            if os.path.exists(db_path):
+                os.unlink(db_path)
+
+
+class TestRestoreManagerTimestampComparison:
+    """Tests for RestoreManager._compare_timestamps() method."""
+    
+    def test_timestamp_comparison_s3_newer(self):
+        """Test timestamp comparison when S3 backup is newer."""
+        config = S3Config(enabled=True)
+        
+        with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
+            db_path = tmp.name
+        
+        try:
+            # Create local file
+            with open(db_path, 'w') as f:
+                f.write('test')
+            
+            # Set local mtime to past
+            old_time = datetime(2025, 11, 27, 10, 0, 0).timestamp()
+            os.utime(db_path, (old_time, old_time))
+            
+            with patch.object(config, 'create_s3_client'):
+                manager = RestoreManager(config, db_path)
+            
+            # S3 timestamp is newer
+            s3_timestamp = datetime(2025, 11, 27, 12, 0, 0, tzinfo=timezone.utc)
+            result = manager._compare_timestamps(db_path, s3_timestamp)
+            
+            assert result is True
+        finally:
+            os.unlink(db_path)
+    
+    def test_timestamp_comparison_local_newer(self):
+        """Test timestamp comparison when local file is newer."""
+        config = S3Config(enabled=True)
+        
+        with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp:
+            db_path = tmp.name
+        
+        try:
+            # Create local file with current time
+            with open(db_path, 'w') as f:
+                f.write('test')
+            
+            with patch.object(config, 'create_s3_client'):
+                manager = RestoreManager(config, db_path)
+            
+            # S3 timestamp is older
+            s3_timestamp = datetime(2025, 11, 27, 10, 0, 0, tzinfo=timezone.utc)
+            result = manager._compare_timestamps(db_path, s3_timestamp)
+            
+            assert result is False
+        finally:
+            os.unlink(db_path)