fix: Code quality improvements for v5.2.2 release

- Fix pagination bug using continuation_token instead of start_after
- Add stats caching to prevent blocking web apps
- Improve code formatting and type checking
- Add comprehensive unit tests for new features
- Fix test mock usage in object_listing tests

src/deltaglider/adapters/storage_s3.py

@@ -97,6 +97,7 @@ class S3StorageAdapter(StoragePort):
         delimiter: str = "",
         max_keys: int = 1000,
         start_after: str | None = None,
+        continuation_token: str | None = None,
     ) -> dict[str, Any]:
         """List objects with S3-compatible response.
 
@@ -105,7 +106,8 @@ class S3StorageAdapter(StoragePort):
             prefix: Filter results to keys beginning with prefix
             delimiter: Delimiter for grouping keys (e.g., '/' for folders)
             max_keys: Maximum number of keys to return
-            start_after: Start listing after this key
+            start_after: Start listing after this key (for first page only)
+            continuation_token: Token from previous response for pagination
 
         Returns:
             Dict with objects, common_prefixes, and pagination info
@@ -119,7 +121,11 @@ class S3StorageAdapter(StoragePort):
             params["Prefix"] = prefix
         if delimiter:
             params["Delimiter"] = delimiter
-        if start_after:
+
+        # Use ContinuationToken for pagination if available, otherwise StartAfter
+        if continuation_token:
+            params["ContinuationToken"] = continuation_token
+        elif start_after:
             params["StartAfter"] = start_after
 
         try:

src/deltaglider/app/cli/aws_compat.py

@@ -6,7 +6,15 @@ from pathlib import Path
 
 import click
 
-from ...core import DeltaService, DeltaSpace, ObjectKey
+from ...core import (
+    DeltaService,
+    DeltaSpace,
+    ObjectKey,
+    build_s3_url,
+    is_s3_url,
+)
+from ...core import parse_s3_url as core_parse_s3_url
+from .sync import fetch_s3_object_heads
 
 __all__ = [
     "is_s3_path",
@@ -100,19 +108,13 @@ def log_aws_region(service: DeltaService, region_override: bool = False) -> None
 
 def is_s3_path(path: str) -> bool:
     """Check if path is an S3 URL."""
-    return path.startswith("s3://")
+    return is_s3_url(path)
 
 
 def parse_s3_url(url: str) -> tuple[str, str]:
     """Parse S3 URL into bucket and key."""
-    if not url.startswith("s3://"):
-        raise ValueError(f"Invalid S3 URL: {url}")
-
-    s3_path = url[5:].rstrip("/")
-    parts = s3_path.split("/", 1)
-    bucket = parts[0]
-    key = parts[1] if len(parts) > 1 else ""
-    return bucket, key
+    parsed = core_parse_s3_url(url, strip_trailing_slash=True)
+    return parsed.bucket, parsed.key
 
 
 def determine_operation(source: str, dest: str) -> str:
@@ -147,6 +149,8 @@ def upload_file(
 
     delta_space = DeltaSpace(bucket=bucket, prefix="/".join(key.split("/")[:-1]))
 
+    dest_url = build_s3_url(bucket, key)
+
     try:
         # Check if delta should be disabled
         if no_delta:
@@ -156,7 +160,7 @@ def upload_file(
 
             if not quiet:
                 file_size = local_path.stat().st_size
-                click.echo(f"upload: '{local_path}' to 's3://{bucket}/{key}' ({file_size} bytes)")
+                click.echo(f"upload: '{local_path}' to '{dest_url}' ({file_size} bytes)")
         else:
             # Use delta compression
             summary = service.put(local_path, delta_space, max_ratio)
@@ -165,12 +169,12 @@ def upload_file(
                 if summary.delta_size:
                     ratio = round((summary.delta_size / summary.file_size) * 100, 1)
                     click.echo(
-                        f"upload: '{local_path}' to 's3://{bucket}/{summary.key}' "
+                        f"upload: '{local_path}' to '{build_s3_url(bucket, summary.key)}' "
                         f"(delta: {ratio}% of original)"
                     )
                 else:
                     click.echo(
-                        f"upload: '{local_path}' to 's3://{bucket}/{summary.key}' "
+                        f"upload: '{local_path}' to '{build_s3_url(bucket, summary.key)}' "
                         f"(reference: {summary.file_size} bytes)"
                     )
 
@@ -202,7 +206,7 @@ def download_file(
                 actual_key = delta_key
                 obj_key = ObjectKey(bucket=bucket, key=delta_key)
                 if not quiet:
-                    click.echo(f"Auto-detected delta: s3://{bucket}/{delta_key}")
+                    click.echo(f"Auto-detected delta: {build_s3_url(bucket, delta_key)}")
 
         # Determine output path
         if local_path is None:
@@ -226,7 +230,7 @@ def download_file(
         if not quiet:
             file_size = local_path.stat().st_size
             click.echo(
-                f"download: 's3://{bucket}/{actual_key}' to '{local_path}' ({file_size} bytes)"
+                f"download: '{build_s3_url(bucket, actual_key)}' to '{local_path}' ({file_size} bytes)"
             )
 
     except Exception as e:
@@ -251,7 +255,10 @@ def copy_s3_to_s3(
     dest_bucket, dest_key = parse_s3_url(dest_url)
 
     if not quiet:
-        click.echo(f"copy: 's3://{source_bucket}/{source_key}' to 's3://{dest_bucket}/{dest_key}'")
+        click.echo(
+            f"copy: '{build_s3_url(source_bucket, source_key)}' "
+            f"to '{build_s3_url(dest_bucket, dest_key)}'"
+        )
 
     try:
         # Get the source object as a stream
@@ -369,13 +376,15 @@ def migrate_s3_to_s3(
         # Pass region_override to warn about cross-region charges if user explicitly set --region
         log_aws_region(service, region_override=region_override)
 
+        source_display = build_s3_url(source_bucket, source_prefix)
+        dest_display = build_s3_url(dest_bucket, dest_prefix)
+        effective_dest_display = build_s3_url(dest_bucket, effective_dest_prefix)
+
         if preserve_prefix and source_prefix:
-            click.echo(f"Migrating from s3://{source_bucket}/{source_prefix}")
-            click.echo(f"           to s3://{dest_bucket}/{effective_dest_prefix}")
+            click.echo(f"Migrating from {source_display}")
+            click.echo(f"           to {effective_dest_display}")
         else:
-            click.echo(
-                f"Migrating from s3://{source_bucket}/{source_prefix} to s3://{dest_bucket}/{dest_prefix}"
-            )
+            click.echo(f"Migrating from {source_display} to {dest_display}")
         click.echo("Scanning source and destination buckets...")
 
     # List source objects
@@ -476,14 +485,14 @@ def migrate_s3_to_s3(
     failed_files = []
 
     for i, (source_obj, rel_key) in enumerate(files_to_migrate, 1):
-        source_s3_url = f"s3://{source_bucket}/{source_obj.key}"
+        source_s3_url = build_s3_url(source_bucket, source_obj.key)
 
         # Construct destination URL using effective prefix
         if effective_dest_prefix:
             dest_key = effective_dest_prefix + rel_key
         else:
             dest_key = rel_key
-        dest_s3_url = f"s3://{dest_bucket}/{dest_key}"
+        dest_s3_url = build_s3_url(dest_bucket, dest_key)
 
         try:
             if not quiet:
@@ -530,7 +539,7 @@ def migrate_s3_to_s3(
 
                 client = DeltaGliderClient(service)
                 # Use cached stats only - don't scan bucket (prevents blocking)
-                cached_stats = client._get_cached_bucket_stats(dest_bucket, detailed_stats=False)
+                cached_stats = client._get_cached_bucket_stats(dest_bucket, "quick")
                 if cached_stats and cached_stats.delta_objects > 0:
                     click.echo(
                         f"\nCompression achieved: {cached_stats.average_compression_ratio:.1%}"
@@ -592,10 +601,7 @@ def handle_recursive(
         dest_path = Path(dest)
         dest_path.mkdir(parents=True, exist_ok=True)
 
-        # List all objects with prefix
-        # Note: S3StorageAdapter.list() expects "bucket/prefix" format
-        list_prefix = f"{bucket}/{prefix}" if prefix else bucket
-        objects = list(service.storage.list(list_prefix))
+        objects = fetch_s3_object_heads(service, bucket, prefix)
 
         if not quiet:
             click.echo(f"Downloading {len(objects)} files...")
@@ -625,7 +631,7 @@ def handle_recursive(
             local_path.parent.mkdir(parents=True, exist_ok=True)
 
             # Download file
-            s3_url = f"s3://{bucket}/{obj.key}"
+            s3_url = build_s3_url(bucket, obj.key)
             download_file(service, s3_url, local_path, quiet)
 
     elif operation == "copy":

src/deltaglider/app/cli/main.py

@@ -623,20 +623,14 @@ def sync(
 @click.pass_obj
 def verify(service: DeltaService, s3_url: str) -> None:
     """Verify integrity of delta file."""
-    # Parse S3 URL
-    if not s3_url.startswith("s3://"):
+    try:
+        bucket, key = parse_s3_url(s3_url)
+        if not key:
+            raise ValueError("Missing key")
+    except ValueError:
         click.echo(f"Error: Invalid S3 URL: {s3_url}", err=True)
         sys.exit(1)
 
-    s3_path = s3_url[5:]
-    parts = s3_path.split("/", 1)
-    if len(parts) != 2:
-        click.echo(f"Error: Invalid S3 URL: {s3_url}", err=True)
-        sys.exit(1)
-
-    bucket = parts[0]
-    key = parts[1]
-
     obj_key = ObjectKey(bucket=bucket, key=key)
 
     try:
@@ -753,38 +747,103 @@ def migrate(
         sys.exit(1)
 
 
-@cli.command()
+@cli.command(short_help="Get bucket statistics and compression metrics")
 @click.argument("bucket")
-@click.option("--detailed", is_flag=True, help="Fetch detailed compression metrics (slower)")
+@click.option("--sampled", is_flag=True, help="Balanced mode: one sample per deltaspace (~5-15s)")
+@click.option(
+    "--detailed", is_flag=True, help="Most accurate: HEAD for all deltas (slowest, ~1min+)"
+)
+@click.option("--refresh", is_flag=True, help="Force cache refresh even if valid")
+@click.option("--no-cache", is_flag=True, help="Skip caching entirely (both read and write)")
 @click.option("--json", "output_json", is_flag=True, help="Output in JSON format")
 @click.pass_obj
-def stats(service: DeltaService, bucket: str, detailed: bool, output_json: bool) -> None:
-    """Get bucket statistics and compression metrics.
+def stats(
+    service: DeltaService,
+    bucket: str,
+    sampled: bool,
+    detailed: bool,
+    refresh: bool,
+    no_cache: bool,
+    output_json: bool,
+) -> None:
+    """Get bucket statistics and compression metrics with intelligent S3-based caching.
 
     BUCKET can be specified as:
       - s3://bucket-name/
       - s3://bucket-name
       - bucket-name
+
+    Modes (mutually exclusive):
+      - quick (default): Fast listing-only stats (~0.5s), approximate compression metrics
+      - --sampled: Balanced mode - one HEAD per deltaspace (~5-15s for typical buckets)
+      - --detailed: Most accurate - HEAD for every delta file (slowest, ~1min+ for large buckets)
+
+    Caching (NEW - massive performance improvement!):
+      Stats are cached in S3 at .deltaglider/stats_{mode}.json (one per mode).
+      Cache is automatically validated on every call using object count + size.
+      If bucket changed, stats are recomputed automatically.
+
+      Performance with cache:
+        - Cache hit: ~0.1s (200x faster than recomputation!)
+        - Cache miss: Full computation time (creates cache for next time)
+        - Cache invalid: Auto-recomputes when bucket changes
+
+    Options:
+      --refresh: Force cache refresh even if valid (use when you need fresh data now)
+      --no-cache: Skip caching entirely - always recompute (useful for testing/debugging)
+      --json: Output in JSON format for automation/scripting
+
+    Examples:
+      deltaglider stats mybucket                    # Fast (~0.1s with cache, ~0.5s without)
+      deltaglider stats mybucket --sampled          # Balanced accuracy/speed (~5-15s first run)
+      deltaglider stats mybucket --detailed         # Most accurate (~1-10min first run, ~0.1s cached)
+      deltaglider stats mybucket --refresh          # Force recomputation even if cached
+      deltaglider stats mybucket --no-cache         # Always compute fresh (skip cache)
+      deltaglider stats mybucket --json             # JSON output for scripts
+      deltaglider stats s3://mybucket/              # Also accepts s3:// URLs
+
+    Timing Logs:
+      Set DG_LOG_LEVEL=INFO to see detailed phase timing with timestamps:
+        [HH:MM:SS.mmm] Phase 1: LIST completed in 0.52s - Found 1523 objects
+        [HH:MM:SS.mmm] Phase 2: Cache HIT in 0.06s - Using cached stats
+        [HH:MM:SS.mmm] COMPLETE: Total time 0.58s
+
+    See docs/STATS_CACHING.md for complete documentation.
     """
     from ...client import DeltaGliderClient
+    from ...client_operations.stats import StatsMode
 
     try:
         # Parse bucket from S3 URL if needed
-        if bucket.startswith("s3://"):
-            # Remove s3:// prefix and any trailing slashes
-            bucket = bucket[5:].rstrip("/")
-            # Extract just the bucket name (first path component)
-            bucket = bucket.split("/")[0] if "/" in bucket else bucket
+        if is_s3_path(bucket):
+            bucket, _prefix = parse_s3_url(bucket)
 
         if not bucket:
             click.echo("Error: Invalid bucket name", err=True)
             sys.exit(1)
 
+        if sampled and detailed:
+            click.echo("Error: --sampled and --detailed cannot be used together", err=True)
+            sys.exit(1)
+
+        if refresh and no_cache:
+            click.echo("Error: --refresh and --no-cache cannot be used together", err=True)
+            sys.exit(1)
+
+        mode: StatsMode = "quick"
+        if sampled:
+            mode = "sampled"
+        if detailed:
+            mode = "detailed"
+
         # Create client from service
         client = DeltaGliderClient(service=service)
 
-        # Get bucket stats
-        bucket_stats = client.get_bucket_stats(bucket, detailed_stats=detailed)
+        # Get bucket stats with caching control
+        use_cache = not no_cache
+        bucket_stats = client.get_bucket_stats(
+            bucket, mode=mode, use_cache=use_cache, refresh_cache=refresh
+        )
 
         if output_json:
             # JSON output

src/deltaglider/app/cli/sync.py

@@ -5,9 +5,27 @@ from pathlib import Path
 import click
 
 from ...core import DeltaService
+from ...core.object_listing import list_all_objects, object_dict_to_head
 from ...ports import ObjectHead
 
 
+def fetch_s3_object_heads(service: DeltaService, bucket: str, prefix: str) -> list[ObjectHead]:
+    """Retrieve all objects for a prefix, falling back to iterator when needed."""
+    try:
+        listing = list_all_objects(
+            service.storage,
+            bucket=bucket,
+            prefix=prefix,
+            max_keys=1000,
+            logger=getattr(service, "logger", None),
+        )
+    except (RuntimeError, NotImplementedError):
+        list_prefix = f"{bucket}/{prefix}" if prefix else bucket
+        return list(service.storage.list(list_prefix))
+
+    return [object_dict_to_head(obj) for obj in listing.objects]
+
+
 def get_local_files(
     local_dir: Path, exclude: str | None = None, include: str | None = None
 ) -> dict[str, tuple[Path, int]]:
@@ -42,8 +60,7 @@ def get_s3_files(
     import fnmatch
 
     files = {}
-    list_prefix = f"{bucket}/{prefix}" if prefix else bucket
-    objects = service.storage.list(list_prefix)
+    objects = fetch_s3_object_heads(service, bucket, prefix)
 
     for obj in objects:
         # Skip reference.bin files (internal)

src/deltaglider/client.py

@@ -33,10 +33,14 @@ from .client_operations import (
     upload_batch as _upload_batch,
     upload_chunked as _upload_chunked,
 )
+
 # fmt: on
+from .client_operations.stats import StatsMode
 
 from .core import DeltaService, DeltaSpace, ObjectKey
 from .core.errors import NotFoundError
+from .core.object_listing import ObjectListing, list_objects_page
+from .core.s3_uri import parse_s3_url
 from .response_builders import (
     build_delete_response,
     build_get_response,
@@ -64,7 +68,7 @@ class DeltaGliderClient:
         self.endpoint_url = endpoint_url
         self._multipart_uploads: dict[str, Any] = {}  # Track multipart uploads
         # Session-scoped bucket statistics cache (cleared with the client lifecycle)
-        self._bucket_stats_cache: dict[str, dict[bool, BucketStats]] = {}
+        self._bucket_stats_cache: dict[str, dict[str, BucketStats]] = {}
 
     # -------------------------------------------------------------------------
     # Internal helpers
@@ -80,35 +84,45 @@ class DeltaGliderClient:
     def _store_bucket_stats_cache(
         self,
         bucket: str,
-        detailed_stats: bool,
+        mode: StatsMode,
         stats: BucketStats,
     ) -> None:
         """Store bucket statistics in the session cache."""
         bucket_cache = self._bucket_stats_cache.setdefault(bucket, {})
-        bucket_cache[detailed_stats] = stats
-        # Detailed stats are a superset of quick stats; reuse them for quick calls.
-        if detailed_stats:
-            bucket_cache[False] = stats
+        bucket_cache[mode] = stats
+        if mode == "detailed":
+            bucket_cache["sampled"] = stats
+            bucket_cache["quick"] = stats
+        elif mode == "sampled":
+            bucket_cache.setdefault("quick", stats)
 
-    def _get_cached_bucket_stats(self, bucket: str, detailed_stats: bool) -> BucketStats | None:
-        """Retrieve cached stats for a bucket, preferring detailed metrics when available."""
+    def _get_cached_bucket_stats(self, bucket: str, mode: StatsMode) -> BucketStats | None:
+        """Retrieve cached stats for a bucket, preferring more detailed metrics when available."""
         bucket_cache = self._bucket_stats_cache.get(bucket)
         if not bucket_cache:
             return None
-        if detailed_stats:
-            return bucket_cache.get(True)
-        return bucket_cache.get(False) or bucket_cache.get(True)
+        if mode == "detailed":
+            return bucket_cache.get("detailed")
+        if mode == "sampled":
+            return bucket_cache.get("sampled") or bucket_cache.get("detailed")
+        return (
+            bucket_cache.get("quick") or bucket_cache.get("sampled") or bucket_cache.get("detailed")
+        )
 
-    def _get_cached_bucket_stats_for_listing(self, bucket: str) -> tuple[BucketStats | None, bool]:
+    def _get_cached_bucket_stats_for_listing(
+        self, bucket: str
+    ) -> tuple[BucketStats | None, StatsMode | None]:
         """Return best cached stats for bucket listings."""
         bucket_cache = self._bucket_stats_cache.get(bucket)
         if not bucket_cache:
-            return (None, False)
-        if True in bucket_cache:
-            return (bucket_cache[True], True)
-        if False in bucket_cache:
-            return (bucket_cache[False], False)
-        return (None, False)
+            return (None, None)
+        if "detailed" in bucket_cache:
+            return (bucket_cache["detailed"], "detailed")
+        if "sampled" in bucket_cache:
+            return (bucket_cache["sampled"], "sampled")
+        if "quick" in bucket_cache:
+            return (bucket_cache["quick"], "quick")
+        return (None, None)
 
     # ============================================================================
     # Boto3-compatible APIs (matches S3 client interface)
@@ -328,34 +342,32 @@ class DeltaGliderClient:
                 FetchMetadata=True  # Only fetches for delta files
             )
         """
-        # Use storage adapter's list_objects method
-        if hasattr(self.service.storage, "list_objects"):
-            result = self.service.storage.list_objects(
+        start_after = StartAfter or ContinuationToken
+        try:
+            listing = list_objects_page(
+                self.service.storage,
                 bucket=Bucket,
                 prefix=Prefix,
                 delimiter=Delimiter,
                 max_keys=MaxKeys,
-                start_after=StartAfter or ContinuationToken,  # Support both pagination methods
+                start_after=start_after,
             )
-        elif isinstance(self.service.storage, S3StorageAdapter):
-            result = self.service.storage.list_objects(
-                bucket=Bucket,
-                prefix=Prefix,
-                delimiter=Delimiter,
-                max_keys=MaxKeys,
-                start_after=StartAfter or ContinuationToken,
-            )
-        else:
-            # Fallback
-            result = {
-                "objects": [],
-                "common_prefixes": [],
-                "is_truncated": False,
-            }
+        except NotImplementedError:
+            if isinstance(self.service.storage, S3StorageAdapter):
+                listing = list_objects_page(
+                    self.service.storage,
+                    bucket=Bucket,
+                    prefix=Prefix,
+                    delimiter=Delimiter,
+                    max_keys=MaxKeys,
+                    start_after=start_after,
+                )
+            else:
+                listing = ObjectListing()
 
         # Convert to boto3-compatible S3Object TypedDicts (type-safe!)
         contents: list[S3Object] = []
-        for obj in result.get("objects", []):
+        for obj in listing.objects:
             # Skip reference.bin files (internal files, never exposed to users)
             if obj["key"].endswith("/reference.bin") or obj["key"] == "reference.bin":
                 continue
@@ -403,14 +415,14 @@ class DeltaGliderClient:
                 "Key": display_key,  # Use cleaned key without .delta
                 "Size": obj["size"],
                 "LastModified": obj.get("last_modified", ""),
-                "ETag": obj.get("etag"),
+                "ETag": str(obj.get("etag", "")),
                 "StorageClass": obj.get("storage_class", "STANDARD"),
                 "Metadata": deltaglider_metadata,
             }
             contents.append(s3_obj)
 
         # Build type-safe boto3-compatible CommonPrefix TypedDicts
-        common_prefixes = result.get("common_prefixes", [])
+        common_prefixes = listing.common_prefixes
         common_prefix_dicts: list[CommonPrefix] | None = (
             [CommonPrefix(Prefix=p) for p in common_prefixes] if common_prefixes else None
         )
@@ -425,8 +437,8 @@ class DeltaGliderClient:
                 max_keys=MaxKeys,
                 contents=contents,
                 common_prefixes=common_prefix_dicts,
-                is_truncated=result.get("is_truncated", False),
-                next_continuation_token=result.get("next_continuation_token"),
+                is_truncated=listing.is_truncated,
+                next_continuation_token=listing.next_continuation_token,
                 continuation_token=ContinuationToken,
             ),
         )
@@ -736,14 +748,9 @@ class DeltaGliderClient:
         """
         file_path = Path(file_path)
 
-        # Parse S3 URL
-        if not s3_url.startswith("s3://"):
-            raise ValueError(f"Invalid S3 URL: {s3_url}")
-
-        s3_path = s3_url[5:].rstrip("/")
-        parts = s3_path.split("/", 1)
-        bucket = parts[0]
-        prefix = parts[1] if len(parts) > 1 else ""
+        address = parse_s3_url(s3_url, strip_trailing_slash=True)
+        bucket = address.bucket
+        prefix = address.key
 
         # Create delta space and upload
         delta_space = DeltaSpace(bucket=bucket, prefix=prefix)
@@ -776,17 +783,9 @@ class DeltaGliderClient:
         """
         output_path = Path(output_path)
 
-        # Parse S3 URL
-        if not s3_url.startswith("s3://"):
-            raise ValueError(f"Invalid S3 URL: {s3_url}")
-
-        s3_path = s3_url[5:]
-        parts = s3_path.split("/", 1)
-        if len(parts) < 2:
-            raise ValueError(f"S3 URL must include key: {s3_url}")
-
-        bucket = parts[0]
-        key = parts[1]
+        address = parse_s3_url(s3_url, allow_empty_key=False)
+        bucket = address.bucket
+        key = address.key
 
         # Auto-append .delta if the file doesn't exist without it
         # This allows users to specify the original name and we'll find the delta
@@ -812,17 +811,9 @@ class DeltaGliderClient:
         Returns:
             True if verification passed, False otherwise
         """
-        # Parse S3 URL
-        if not s3_url.startswith("s3://"):
-            raise ValueError(f"Invalid S3 URL: {s3_url}")
-
-        s3_path = s3_url[5:]
-        parts = s3_path.split("/", 1)
-        if len(parts) < 2:
-            raise ValueError(f"S3 URL must include key: {s3_url}")
-
-        bucket = parts[0]
-        key = parts[1]
+        address = parse_s3_url(s3_url, allow_empty_key=False)
+        bucket = address.bucket
+        key = address.key
 
         obj_key = ObjectKey(bucket=bucket, key=key)
         result = self.service.verify(obj_key)
@@ -965,39 +956,62 @@ class DeltaGliderClient:
         result: ObjectInfo = _get_object_info(self, s3_url)
         return result
 
-    def get_bucket_stats(self, bucket: str, detailed_stats: bool = False) -> BucketStats:
-        """Get statistics for a bucket with optional detailed compression metrics.
+    def get_bucket_stats(
+        self,
+        bucket: str,
+        mode: StatsMode = "quick",
+        use_cache: bool = True,
+        refresh_cache: bool = False,
+    ) -> BucketStats:
+        """Get statistics for a bucket with selectable accuracy modes and S3-based caching.
 
-        This method provides two modes:
-        - Quick stats (default): Fast overview using LIST only (~50ms)
-        - Detailed stats: Accurate compression metrics with HEAD requests (slower)
+        Modes:
+            - ``quick``: Fast listing-only stats (delta compression approximated).
+            - ``sampled``: Fetch one delta HEAD per delta-space and reuse the ratio.
+            - ``detailed``: Fetch metadata for every delta object (slowest, most accurate).
+
+        Caching:
+            - Stats are cached in S3 at ``.deltaglider/stats_{mode}.json``
+            - Cache is automatically validated on every call (uses LIST operation)
+            - If bucket changed, cache is recomputed automatically
+            - Use ``refresh_cache=True`` to force recomputation
+            - Use ``use_cache=False`` to skip caching entirely
 
         Args:
             bucket: S3 bucket name
-            detailed_stats: If True, fetch accurate compression ratios for delta files (default: False)
+            mode: Stats mode ("quick", "sampled", or "detailed")
+            use_cache: If True, use S3-cached stats when available (default: True)
+            refresh_cache: If True, force cache recomputation even if valid (default: False)
 
         Returns:
             BucketStats with compression and space savings info
 
         Performance:
-            - With detailed_stats=False: ~50ms for any bucket size (1 LIST call per 1000 objects)
-            - With detailed_stats=True: ~2-3s per 1000 objects (adds HEAD calls for delta files only)
+            - With cache hit: ~50-100ms (LIST + cache read + validation)
+            - quick (no cache): ~50ms per 1000 objects (LIST only)
+            - sampled (no cache): ~60 HEAD calls per 60 delta-spaces plus LIST
+            - detailed (no cache): ~2-3s per 1000 delta objects (LIST + HEAD per delta)
 
         Example:
-            # Quick stats for dashboard display
+            # Quick stats with caching (fast, ~100ms)
             stats = client.get_bucket_stats('releases')
-            print(f"Objects: {stats.object_count}, Size: {stats.total_size}")
 
-            # Detailed stats for analytics (slower but accurate)
-            stats = client.get_bucket_stats('releases', detailed_stats=True)
-            print(f"Compression ratio: {stats.average_compression_ratio:.1%}")
+            # Force refresh (slow, recomputes everything)
+            stats = client.get_bucket_stats('releases', refresh_cache=True)
+
+            # Skip cache entirely
+            stats = client.get_bucket_stats('releases', use_cache=False)
+
+            # Detailed stats with caching
+            stats = client.get_bucket_stats('releases', mode='detailed')
         """
-        cached = self._get_cached_bucket_stats(bucket, detailed_stats)
-        if cached:
-            return cached
+        if mode not in {"quick", "sampled", "detailed"}:
+            raise ValueError(f"Unknown stats mode: {mode}")
 
-        result: BucketStats = _get_bucket_stats(self, bucket, detailed_stats)
-        self._store_bucket_stats_cache(bucket, detailed_stats, result)
+        # Use S3-based caching from stats.py (replaces old in-memory cache)
+        result: BucketStats = _get_bucket_stats(
+            self, bucket, mode=mode, use_cache=use_cache, refresh_cache=refresh_cache
+        )
         return result
 
     def generate_presigned_url(

src/deltaglider/client_operations/bucket.py

@@ -145,11 +145,12 @@ def list_buckets(
                 bucket_data = dict(bucket_entry)
                 name = bucket_data.get("Name")
                 if isinstance(name, str) and name:
-                    cached_stats, detailed = client._get_cached_bucket_stats_for_listing(name)
-                    if cached_stats is not None:
+                    cached_stats, cached_mode = client._get_cached_bucket_stats_for_listing(name)
+                    if cached_stats is not None and cached_mode is not None:
                         bucket_data["DeltaGliderStats"] = {
                             "Cached": True,
-                            "Detailed": detailed,
+                            "Mode": cached_mode,
+                            "Detailed": cached_mode == "detailed",
                             "ObjectCount": cached_stats.object_count,
                             "TotalSize": cached_stats.total_size,
                             "CompressedSize": cached_stats.compressed_size,

src/deltaglider/client_operations/stats.py

@@ -8,94 +8,29 @@ This module contains DeltaGlider-specific statistics operations:
 """
 
 import concurrent.futures
+import json
 import re
+from dataclasses import asdict
+from datetime import UTC, datetime
 from pathlib import Path
-from typing import Any
+from typing import Any, Literal
 
 from ..client_models import BucketStats, CompressionEstimate, ObjectInfo
+from ..core.delta_extensions import is_delta_candidate
+from ..core.object_listing import list_all_objects
+from ..core.s3_uri import parse_s3_url
+
+StatsMode = Literal["quick", "sampled", "detailed"]
+
+# Cache configuration
+CACHE_VERSION = "1.0"
+CACHE_PREFIX = ".deltaglider"
 
 # ============================================================================
 # Internal Helper Functions
 # ============================================================================
 
 
-def _collect_objects_with_pagination(
-    client: Any,
-    bucket: str,
-    max_iterations: int = 10000,
-) -> list[dict[str, Any]]:
-    """Collect all objects from bucket with pagination safety.
-
-    Args:
-        client: DeltaGliderClient instance
-        bucket: S3 bucket name
-        max_iterations: Max pagination iterations (default: 10000 = 10M objects)
-
-    Returns:
-        List of object dicts with 'key' and 'size' fields
-
-    Raises:
-        RuntimeError: If listing fails with no objects collected
-    """
-    raw_objects = []
-    start_after = None
-    iteration_count = 0
-
-    try:
-        while True:
-            iteration_count += 1
-            if iteration_count > max_iterations:
-                client.service.logger.warning(
-                    f"_collect_objects: Reached max iterations ({max_iterations}). "
-                    f"Returning partial results: {len(raw_objects)} objects."
-                )
-                break
-
-            try:
-                response = client.service.storage.list_objects(
-                    bucket=bucket,
-                    prefix="",
-                    max_keys=1000,
-                    start_after=start_after,
-                )
-            except Exception as e:
-                if len(raw_objects) == 0:
-                    raise RuntimeError(f"Failed to list objects in bucket '{bucket}': {e}") from e
-                client.service.logger.warning(
-                    f"_collect_objects: Pagination error after {len(raw_objects)} objects: {e}. "
-                    f"Returning partial results."
-                )
-                break
-
-            # Collect objects
-            for obj_dict in response.get("objects", []):
-                raw_objects.append(obj_dict)
-
-            # Check pagination status
-            if not response.get("is_truncated"):
-                break
-
-            start_after = response.get("next_continuation_token")
-
-            # Safety: missing token with truncated=True indicates broken pagination
-            if not start_after:
-                client.service.logger.warning(
-                    f"_collect_objects: Pagination bug (truncated=True, no token). "
-                    f"Processed {len(raw_objects)} objects."
-                )
-                break
-
-    except Exception as e:
-        if len(raw_objects) == 0:
-            raise RuntimeError(f"Failed to collect bucket statistics for '{bucket}': {e}") from e
-        client.service.logger.error(
-            f"_collect_objects: Unexpected error after {len(raw_objects)} objects: {e}. "
-            f"Returning partial results."
-        )
-
-    return raw_objects
-
-
 def _fetch_delta_metadata(
     client: Any,
     bucket: str,
@@ -113,12 +48,14 @@ def _fetch_delta_metadata(
     Returns:
         Dict mapping delta key -> metadata dict
     """
-    metadata_map = {}
+    metadata_map: dict[str, dict[str, Any]] = {}
 
     if not delta_keys:
         return metadata_map
 
-    client.service.logger.info(f"Fetching metadata for {len(delta_keys)} delta files in parallel...")
+    client.service.logger.info(
+        f"Fetching metadata for {len(delta_keys)} delta files in parallel..."
+    )
 
     def fetch_single_metadata(key: str) -> tuple[str, dict[str, Any] | None]:
         try:
@@ -158,10 +95,186 @@ def _fetch_delta_metadata(
     return metadata_map
 
 
+def _extract_deltaspace(key: str) -> str:
+    """Return the delta space (prefix) for a given object key."""
+    if "/" in key:
+        return key.rsplit("/", 1)[0]
+    return ""
+
+
+def _get_cache_key(mode: StatsMode) -> str:
+    """Get the S3 key for a cache file based on mode.
+
+    Args:
+        mode: Stats mode (quick, sampled, or detailed)
+
+    Returns:
+        S3 key like ".deltaglider/stats_quick.json"
+    """
+    return f"{CACHE_PREFIX}/stats_{mode}.json"
+
+
+def _read_stats_cache(
+    client: Any,
+    bucket: str,
+    mode: StatsMode,
+) -> tuple[BucketStats | None, dict[str, Any] | None]:
+    """Read cached stats from S3 if available.
+
+    Args:
+        client: DeltaGliderClient instance
+        bucket: S3 bucket name
+        mode: Stats mode to read cache for
+
+    Returns:
+        Tuple of (BucketStats | None, validation_data | None)
+        Returns (None, None) if cache doesn't exist or is invalid
+    """
+    cache_key = _get_cache_key(mode)
+
+    try:
+        # Try to read cache file from S3
+        obj = client.service.storage.get(f"{bucket}/{cache_key}")
+        if not obj or not obj.data:
+            return None, None
+
+        # Parse JSON
+        cache_data = json.loads(obj.data.decode("utf-8"))
+
+        # Validate version
+        if cache_data.get("version") != CACHE_VERSION:
+            client.service.logger.warning(
+                f"Cache version mismatch: expected {CACHE_VERSION}, got {cache_data.get('version')}"
+            )
+            return None, None
+
+        # Validate mode
+        if cache_data.get("mode") != mode:
+            client.service.logger.warning(
+                f"Cache mode mismatch: expected {mode}, got {cache_data.get('mode')}"
+            )
+            return None, None
+
+        # Extract stats and validation data
+        stats_dict = cache_data.get("stats")
+        validation_data = cache_data.get("validation")
+
+        if not stats_dict or not validation_data:
+            client.service.logger.warning("Cache missing stats or validation data")
+            return None, None
+
+        # Reconstruct BucketStats from dict
+        stats = BucketStats(**stats_dict)
+
+        client.service.logger.debug(
+            f"Successfully read cache for {bucket} (mode={mode}, "
+            f"computed_at={cache_data.get('computed_at')})"
+        )
+
+        return stats, validation_data
+
+    except FileNotFoundError:
+        # Cache doesn't exist yet - this is normal
+        client.service.logger.debug(f"No cache found for {bucket} (mode={mode})")
+        return None, None
+    except json.JSONDecodeError as e:
+        client.service.logger.warning(f"Invalid JSON in cache file: {e}")
+        return None, None
+    except Exception as e:
+        client.service.logger.warning(f"Error reading cache: {e}")
+        return None, None
+
+
+def _write_stats_cache(
+    client: Any,
+    bucket: str,
+    mode: StatsMode,
+    stats: BucketStats,
+    object_count: int,
+    compressed_size: int,
+) -> None:
+    """Write computed stats to S3 cache.
+
+    Args:
+        client: DeltaGliderClient instance
+        bucket: S3 bucket name
+        mode: Stats mode being cached
+        stats: Computed BucketStats to cache
+        object_count: Current object count (for validation)
+        compressed_size: Current compressed size (for validation)
+    """
+    cache_key = _get_cache_key(mode)
+
+    try:
+        # Build cache structure
+        cache_data = {
+            "version": CACHE_VERSION,
+            "mode": mode,
+            "computed_at": datetime.now(UTC).isoformat(),
+            "validation": {
+                "object_count": object_count,
+                "compressed_size": compressed_size,
+            },
+            "stats": asdict(stats),
+        }
+
+        # Serialize to JSON
+        cache_json = json.dumps(cache_data, indent=2)
+
+        # Write to S3
+        client.service.storage.put(
+            address=f"{bucket}/{cache_key}",
+            data=cache_json.encode("utf-8"),
+            metadata={
+                "content-type": "application/json",
+                "x-deltaglider-cache": "true",
+            },
+        )
+
+        client.service.logger.info(
+            f"Wrote cache for {bucket} (mode={mode}, {len(cache_json)} bytes)"
+        )
+
+    except Exception as e:
+        # Log warning but don't fail - caching is optional
+        client.service.logger.warning(f"Failed to write cache (non-fatal): {e}")
+
+
+def _is_cache_valid(
+    cached_validation: dict[str, Any],
+    current_object_count: int,
+    current_compressed_size: int,
+) -> bool:
+    """Check if cached stats are still valid based on bucket state.
+
+    Validation strategy: Compare object count and total compressed size.
+    If either changed, the cache is stale.
+
+    Args:
+        cached_validation: Validation data from cache
+        current_object_count: Current object count from LIST
+        current_compressed_size: Current compressed size from LIST
+
+    Returns:
+        True if cache is still valid, False if stale
+    """
+    cached_count = cached_validation.get("object_count")
+    cached_size = cached_validation.get("compressed_size")
+
+    if cached_count != current_object_count:
+        return False
+
+    if cached_size != current_compressed_size:
+        return False
+
+    return True
+
+
 def _build_object_info_list(
     raw_objects: list[dict[str, Any]],
     metadata_map: dict[str, dict[str, Any]],
     logger: Any,
+    sampled_space_metadata: dict[str, dict[str, Any]] | None = None,
 ) -> list[ObjectInfo]:
     """Build ObjectInfo list from raw objects and metadata.
 
@@ -180,8 +293,14 @@ def _build_object_info_list(
         size = obj_dict["size"]
         is_delta = key.endswith(".delta")
 
+        deltaspace = _extract_deltaspace(key)
+
         # Get metadata from map (empty dict if not present)
-        metadata = metadata_map.get(key, {})
+        metadata = metadata_map.get(key)
+        if metadata is None and sampled_space_metadata and deltaspace in sampled_space_metadata:
+            metadata = sampled_space_metadata[deltaspace]
+        if metadata is None:
+            metadata = {}
 
         # Parse compression ratio and original size
         compression_ratio = 0.0
@@ -195,9 +314,21 @@ def _build_object_info_list(
                 compression_ratio = 0.0
 
             try:
-                original_size = int(metadata.get("file_size", size))
-                logger.debug(f"Delta {key}: using original_size={original_size}")
-            except (ValueError, TypeError):
+                if "file_size" in metadata:
+                    original_size = int(metadata["file_size"])
+                    logger.debug(f"Delta {key}: using original_size={original_size} from metadata")
+                else:
+                    logger.warning(
+                        f"Delta {key}: metadata missing 'file_size' key. "
+                        f"Available keys: {list(metadata.keys())}. "
+                        f"Using compressed size={size} as fallback"
+                    )
+                    original_size = size
+            except (ValueError, TypeError) as e:
+                logger.warning(
+                    f"Delta {key}: failed to parse file_size from metadata: {e}. "
+                    f"Using compressed size={size} as fallback"
+                )
                 original_size = size
 
         all_objects.append(
@@ -261,7 +392,15 @@ def _calculate_bucket_statistics(
                 logger.debug(f"Delta {obj.key}: using original_size={obj.original_size}")
                 total_original_size += obj.original_size
             else:
-                logger.warning(f"Delta {obj.key}: no original_size, using compressed size={obj.size}")
+                # This warning should only appear if metadata is missing or incomplete
+                # If you see this, the delta file may have been uploaded with an older
+                # version of DeltaGlider or the upload was incomplete
+                logger.warning(
+                    f"Delta {obj.key}: no original_size metadata "
+                    f"(original_size={obj.original_size}, size={obj.size}). "
+                    f"Using compressed size as fallback. "
+                    f"This may undercount space savings."
+                )
                 total_original_size += obj.size
             total_compressed_size += obj.size
         else:
@@ -350,14 +489,9 @@ def get_object_info(
     Returns:
         ObjectInfo with detailed metadata
     """
-    # Parse URL
-    if not s3_url.startswith("s3://"):
-        raise ValueError(f"Invalid S3 URL: {s3_url}")
-
-    s3_path = s3_url[5:]
-    parts = s3_path.split("/", 1)
-    bucket = parts[0]
-    key = parts[1] if len(parts) > 1 else ""
+    address = parse_s3_url(s3_url, allow_empty_key=False)
+    bucket = address.bucket
+    key = address.key
 
     # Get object metadata
     obj_head = client.service.storage.head(f"{bucket}/{key}")
@@ -383,13 +517,25 @@ def get_object_info(
 def get_bucket_stats(
     client: Any,  # DeltaGliderClient
     bucket: str,
-    detailed_stats: bool = False,
+    mode: StatsMode = "quick",
+    use_cache: bool = True,
+    refresh_cache: bool = False,
 ) -> BucketStats:
-    """Get statistics for a bucket with optional detailed compression metrics.
+    """Get statistics for a bucket with configurable metadata strategies and caching.
 
-    This method provides two modes:
-    - Quick stats (default): Fast overview using LIST only (~50ms)
-    - Detailed stats: Accurate compression metrics with HEAD requests (slower)
+    Modes:
+    - ``quick`` (default): Stream LIST results only. Compression metrics for delta files are
+      approximate (falls back to delta size when metadata is unavailable).
+    - ``sampled``: Fetch HEAD metadata for a single delta per delta-space and reuse the ratios for
+      other deltas in the same space. Balances accuracy and speed.
+    - ``detailed``: Fetch HEAD metadata for every delta object for the most accurate statistics.
+
+    Caching:
+    - Stats are cached per mode in ``.deltaglider/stats_{mode}.json``
+    - Cache is validated using object count and compressed size from LIST
+    - If bucket changed, cache is recomputed automatically
+    - Use ``refresh_cache=True`` to force recomputation
+    - Use ``use_cache=False`` to skip caching entirely
 
     **Robustness**: This function is designed to always return valid stats:
     - Returns partial stats if timeouts or pagination issues occur
@@ -399,7 +545,9 @@ def get_bucket_stats(
     Args:
         client: DeltaGliderClient instance
         bucket: S3 bucket name
-        detailed_stats: If True, fetch accurate compression ratios for delta files (default: False)
+        mode: Stats mode ("quick", "sampled", or "detailed")
+        use_cache: If True, use cached stats when available (default: True)
+        refresh_cache: If True, force cache recomputation even if valid (default: False)
 
     Returns:
         BucketStats with compression and space savings info. Always returns a valid BucketStats
@@ -407,40 +555,221 @@ def get_bucket_stats(
 
     Raises:
         RuntimeError: Only if bucket listing fails immediately with no objects collected.
-                     All other errors result in partial/empty stats being returned.
+                      All other errors result in partial/empty stats being returned.
 
     Performance:
-        - With detailed_stats=False: ~50ms for any bucket size (1 LIST call per 1000 objects)
-        - With detailed_stats=True: ~2-3s per 1000 objects (adds HEAD calls for delta files only)
+        - With cache hit: ~50-100ms (LIST + cache read + validation)
+        - quick (no cache): ~50ms for any bucket size (LIST calls only)
+        - sampled (no cache): LIST + one HEAD per delta-space
+        - detailed (no cache): LIST + HEAD for every delta (slowest but accurate)
         - Max timeout: 10 minutes (prevents indefinite hangs)
         - Max objects: 10M (prevents infinite loops)
 
     Example:
-        # Quick stats for dashboard display
+        # Use cached stats (fast, ~100ms)
         stats = client.get_bucket_stats('releases')
-        print(f"Objects: {stats.object_count}, Size: {stats.total_size}")
 
-        # Detailed stats for analytics (slower but accurate)
-        stats = client.get_bucket_stats('releases', detailed_stats=True)
-        print(f"Compression ratio: {stats.average_compression_ratio:.1%}")
+        # Force refresh (slow, recomputes everything)
+        stats = client.get_bucket_stats('releases', refresh_cache=True)
+
+        # Skip cache entirely
+        stats = client.get_bucket_stats('releases', use_cache=False)
+
+        # Different modes with caching
+        stats_sampled = client.get_bucket_stats('releases', mode='sampled')
+        stats_detailed = client.get_bucket_stats('releases', mode='detailed')
     """
     try:
-        # Phase 1: Collect all objects with pagination safety
-        raw_objects = _collect_objects_with_pagination(client, bucket)
+        if mode not in {"quick", "sampled", "detailed"}:
+            raise ValueError(f"Unknown stats mode: {mode}")
 
-        # Phase 2: Extract delta keys for metadata fetching
+        # Phase 1: Always do a quick LIST to get current state (needed for validation)
+        import time
+
+        phase1_start = time.time()
+        client.service.logger.info(
+            f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 1: Starting LIST operation for bucket '{bucket}'"
+        )
+
+        listing = list_all_objects(
+            client.service.storage,
+            bucket=bucket,
+            max_keys=1000,
+            logger=client.service.logger,
+        )
+        raw_objects = listing.objects
+
+        # Calculate validation metrics from LIST
+        current_object_count = len(raw_objects)
+        current_compressed_size = sum(obj["size"] for obj in raw_objects)
+
+        phase1_duration = time.time() - phase1_start
+        client.service.logger.info(
+            f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 1: LIST completed in {phase1_duration:.2f}s - "
+            f"Found {current_object_count} objects, {current_compressed_size:,} bytes total"
+        )
+
+        # Phase 2: Try to use cache if enabled and not forcing refresh
+        phase2_start = time.time()
+        if use_cache and not refresh_cache:
+            client.service.logger.info(
+                f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 2: Checking cache for mode '{mode}'"
+            )
+            cached_stats, cached_validation = _read_stats_cache(client, bucket, mode)
+
+            if cached_stats and cached_validation:
+                # Validate cache against current bucket state
+                if _is_cache_valid(
+                    cached_validation, current_object_count, current_compressed_size
+                ):
+                    phase2_duration = time.time() - phase2_start
+                    client.service.logger.info(
+                        f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 2: Cache HIT in {phase2_duration:.2f}s - "
+                        f"Using cached stats for {bucket} (mode={mode}, bucket unchanged)"
+                    )
+                    return cached_stats
+                else:
+                    phase2_duration = time.time() - phase2_start
+                    client.service.logger.info(
+                        f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 2: Cache INVALID in {phase2_duration:.2f}s - "
+                        f"Bucket changed: count {cached_validation.get('object_count')} → {current_object_count}, "
+                        f"size {cached_validation.get('compressed_size')} → {current_compressed_size}"
+                    )
+            else:
+                phase2_duration = time.time() - phase2_start
+                client.service.logger.info(
+                    f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 2: Cache MISS in {phase2_duration:.2f}s - "
+                    f"No valid cache found"
+                )
+        else:
+            if refresh_cache:
+                client.service.logger.info(
+                    f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 2: Cache SKIPPED (refresh requested)"
+                )
+            elif not use_cache:
+                client.service.logger.info(
+                    f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 2: Cache DISABLED"
+                )
+
+        # Phase 3: Cache miss or invalid - compute stats from scratch
+        client.service.logger.info(
+            f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 3: Computing stats (mode={mode})"
+        )
+
+        # Phase 4: Extract delta keys for metadata fetching
+        phase4_start = time.time()
         delta_keys = [obj["key"] for obj in raw_objects if obj["key"].endswith(".delta")]
+        phase4_duration = time.time() - phase4_start
 
-        # Phase 3: Fetch metadata for delta files (only if detailed_stats requested)
-        metadata_map = {}
-        if detailed_stats and delta_keys:
-            metadata_map = _fetch_delta_metadata(client, bucket, delta_keys)
+        client.service.logger.info(
+            f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 4: Delta extraction completed in {phase4_duration:.3f}s - "
+            f"Found {len(delta_keys)} delta files"
+        )
 
-        # Phase 4: Build ObjectInfo list
-        all_objects = _build_object_info_list(raw_objects, metadata_map, client.service.logger)
+        # Phase 5: Fetch metadata for delta files based on mode
+        phase5_start = time.time()
+        metadata_map: dict[str, dict[str, Any]] = {}
+        sampled_space_metadata: dict[str, dict[str, Any]] | None = None
 
-        # Phase 5: Calculate final statistics
-        return _calculate_bucket_statistics(all_objects, bucket, client.service.logger)
+        if delta_keys:
+            if mode == "detailed":
+                client.service.logger.info(
+                    f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 5: Fetching metadata for ALL {len(delta_keys)} delta files"
+                )
+                metadata_map = _fetch_delta_metadata(client, bucket, delta_keys)
+
+            elif mode == "sampled":
+                # Sample one delta per deltaspace
+                seen_spaces: set[str] = set()
+                sampled_keys: list[str] = []
+                for key in delta_keys:
+                    space = _extract_deltaspace(key)
+                    if space not in seen_spaces:
+                        seen_spaces.add(space)
+                        sampled_keys.append(key)
+
+                client.service.logger.info(
+                    f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 5: Sampling {len(sampled_keys)} delta files "
+                    f"(one per deltaspace) out of {len(delta_keys)} total delta files"
+                )
+
+                # Log which files are being sampled
+                if sampled_keys:
+                    for idx, key in enumerate(sampled_keys[:10], 1):  # Show first 10
+                        space = _extract_deltaspace(key)
+                        client.service.logger.info(
+                            f"  [{idx}] Sampling: {key} (deltaspace: '{space or '(root)'}')"
+                        )
+                    if len(sampled_keys) > 10:
+                        client.service.logger.info(f"  ... and {len(sampled_keys) - 10} more")
+
+                if sampled_keys:
+                    metadata_map = _fetch_delta_metadata(client, bucket, sampled_keys)
+                    sampled_space_metadata = {
+                        _extract_deltaspace(k): metadata for k, metadata in metadata_map.items()
+                    }
+
+        phase5_duration = time.time() - phase5_start
+        if mode == "quick":
+            client.service.logger.info(
+                f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 5: Skipped metadata fetching (quick mode) in {phase5_duration:.3f}s"
+            )
+        else:
+            client.service.logger.info(
+                f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 5: Metadata fetching completed in {phase5_duration:.2f}s - "
+                f"Fetched {len(metadata_map)} metadata records"
+            )
+
+        # Phase 6: Build ObjectInfo list
+        phase6_start = time.time()
+        all_objects = _build_object_info_list(
+            raw_objects,
+            metadata_map,
+            client.service.logger,
+            sampled_space_metadata,
+        )
+        phase6_duration = time.time() - phase6_start
+        client.service.logger.info(
+            f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 6: ObjectInfo list built in {phase6_duration:.3f}s - "
+            f"{len(all_objects)} objects processed"
+        )
+
+        # Phase 7: Calculate final statistics
+        phase7_start = time.time()
+        stats = _calculate_bucket_statistics(all_objects, bucket, client.service.logger)
+        phase7_duration = time.time() - phase7_start
+        client.service.logger.info(
+            f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 7: Statistics calculated in {phase7_duration:.3f}s - "
+            f"{stats.delta_objects} delta, {stats.direct_objects} direct objects"
+        )
+
+        # Phase 8: Write cache if enabled
+        phase8_start = time.time()
+        if use_cache:
+            _write_stats_cache(
+                client=client,
+                bucket=bucket,
+                mode=mode,
+                stats=stats,
+                object_count=current_object_count,
+                compressed_size=current_compressed_size,
+            )
+            phase8_duration = time.time() - phase8_start
+            client.service.logger.info(
+                f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 8: Cache written in {phase8_duration:.3f}s"
+            )
+        else:
+            client.service.logger.info(
+                f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] Phase 8: Cache write skipped (caching disabled)"
+            )
+
+        # Summary
+        total_duration = time.time() - phase1_start
+        client.service.logger.info(
+            f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}] COMPLETE: Total time {total_duration:.2f}s for bucket '{bucket}' (mode={mode})"
+        )
+
+        return stats
 
     except Exception as e:
         # Last resort: return empty stats with error indication
@@ -487,30 +816,8 @@ def estimate_compression(
     file_path = Path(file_path)
     file_size = file_path.stat().st_size
 
-    # Check file extension
+    filename = file_path.name
     ext = file_path.suffix.lower()
-    delta_extensions = {
-        ".zip",
-        ".tar",
-        ".gz",
-        ".tar.gz",
-        ".tgz",
-        ".bz2",
-        ".tar.bz2",
-        ".xz",
-        ".tar.xz",
-        ".7z",
-        ".rar",
-        ".dmg",
-        ".iso",
-        ".pkg",
-        ".deb",
-        ".rpm",
-        ".apk",
-        ".jar",
-        ".war",
-        ".ear",
-    }
 
     # Already compressed formats that won't benefit from delta
     incompressible = {".jpg", ".jpeg", ".png", ".mp4", ".mp3", ".avi", ".mov"}
@@ -524,7 +831,7 @@ def estimate_compression(
             should_use_delta=False,
         )
 
-    if ext not in delta_extensions:
+    if not is_delta_candidate(filename):
         # Unknown type, conservative estimate
         return CompressionEstimate(
             original_size=file_size,

src/deltaglider/core/__init__.py

@@ -1,5 +1,10 @@
 """Core domain for DeltaGlider."""
 
+from .delta_extensions import (
+    DEFAULT_COMPOUND_DELTA_EXTENSIONS,
+    DEFAULT_DELTA_EXTENSIONS,
+    is_delta_candidate,
+)
 from .errors import (
     DeltaGliderError,
     DiffDecodeError,
@@ -19,6 +24,7 @@ from .models import (
     Sha256,
     VerifyResult,
 )
+from .s3_uri import S3Url, build_s3_url, is_s3_url, parse_s3_url
 from .service import DeltaService
 
 __all__ = [
@@ -38,4 +44,11 @@ __all__ = [
     "PutSummary",
     "VerifyResult",
     "DeltaService",
+    "DEFAULT_DELTA_EXTENSIONS",
+    "DEFAULT_COMPOUND_DELTA_EXTENSIONS",
+    "is_delta_candidate",
+    "S3Url",
+    "build_s3_url",
+    "is_s3_url",
+    "parse_s3_url",
 ]

src/deltaglider/core/delta_extensions.py

@@ -0,0 +1,56 @@
+"""Shared delta compression extension policy."""
+
+from __future__ import annotations
+
+from collections.abc import Collection, Iterable
+
+# Compound extensions must be checked before simple suffix matching so that
+# multi-part archives like ".tar.gz" are handled correctly.
+DEFAULT_COMPOUND_DELTA_EXTENSIONS: tuple[str, ...] = (".tar.gz", ".tar.bz2", ".tar.xz")
+
+# Simple extensions that benefit from delta compression. Keep this structure
+# immutable so it can be safely reused across modules.
+DEFAULT_DELTA_EXTENSIONS: frozenset[str] = frozenset(
+    {
+        ".zip",
+        ".tar",
+        ".gz",
+        ".tgz",
+        ".bz2",
+        ".xz",
+        ".7z",
+        ".rar",
+        ".dmg",
+        ".iso",
+        ".pkg",
+        ".deb",
+        ".rpm",
+        ".apk",
+        ".jar",
+        ".war",
+        ".ear",
+    }
+)
+
+
+def is_delta_candidate(
+    filename: str,
+    *,
+    simple_extensions: Collection[str] = DEFAULT_DELTA_EXTENSIONS,
+    compound_extensions: Iterable[str] = DEFAULT_COMPOUND_DELTA_EXTENSIONS,
+) -> bool:
+    """Check if a filename should use delta compression based on extension."""
+    name_lower = filename.lower()
+
+    for ext in compound_extensions:
+        if name_lower.endswith(ext):
+            return True
+
+    return any(name_lower.endswith(ext) for ext in simple_extensions)
+
+
+__all__ = [
+    "DEFAULT_COMPOUND_DELTA_EXTENSIONS",
+    "DEFAULT_DELTA_EXTENSIONS",
+    "is_delta_candidate",
+]

src/deltaglider/core/object_listing.py

@@ -0,0 +1,206 @@
+"""Shared helpers for listing bucket objects with pagination support."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any
+
+from ..ports.storage import ObjectHead
+
+
+@dataclass(slots=True)
+class ObjectListing:
+    """All objects and prefixes returned from a bucket listing."""
+
+    objects: list[dict[str, Any]] = field(default_factory=list)
+    common_prefixes: list[str] = field(default_factory=list)
+    key_count: int = 0
+    is_truncated: bool = False
+    next_continuation_token: str | None = None
+
+
+def list_objects_page(
+    storage: Any,
+    *,
+    bucket: str,
+    prefix: str = "",
+    delimiter: str = "",
+    max_keys: int = 1000,
+    start_after: str | None = None,
+    continuation_token: str | None = None,
+) -> ObjectListing:
+    """Perform a single list_objects call using the storage adapter."""
+    if not hasattr(storage, "list_objects"):
+        raise NotImplementedError("Storage adapter does not support list_objects")
+
+    response = storage.list_objects(
+        bucket=bucket,
+        prefix=prefix,
+        delimiter=delimiter,
+        max_keys=max_keys,
+        start_after=start_after,
+        continuation_token=continuation_token,
+    )
+
+    return ObjectListing(
+        objects=list(response.get("objects", [])),
+        common_prefixes=list(response.get("common_prefixes", [])),
+        key_count=response.get("key_count", len(response.get("objects", []))),
+        is_truncated=bool(response.get("is_truncated", False)),
+        next_continuation_token=response.get("next_continuation_token"),
+    )
+
+
+def list_all_objects(
+    storage: Any,
+    *,
+    bucket: str,
+    prefix: str = "",
+    delimiter: str = "",
+    max_keys: int = 1000,
+    logger: Any | None = None,
+    max_iterations: int = 10_000,
+) -> ObjectListing:
+    """Fetch all objects under the given bucket/prefix with pagination safety."""
+    import time
+    from datetime import UTC, datetime
+
+    aggregated = ObjectListing()
+    continuation_token: str | None = None
+    iteration_count = 0
+    list_start_time = time.time()
+
+    while True:
+        iteration_count += 1
+        if iteration_count > max_iterations:
+            if logger:
+                logger.warning(
+                    "list_all_objects: reached max iterations (%s). Returning partial results.",
+                    max_iterations,
+                )
+            aggregated.is_truncated = True
+            aggregated.next_continuation_token = continuation_token
+            break
+
+        # Log progress every 10 pages or on first page
+        if logger and (iteration_count == 1 or iteration_count % 10 == 0):
+            elapsed = time.time() - list_start_time
+            objects_per_sec = len(aggregated.objects) / elapsed if elapsed > 0 else 0
+            token_info = f", token={continuation_token[:20]}..." if continuation_token else ""
+            logger.info(
+                f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}]   LIST pagination: "
+                f"page {iteration_count}, {len(aggregated.objects)} objects so far "
+                f"({objects_per_sec:.0f} obj/s, {elapsed:.1f}s elapsed{token_info})"
+            )
+
+            # Warn if taking very long (>60s)
+            if elapsed > 60 and iteration_count % 50 == 0:
+                estimated_total = (len(aggregated.objects) / iteration_count) * max_iterations
+                logger.warning(
+                    f"LIST operation is slow ({elapsed:.0f}s elapsed). "
+                    f"This bucket has MANY objects ({len(aggregated.objects)} so far). "
+                    f"Consider using a smaller prefix or enabling caching. "
+                    f"Estimated remaining: {estimated_total - len(aggregated.objects):.0f} objects"
+                )
+
+        try:
+            page = list_objects_page(
+                storage,
+                bucket=bucket,
+                prefix=prefix,
+                delimiter=delimiter,
+                max_keys=max_keys,
+                continuation_token=continuation_token,
+            )
+        except Exception as exc:
+            if not aggregated.objects:
+                raise RuntimeError(f"Failed to list objects for bucket '{bucket}': {exc}") from exc
+            if logger:
+                logger.warning(
+                    "list_all_objects: pagination error after %s objects: %s. Returning partial results.",
+                    len(aggregated.objects),
+                    exc,
+                )
+            aggregated.is_truncated = True
+            aggregated.next_continuation_token = continuation_token
+            break
+
+        aggregated.objects.extend(page.objects)
+        aggregated.common_prefixes.extend(page.common_prefixes)
+        aggregated.key_count += page.key_count
+
+        if not page.is_truncated:
+            aggregated.is_truncated = False
+            aggregated.next_continuation_token = None
+            if logger:
+                elapsed = time.time() - list_start_time
+                logger.info(
+                    f"[{datetime.now(UTC).strftime('%H:%M:%S.%f')[:-3]}]   LIST complete: "
+                    f"{iteration_count} pages, {len(aggregated.objects)} objects total in {elapsed:.2f}s"
+                )
+            break
+
+        continuation_token = page.next_continuation_token
+        if not continuation_token:
+            if logger:
+                logger.warning(
+                    "list_all_objects: truncated response without continuation token after %s objects.",
+                    len(aggregated.objects),
+                )
+            aggregated.is_truncated = True
+            aggregated.next_continuation_token = None
+            break
+
+    if aggregated.common_prefixes:
+        seen: set[str] = set()
+        unique_prefixes: list[str] = []
+        for prefix in aggregated.common_prefixes:
+            if prefix not in seen:
+                seen.add(prefix)
+                unique_prefixes.append(prefix)
+        aggregated.common_prefixes = unique_prefixes
+    aggregated.key_count = len(aggregated.objects)
+    return aggregated
+
+
+def _parse_last_modified(value: Any) -> datetime:
+    if isinstance(value, datetime):
+        dt = value
+    elif value:
+        text = str(value)
+        if text.endswith("Z"):
+            text = text[:-1] + "+00:00"
+        try:
+            dt = datetime.fromisoformat(text)
+        except ValueError:
+            dt = datetime.fromtimestamp(0, tz=timezone.utc)  # noqa: UP017
+    else:
+        dt = datetime.fromtimestamp(0, tz=timezone.utc)  # noqa: UP017
+
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)  # noqa: UP017
+    return dt
+
+
+def object_dict_to_head(obj: dict[str, Any]) -> ObjectHead:
+    """Convert a list_objects entry into ObjectHead for compatibility uses."""
+    metadata = obj.get("metadata")
+    if metadata is None or not isinstance(metadata, dict):
+        metadata = {}
+
+    return ObjectHead(
+        key=obj["key"],
+        size=int(obj.get("size", 0)),
+        etag=str(obj.get("etag", "")),
+        last_modified=_parse_last_modified(obj.get("last_modified")),
+        metadata=metadata,
+    )
+
+
+__all__ = [
+    "ObjectListing",
+    "list_objects_page",
+    "list_all_objects",
+    "object_dict_to_head",
+]

src/deltaglider/core/s3_uri.py

@@ -0,0 +1,85 @@
+"""Utilities for working with S3-style URLs and keys."""
+
+from __future__ import annotations
+
+from typing import NamedTuple
+
+S3_SCHEME = "s3://"
+
+
+class S3Url(NamedTuple):
+    """Normalized representation of an S3 URL."""
+
+    bucket: str
+    key: str = ""
+
+    def to_url(self) -> str:
+        """Return the canonical string form."""
+        if self.key:
+            return f"{S3_SCHEME}{self.bucket}/{self.key}"
+        return f"{S3_SCHEME}{self.bucket}"
+
+    def with_key(self, key: str) -> S3Url:
+        """Return a new S3Url with a different key."""
+        return S3Url(self.bucket, key.lstrip("/"))
+
+    def join_key(self, suffix: str) -> S3Url:
+        """Append a suffix to the key using '/' semantics."""
+        suffix = suffix.lstrip("/")
+        if not self.key:
+            return self.with_key(suffix)
+        if not suffix:
+            return self
+        return self.with_key(f"{self.key.rstrip('/')}/{suffix}")
+
+
+def is_s3_url(value: str) -> bool:
+    """Check if a string is an S3 URL."""
+    return value.startswith(S3_SCHEME)
+
+
+def parse_s3_url(
+    url: str,
+    *,
+    allow_empty_key: bool = True,
+    strip_trailing_slash: bool = False,
+) -> S3Url:
+    """Parse an S3 URL into bucket and key components."""
+    if not is_s3_url(url):
+        raise ValueError(f"Invalid S3 URL: {url}")
+
+    path = url[len(S3_SCHEME) :]
+    if strip_trailing_slash:
+        path = path.rstrip("/")
+
+    bucket, sep, key = path.partition("/")
+    if not bucket:
+        raise ValueError(f"S3 URL missing bucket: {url}")
+
+    if not sep:
+        key = ""
+
+    key = key.lstrip("/")
+    if not key and not allow_empty_key:
+        raise ValueError(f"S3 URL must include a key: {url}")
+
+    return S3Url(bucket=bucket, key=key)
+
+
+def build_s3_url(bucket: str, key: str | None = None) -> str:
+    """Build an S3 URL from components."""
+    if not bucket:
+        raise ValueError("Bucket name cannot be empty")
+
+    if key:
+        key = key.lstrip("/")
+        return f"{S3_SCHEME}{bucket}/{key}"
+    return f"{S3_SCHEME}{bucket}"
+
+
+__all__ = [
+    "S3Url",
+    "build_s3_url",
+    "is_s3_url",
+    "parse_s3_url",
+]

src/deltaglider/core/service.py

@@ -15,6 +15,11 @@ from ..ports import (
     StoragePort,
 )
 from ..ports.storage import ObjectHead
+from .delta_extensions import (
+    DEFAULT_COMPOUND_DELTA_EXTENSIONS,
+    DEFAULT_DELTA_EXTENSIONS,
+    is_delta_candidate,
+)
 from .errors import (
     DiffDecodeError,
     DiffEncodeError,
@@ -58,39 +63,18 @@ class DeltaService:
         self.tool_version = tool_version
         self.max_ratio = max_ratio
 
-        # File extensions that should use delta compression
-        self.delta_extensions = {
-            ".zip",
-            ".tar",
-            ".gz",
-            ".tar.gz",
-            ".tgz",
-            ".bz2",
-            ".tar.bz2",
-            ".xz",
-            ".tar.xz",
-            ".7z",
-            ".rar",
-            ".dmg",
-            ".iso",
-            ".pkg",
-            ".deb",
-            ".rpm",
-            ".apk",
-            ".jar",
-            ".war",
-            ".ear",
-        }
+        # File extensions that should use delta compression. Keep mutable copies
+        # so advanced callers can customize the policy if needed.
+        self.delta_extensions = set(DEFAULT_DELTA_EXTENSIONS)
+        self.compound_delta_extensions = DEFAULT_COMPOUND_DELTA_EXTENSIONS
 
     def should_use_delta(self, filename: str) -> bool:
         """Check if file should use delta compression based on extension."""
-        name_lower = filename.lower()
-        # Check compound extensions first
-        for ext in [".tar.gz", ".tar.bz2", ".tar.xz"]:
-            if name_lower.endswith(ext):
-                return True
-        # Check simple extensions
-        return any(name_lower.endswith(ext) for ext in self.delta_extensions)
+        return is_delta_candidate(
+            filename,
+            simple_extensions=self.delta_extensions,
+            compound_extensions=self.compound_delta_extensions,
+        )
 
     def put(
         self,