feat(sync): Add metadata logging for sync operations

Author: AlePouroullis

src/humanloop/sync/metadata_handler.py

@@ -0,0 +1,110 @@
+import json
+import time
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, List, Optional
+import logging
+
+logger = logging.getLogger(__name__)
+
+class MetadataHandler:
+    """Handles metadata storage and retrieval for sync operations.
+    
+    This class manages a JSON file that stores the last 5 sync operations
+    and maintains a record of the most recent operation with detailed information.
+    """
+    
+    def __init__(self, base_dir: Path, max_history: int = 5):
+        """Initialize the metadata handler.
+        
+        Args:
+            base_dir: Base directory where metadata will be stored
+            max_history: Maximum number of operations to keep in history
+        """
+        self.base_dir = base_dir
+        self.metadata_file = base_dir / ".sync_metadata.json"
+        self.max_history = max_history
+        self._ensure_metadata_file()
+    
+    def _ensure_metadata_file(self) -> None:
+        """Ensure the metadata file exists with proper structure."""
+        if not self.metadata_file.exists():
+            initial_data = {
+                "last_operation": None,
+                "history": []
+            }
+            self._write_metadata(initial_data)
+    
+    def _read_metadata(self) -> Dict:
+        """Read the current metadata from file."""
+        try:
+            with open(self.metadata_file, 'r') as f:
+                return json.load(f)
+        except Exception as e:
+            logger.error(f"Error reading metadata file: {e}")
+            return {"last_operation": None, "history": []}
+    
+    def _write_metadata(self, data: Dict) -> None:
+        """Write metadata to file."""
+        try:
+            self.metadata_file.parent.mkdir(parents=True, exist_ok=True)
+            with open(self.metadata_file, 'w') as f:
+                json.dump(data, f, indent=2)
+        except Exception as e:
+            logger.error(f"Error writing metadata file: {e}")
+    
+    def log_operation(
+        self,
+        operation_type: str,
+        path: str,
+        environment: Optional[str] = None,
+        successful_files: Optional[List[str]] = None,
+        failed_files: Optional[List[str]] = None,
+        error: Optional[str] = None,
+        start_time: Optional[float] = None
+    ) -> None:
+        """Log a sync operation.
+        
+        Args:
+            operation_type: Type of operation (e.g., "pull", "push")
+            path: The path that was synced
+            environment: Optional environment name
+            successful_files: List of successfully processed files
+            failed_files: List of files that failed to process
+            error: Any error message if the operation failed
+            start_time: Optional timestamp when the operation started (from time.time())
+        """
+        current_time = datetime.now().isoformat()
+        duration_ms = int((time.time() - (start_time or time.time())) * 1000) if start_time else 0
+        
+        operation_data = {
+            "timestamp": current_time,
+            "operation_type": operation_type,
+            "path": path,
+            "environment": environment,
+            "successful_files": successful_files or [],
+            "failed_files": failed_files or [],
+            "error": error,
+            "duration_ms": duration_ms
+        }
+        
+        metadata = self._read_metadata()
+        
+        # Update last operation
+        metadata["last_operation"] = operation_data
+        
+        # Update history
+        metadata["history"].insert(0, operation_data)
+        metadata["history"] = metadata["history"][:self.max_history]
+        
+        self._write_metadata(metadata)
+    
+    def get_last_operation(self) -> Optional[Dict]:
+        """Get the most recent operation details."""
+        metadata = self._read_metadata()
+        return metadata.get("last_operation")
+    
+    def get_history(self) -> List[Dict]:
+        """Get the operation history."""
+        metadata = self._read_metadata()
+        return metadata.get("history", []) 

src/humanloop/sync/sync_client.py

@@ -3,6 +3,8 @@
 from typing import List, TYPE_CHECKING, Optional
 from functools import lru_cache
 from humanloop.types import FileType
+from .metadata_handler import MetadataHandler
+import time
 
 if TYPE_CHECKING:
     from humanloop.base_client import BaseHumanloop
@@ -49,6 +51,8 @@ def __init__(
         self._cache_size = cache_size
         # Create a new cached version of get_file_content with the specified cache size
         self.get_file_content = lru_cache(maxsize=cache_size)(self._get_file_content_impl)
+        # Initialize metadata handler
+        self.metadata = MetadataHandler(self.base_dir)
 
     def _get_file_content_impl(self, path: str, file_type: FileType) -> Optional[str]:
         """Implementation of get_file_content without the cache decorator.
@@ -227,10 +231,6 @@ def _pull_directory(self,
                         logger.warning(f"Skipping unsupported file type: {file.type}")
                         continue
 
-                    if not file.path.startswith(path):
-                        # Filter by path
-                        continue
-
                     # Skip if no content
                     if not getattr(file, "content", None):
                         logger.warning(f"No content found for {file.type} {getattr(file, 'id', '<unknown>')}")
@@ -256,11 +256,12 @@ def _pull_directory(self,
 
         return successful_files
 
-    def pull(self, path: str, environment: str | None = None) -> List[str]:
+    def pull(self, path: str | None = None, environment: str | None = None) -> List[str]:
         """Pull files from Humanloop to local filesystem.
 
         If the path ends with .prompt or .agent, pulls that specific file.
         Otherwise, pulls all files under the specified directory path.
+        If no path is provided, pulls all files from the root.
 
         Args:
             path: The path to pull from (either a specific file or directory)
@@ -269,9 +270,39 @@ def pull(self, path: str, environment: str | None = None) -> List[str]:
         Returns:
             List of successfully processed file paths
         """
-        normalized_path = self._normalize_path(path)
-        if self.is_file(path):
-            self._pull_file(normalized_path, environment)
-            return [path]
-        else:
-            return self._pull_directory(normalized_path, environment)
+        start_time = time.time()
+        try:
+            if path is None:
+                successful_files = self._pull_directory(None, environment)
+                failed_files = []  # Failed files are already logged in _pull_directory
+            else:
+                normalized_path = self._normalize_path(path)
+                if self.is_file(path):
+                    self._pull_file(normalized_path, environment)
+                    successful_files = [path]
+                    failed_files = []
+                else:
+                    successful_files = self._pull_directory(normalized_path, environment)
+                    failed_files = []  # Failed files are already logged in _pull_directory
+            
+            # Log the successful operation
+            self.metadata.log_operation(
+                operation_type="pull",
+                path=path or "",  # Use empty string if path is None
+                environment=environment,
+                successful_files=successful_files,
+                failed_files=failed_files,
+                start_time=start_time
+            )
+            
+            return successful_files
+        except Exception as e:
+            # Log the failed operation
+            self.metadata.log_operation(
+                operation_type="pull",
+                path=path or "",  # Use empty string if path is None
+                environment=environment,
+                error=str(e),
+                start_time=start_time
+            )
+            raise