☁️ Integrate Hugging Face Hub Cloud Storage

NEW FEATURE: AI can now write code directly to HF Hub!

MAJOR ADDITIONS:
- code/hub_storage.py: Complete cloud storage module
- HubStorage class with full CRUD operations
- Support for local, hub, and hybrid modes
- Automatic sync between local and cloud
- Dataset repo support for persistent Spaces storage
- Batch operations for efficiency
- File caching and metadata tracking

- Updated code/tools/fs.py: Transparent Hub integration
- read_file: Falls back to Hub if not local
- write_file: Auto-syncs to Hub after local write
- edit_file/multi_edit: Syncs changes back
- list_dir/glob/grep: Merges local + Hub files
- New functions: sync_to_hub, sync_from_hub, get_hub_status

- Updated code/server/routes.py: 11 new API endpoints
- /api/hub_status: Check Hub configuration status
- /api/hub_init: Initialize/configure Hub storage
- /api/hub_sync_push: Push files to Hub
- /api/hub_sync_pull: Pull files from Hub
- /api/hub_full_sync: Bidirectional sync
- /api/hub_list_files: List Hub files
- /api/hub_read_file: Read from Hub only
- /api/hub_write_file: Write to Hub directly (AI uses this!)
- /api/hub_delete_file: Delete from Hub
- /api/hub_create_repo: Create new repo on Hub
- /api/hub_stats: Storage statistics

- Updated app.py: Auto-initialization
- Detects HF Spaces environment automatically
- Uses SPACE_ID and HF_TOKEN env vars
- Creates dataset repo for workspace persistence
- Graceful shutdown syncs to Hub
- Logs storage mode on startup

HOW IT WORKS:
1. On HF Spaces: Auto-detects environment
2. Creates/uses dataset repo for persistence
3. All file ops transparently sync to Hub
4. AI can call hub_write_file() to persist code
5. Workspace survives Space restarts!

CONFIGURATION:
- Set HF_TOKEN env variable (auto-detected)
- Optional: HUB_DATASET_REPO for custom storage
- Optional: HUB_STORAGE_MODE (local/hub/hybrid/auto)

app.py

@@ -9,6 +9,8 @@ Enhanced version with:
 - Health check endpoint
 - Session management initialization
 - Model optimization on startup
+- **Hugging Face Hub cloud storage integration**
+- Auto-detection for HF Spaces environment
 
 Blog: https://huggingface.co/blog/introducing-gradio-server
 """
@@ -16,6 +18,7 @@ Blog: https://huggingface.co/blog/introducing-gradio-server
 from __future__ import annotations
 
 import logging
+import os
 import signal
 import sys
 import time
@@ -38,6 +41,17 @@ def setup_signal_handlers(app):
     """Setup graceful shutdown handlers."""
     def shutdown_handler(signum, frame):
         logger.info("Received shutdown signal, cleaning up...")
+        
+        # Sync to Hub before shutdown if enabled
+        try:
+            from code.hub_storage import is_hub_available, get_hub_storage
+            if is_hub_available():
+                hub = get_hub_storage()
+                hub.sync_to_hub(commit_message="Auto-sync before shutdown")
+                logger.info("Workspace synced to Hub before shutdown")
+        except Exception as e:
+            logger.warning("Failed to sync to Hub: %s", e)
+        
         try:
             # Save sessions
             from code.server.session_manager import get_session_manager
@@ -54,7 +68,7 @@ def setup_signal_handlers(app):
             _unload_model()
             logger.info("Model unloaded")
         except Exception as e:
-            logger.warning("Failed to unload model: %s", e)
+            logger.warning("Failed to unload model: %e", e)
         
         logger.info("Shutdown complete")
         sys.exit(0)
@@ -83,15 +97,107 @@ def initialize_session_manager():
         return None
 
 
+def initialize_hub_storage():
+    """
+    Initialize Hugging Face Hub cloud storage.
+    
+    Auto-detects environment:
+    - On HF Spaces: Uses SPACE_ID, HF_TOKEN env vars
+    - Local: Requires explicit configuration or HF_TOKEN
+    
+    Returns:
+        Initialized HubStorage instance or None if not configured.
+    """
+    try:
+        from code.hub_storage import init_hub_storage, is_hub_available
+        
+        # Get configuration from environment
+        token = os.environ.get("HF_TOKEN", "") or os.environ.get("HF_AUTH_TOKEN", "")
+        space_id = os.environ.get("SPACE_ID", "")
+        
+        # Determine mode based on environment
+        if space_id:
+            # Running on HF Spaces
+            logger.info("Detected HF Spaces environment: %s", space_id)
+            
+            # Extract repo_id from space_id (format: "user/space-name")
+            if space_id and token:
+                # Use a dataset repo for workspace persistence
+                # This allows the AI's code to persist across Space restarts
+                user_name = space_id.split("/")[0] if "/" in space_id else "unknown"
+                dataset_repo_id = os.environ.get(
+                    "HUB_DATASET_REPO",
+                    f"{user_name}/sonicoder-workspace"
+                )
+                
+                mode = os.environ.get("HUB_STORAGE_MODE", "hybrid")
+                
+                storage = init_hub_storage(
+                    token=token,
+                    repo_id=space_id,
+                    dataset_repo_id=dataset_repo_id,
+                    mode=mode,
+                )
+                
+                stats = storage.get_storage_stats()
+                logger.info(
+                    "Hub storage initialized: mode=%s, hub=%s, dataset=%s",
+                    stats["mode"],
+                    stats["hub_enabled"],
+                    stats["dataset_enabled"]
+                )
+                
+                return storage
+            else:
+                logger.warning("HF Spaces detected but no token provided, using local storage only")
+                return None
+        elif token:
+            # Local mode with token
+            repo_id = os.environ.get("HUB_REPO_ID", "")
+            dataset_repo_id = os.environ.get("HUB_DATASET_REPO", "")
+            
+            if repo_id or dataset_repo_id:
+                storage = init_hub_storage(
+                    token=token,
+                    repo_id=repo_id,
+                    dataset_repo_id=dataset_repo_id,
+                    mode="hybrid",
+                )
+                
+                logger.info("Hub storage initialized in local mode")
+                return storage
+        
+        # No configuration available
+        logger.info("Hub storage not configured (set HF_TOKEN to enable)")
+        return None
+        
+    except ImportError:
+        logger.debug("huggingface_hub not installed, skipping Hub storage")
+        return None
+    except Exception as exc:
+        logger.warning("Hub storage initialization failed: %s", exc)
+        return None
+
+
 def main():
     start_time = time.time()
     logger.info("=" * 50)
     logger.info("SoniCoder v%s starting...", APP_VERSION)
     logger.info("=" * 50)
 
+    # Detect and log environment
+    space_id = os.environ.get("SPACE_ID", None)
+    if space_id:
+        logger.info("Running on Hugging Face Spaces: %s", space_id)
+    else:
+        logger.info("Running in local mode")
+
     # Initialize session manager
     session_mgr = initialize_session_manager()
 
+    # Initialize Hub storage (cloud persistence)
+    hub_storage = initialize_hub_storage()
+
     # Start loading default model in background
     from code.model.loader import start_background_load, add_progress_callback
 
@@ -137,6 +243,13 @@ def main():
     # Log startup time
     startup_time = time.time() - start_time
     logger.info("Startup completed in %.2f seconds", startup_time)
+    
+    # Log storage info
+    if hub_storage:
+        logger.info("Cloud storage: ENABLED (files will persist on HF Hub)")
+    else:
+        logger.info("Cloud storage: DISABLED (local files only)")
+    
     logger.info("Launching...")
 
     # Launch the server

code/hub_storage.py

@@ -0,0 +1,933 @@
+"""Hugging Face Hub Cloud Storage Integration.
+
+This module provides seamless integration with Hugging Face Hub for:
+- Cloud-based file storage and retrieval
+- Persistent workspace on HF Spaces
+- Direct code writing to repositories
+- Dataset-backed persistence
+
+Architecture:
+┌─────────────┐     ┌──────────────────┐     ┌─────────────────┐
+│   AI Agent  │────▶│  hub_storage.py   │────▶│  HuggingFace Hub │
+│             │     │  (this module)    │     │  (Cloud Storage) │
+└─────────────┘     └──────────────────┘     └─────────────────┘
+                            │
+                            ▼
+                   ┌──────────────────┐
+                   │  Local Fallback  │
+                   │  (when offline)  │
+                   └──────────────────┘
+
+Usage Modes:
+1. LOCAL: Files stored in ./workspace/ (default)
+2. HUB: Files synced to HF Hub repository
+3. HYBRID: Local + Hub sync (recommended for Spaces)
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import tempfile
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Callable, Optional, Iterator
+from contextlib import contextmanager
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class HubConfig:
+    """Configuration for HF Hub storage."""
+    
+    # Repository settings
+    repo_id: str = ""  # e.g., "username/repo-name" or "username/workspace-data"
+    repo_type: str = "space"  # "space", "model", or "dataset"
+    
+    # Authentication
+    token: str = ""
+    
+    # Storage mode
+    mode: str = "local"  # "local", "hub", or "hybrid"
+    
+    # Workspace path (for local/hybrid)
+    local_workspace: str = "./workspace"
+    
+    # Sync settings
+    auto_sync: bool = True
+    sync_interval: int = 60  # seconds
+    
+    # Dataset repo for persistent storage (recommended for Spaces)
+    dataset_repo_id: str = ""  # e.g., "username/sonicoder-workspace"
+    
+    def is_hub_enabled(self) -> bool:
+        """Check if Hub storage is enabled."""
+        return self.mode in ("hub", "hybrid") and bool(self.repo_id) and bool(self.token)
+    
+    def is_dataset_storage_enabled(self) -> bool:
+        """Check if dataset-based storage is enabled."""
+        return bool(self.dataset_repo_id) and bool(self.token)
+
+
+@dataclass 
+class FileMetadata:
+    """Metadata for a file in storage."""
+    path: str
+    size: int = 0
+    modified_at: float = field(default_factory=time.time)
+    created_at: float = field(default_factory=time.time)
+    content_hash: str = ""
+    is_local: bool = True
+    is_remote: bool = False
+
+
+class HubStorageError(Exception):
+    """Custom exception for Hub storage operations."""
+    
+    def __init__(self, message: str, operation: str = "", recoverable: bool = True):
+        self.message = message
+        self.operation = operation
+        self.recoverable = recoverable
+        super().__init__(f"[{operation}] {message}")
+
+
+class HubStorage:
+    """
+    Main class for Hugging Face Hub cloud storage integration.
+    
+    Provides transparent file operations that work with both
+    local filesystem and HF Hub, with automatic sync.
+    
+    Example:
+        >>> storage = HubStorage(
+        ...     repo_id="sonic-coder/sonicoder",
+        ...     token="hf_...",
+        ...     mode="hybrid",
+        ...     dataset_repo_id="sonic-coder/sonicoder-workspace"
+        ... )
+        >>> storage.initialize()
+        >>> storage.write_file("app.py", "print('Hello!')")
+        >>> content = storage.read_file("app.py")
+        >>> files = storage.list_files()
+    """
+    
+    def __init__(
+        self,
+        config: Optional[HubConfig] = None,
+        **kwargs
+    ):
+        self.config = config or HubConfig(**kwargs)
+        self._api = None
+        self._initialized = False
+        self._local_workspace: Path = Path(self.config.local_workspace)
+        self._sync_lock_needed = False
+        
+        # Cache for remote files
+        self._remote_cache: dict[str, FileMetadata] = {}
+        self._cache_timestamp: float = 0
+        self._cache_ttl: int = 30  # seconds
+        
+        # Operation callbacks
+        self._on_sync: Optional[Callable] = None
+        self._on_error: Optional[Callable] = None
+        
+    @property
+    def api(self):
+        """Lazy initialization of HfApi."""
+        if self._api is None:
+            try:
+                from huggingface_hub import HfApi
+                self._api = HfApi(token=self.config.token)
+            except ImportError:
+                raise HubStorageError(
+                    "huggingface_hub package not installed",
+                    operation="initialize",
+                    recoverable=False
+                )
+        return self._api
+    
+    def initialize(self) -> dict[str, Any]:
+        """
+        Initialize the Hub storage system.
+        
+        Creates necessary directories, validates connection,
+        and sets up the storage backend.
+        
+        Returns:
+            Dict with initialization status and info.
+        """
+        result = {
+            "success": False,
+            "mode": self.config.mode,
+            "hub_enabled": False,
+            "dataset_enabled": False,
+            "workspace_path": str(self._local_workspace),
+        }
+        
+        try:
+            # Create local workspace directory
+            self._local_workspace.mkdir(parents=True, exist_ok=True)
+            
+            # Validate Hub connection if enabled
+            if self.config.is_hub_enabled():
+                try:
+                    # Test API connection by getting user info
+                    user_info = self.api.whoami()
+                    result["hub_enabled"] = True
+                    result["user"] = user_info.get("name", "unknown")
+                    logger.info("Connected to HF Hub as: %s", result["user"])
+                    
+                    # Ensure repo exists
+                    self._ensure_repo_exists()
+                    
+                except Exception as e:
+                    logger.warning("Hub connection failed: %s", e)
+                    result["hub_error"] = str(e)
+                    # Fall back to local mode
+                    self.config.mode = "local"
+            
+            # Check dataset storage
+            if self.config.is_dataset_storage_enabled():
+                try:
+                    self._ensure_dataset_repo_exists()
+                    result["dataset_enabled"] = True
+                    logger.info("Dataset storage ready: %s", self.config.dataset_repo_id)
+                except Exception as e:
+                    logger.warning("Dataset setup failed: %s", e)
+                    result["dataset_error"] = str(e)
+            
+            self._initialized = True
+            result["success"] = True
+            
+            logger.info(
+                "HubStorage initialized: mode=%s, hub=%s, dataset=%s",
+                self.config.mode,
+                result["hub_enabled"],
+                result["dataset_enabled"]
+            )
+            
+        except Exception as e:
+            logger.exception("Failed to initialize HubStorage")
+            result["error"] = str(e)
+        
+        return result
+    
+    def _ensure_repo_exists(self) -> None:
+        """Ensure the target repository exists."""
+        try:
+            self.api.repo_info(
+                repo_id=self.config.repo_id,
+                repo_type=self.config.repo_type
+            )
+        except Exception:
+            # Repo doesn't exist, create it
+            logger.info("Creating repo: %s", self.config.repo_id)
+            self.api.create_repo(
+                repo_id=self.config.repo_id,
+                repo_type=self.config.repo_type,
+                exist_ok=True,
+                private=False
+            )
+    
+    def _ensure_dataset_repo_exists(self) -> None:
+        """Ensure the dataset repository exists for persistent storage."""
+        try:
+            self.api.repo_info(
+                repo_id=self.config.dataset_repo_id,
+                repo_type="dataset"
+            )
+        except Exception:
+            logger.info("Creating dataset repo: %s", self.config.dataset_repo_id)
+            self.api.create_repo(
+                repo_id=self.config.dataset_repo_id,
+                repo_type="dataset",
+                exist_ok=True,
+                private=True  # Workspaces are usually private
+            )
+    
+    # ─── File Operations ──────────────────────────────────────────────
+    
+    def write_file(
+        self,
+        path: str,
+        content: str | bytes,
+        commit_message: str | None = None,
+        sync_to_hub: bool = True
+    ) -> FileMetadata:
+        """
+        Write a file to storage.
+        
+        In hybrid/local mode, writes to local filesystem first.
+        If sync_to_hub and hub is enabled, also uploads to HF Hub.
+        
+        Args:
+            path: Relative file path (e.g., "src/app.py")
+            content: File content (string or bytes)
+            commit_message: Optional git commit message
+            sync_to_hub: Whether to sync to Hub immediately
+            
+        Returns:
+            FileMetadata with file information.
+        """
+        # Normalize path
+        path = self._normalize_path(path)
+        
+        # Generate commit message if not provided
+        if not commit_message:
+            commit_message = f"Update {path}"
+        
+        # Write to local filesystem first
+        local_path = self._local_workspace / path
+        local_path.parent.mkdir(parents=True, exist_ok=True)
+        
+        if isinstance(content, bytes):
+            local_path.write_bytes(content)
+        else:
+            local_path.write_text(content, encoding='utf-8')
+        
+        # Create metadata
+        metadata = FileMetadata(
+            path=path,
+            size=local_path.stat().st_size,
+            is_local=True
+        )
+        
+        # Sync to Hub if enabled
+        if sync_to_hub and self._should_sync_to_hub():
+            try:
+                metadata = self._upload_to_hub(path, content, commit_message)
+                metadata.is_remote = True
+            except Exception as e:
+                logger.warning("Failed to sync %s to Hub: %s", path, e)
+                if self._on_error:
+                    self._on_error("write", path, e)
+        
+        return metadata
+    
+    def read_file(self, path: str, use_cache: bool = True) -> str:
+        """
+        Read a file from storage.
+        
+        Tries local first, then falls back to Hub if available.
+        
+        Args:
+            path: Relative file path
+            use_cache: Whether to use cached remote file list
+            
+        Returns:
+            File content as string.
+            
+        Raises:
+            HubStorageError if file not found anywhere.
+        """
+        path = self._normalize_path(path)
+        
+        # Try local first
+        local_path = self._local_workspace / path
+        if local_path.exists():
+            return local_path.read_text(encoding='utf-8')
+        
+        # Try Hub if enabled
+        if self.config.is_hub_enabled() or self.config.is_dataset_storage_enabled():
+            try:
+                return self._download_from_hub(path)
+            except Exception as e:
+                logger.debug("Could not download %s from Hub: %s", path, e)
+        
+        raise HubStorageError(
+            f"File not found: {path}",
+            operation="read_file",
+            recoverable=False
+        )
+    
+    def read_binary(self, path: str) -> bytes:
+        """Read a file as binary."""
+        path = self._normalize_path(path)
+        
+        local_path = self._local_workspace / path
+        if local_path.exists():
+            return local_path.read_bytes()
+        
+        if self.config.is_hub_enabled():
+            # Download to temp and read
+            with tempfile.NamedTemporaryFile(delete=False) as tmp:
+                tmp_path = tmp.name
+            
+            try:
+                self.api.hub_download(
+                    repo_id=self._get_target_repo(),
+                    filename=path,
+                    repo_type=self._get_target_repo_type(),
+                    local_dir=os.path.dirname(tmp_path),
+                )
+                
+                # Find downloaded file
+                downloaded = Path(tmp_path).parent / Path(path).name
+                if downloaded.exists():
+                    return downloaded.read_bytes()
+            finally:
+                # Cleanup temp
+                if os.path.exists(tmp_path):
+                    os.unlink(tmp_path)
+        
+        raise HubStorageError(f"Binary file not found: {path}", operation="read_binary")
+    
+    def delete_file(self, path: str, commit_message: str | None = None) -> bool:
+        """
+        Delete a file from storage.
+        
+        Args:
+            path: Relative file path
+            commit_message: Optional commit message
+            
+        Returns:
+            True if deleted successfully.
+        """
+        path = self._normalize_path(path)
+        
+        if not commit_message:
+            commit_message = f"Delete {path}"
+        
+        success = True
+        
+        # Delete from local
+        local_path = self._local_workspace / path
+        if local_path.exists():
+            local_path.unlink()
+            # Remove empty parent directories
+            self._cleanup_empty_dirs(local_path.parent)
+        
+        # Delete from Hub
+        if self._should_sync_to_hub():
+            try:
+                self.api.delete_file(
+                    path_in_repo=path,
+                    repo_id=self._get_target_repo(),
+                    repo_type=self._get_target_repo_type(),
+                    commit_message=commit_message
+                )
+            except Exception as e:
+                logger.warning("Failed to delete %s from Hub: %s", path, e)
+                success = False
+        
+        return success
+    
+    def file_exists(self, path: str) -> bool:
+        """Check if a file exists in storage."""
+        path = self._normalize_path(path)
+        
+        # Check local
+        if (self._local_workspace / path).exists():
+            return True
+        
+        # Check Hub cache or remote
+        if self.config.is_hub_enabled():
+            try:
+                files = self.list_files(use_cache=True)
+                return any(f["path"] == path for f in files)
+            except Exception:
+                pass
+        
+        return False
+    
+    def list_files(
+        self,
+        path: str = ".",
+        recursive: bool = True,
+        use_cache: bool = True
+    ) -> list[dict[str, Any]]:
+        """
+        List files in storage.
+        
+        Combines local and remote file listings.
+        
+        Args:
+            path: Directory path to list (default: root)
+            recursive: Whether to list recursively
+            use_cache: Whether to use cached file listing
+            
+        Returns:
+            List of file info dicts with 'path', 'size', 'modified' keys.
+        """
+        files = []
+        seen_paths = set()
+        
+        # List local files
+        base_path = self._local_workspace / path
+        if base_path.exists():
+            for file_path in base_path.rglob("*") if recursive else base_path.iterdir():
+                if file_path.is_file():
+                    rel_path = str(file_path.relative_to(self._local_workspace))
+                    stat = file_path.stat()
+                    files.append({
+                        "path": rel_path.replace(os.sep, "/"),
+                        "size": stat.st_size,
+                        "modified": stat.st_mtime,
+                        "is_local": True,
+                        "is_remote": False,
+                    })
+                    seen_paths.add(rel_path)
+        
+        # List remote files if hub enabled
+        if self.config.is_hub_enabled() or self.config.is_dataset_storage_enabled():
+            try:
+                remote_files = self._list_hub_files(use_cache=use_cache)
+                for rf in remote_files:
+                    if rf["path"] not in seen_paths:
+                        rf["is_local"] = False
+                        rf["is_remote"] = True
+                        files.append(rf)
+            except Exception as e:
+                logger.warning("Failed to list remote files: %e", e)
+        
+        # Sort by path
+        files.sort(key=lambda x: x["path"])
+        
+        return files
+    
+    # ─── Hub Operations ────────────────────────────────────────────────
+    
+    def _should_sync_to_hub(self) -> bool:
+        """Check if we should sync to Hub based on config."""
+        return (
+            self.config.is_hub_enabled() or 
+            self.config.is_dataset_storage_enabled()
+        ) and self.config.auto_sync
+    
+    def _get_target_repo(self) -> str:
+        """Get the target repository ID for operations."""
+        # Prefer dataset repo for workspace files
+        if self.config.is_dataset_storage_enabled():
+            return self.config.dataset_repo_id
+        return self.config.repo_id
+    
+    def _get_target_repo_type(self) -> str:
+        """Get the target repository type."""
+        if self.config.is_dataset_storage_enabled():
+            return "dataset"
+        return self.config.repo_type
+    
+    def _upload_to_hub(
+        self,
+        path: str,
+        content: str | bytes,
+        commit_message: str
+    ) -> FileMetadata:
+        """Upload a file to HF Hub."""
+        
+        # For large content, use temp file
+        if isinstance(content, str) and len(content) > 100000:  # > 100KB
+            with tempfile.NamedTemporaryFile(
+                mode='w',
+                suffix=Path(path).suffix,
+                delete=False,
+                encoding='utf-8'
+            ) as tmp:
+                tmp.write(content)
+                tmp_path = tmp.name
+            
+            try:
+                self.api.upload_file(
+                    path_or_fileobj=tmp_path,
+                    path_in_repo=path,
+                    repo_id=self._get_target_repo(),
+                    repo_type=self._get_target_repo_type(),
+                    commit_message=commit_message
+                )
+            finally:
+                os.unlink(tmp_path)
+        else:
+            # Upload directly from content
+            content_bytes = content.encode('utf-8') if isinstance(content, str) else content
+            
+            from io import BytesIO
+            self.api.upload_file(
+                path_or_fileobj=BytesIO(content_bytes),
+                path_in_repo=path,
+                repo_id=self._get_target_repo(),
+                repo_type=self._get_target_repo_type(),
+                commit_message=commit_message
+            )
+        
+        logger.info("Uploaded %s to Hub (%s)", path, self._get_target_repo())
+        
+        return FileMetadata(
+            path=path,
+            size=len(content) if isinstance(content, str) else len(content_bytes),
+            is_remote=True,
+            is_local=(self._local_workspace / path).exists()
+        )
+    
+    def _download_from_hub(self, path: str) -> str:
+        """Download a file from HF Hub."""
+        
+        # Use hf_hub_download for caching
+        from huggingface_hub import hf_hub_download
+        
+        local_path = hf_hub_download(
+            repo_id=self._get_target_repo(),
+            filename=path,
+            repo_type=self._get_target_repo_type(),
+            token=self.config.token
+        )
+        
+        # Read and return content
+        with open(local_path, 'r', encoding='utf-8') as f:
+            return f.read()
+    
+    def _list_hub_files(
+        self,
+        use_cache: bool = True
+    ) -> list[dict[str, Any]]:
+        """List files in the Hub repository."""
+        
+        # Check cache
+        if use_cache and self._remote_cache:
+            if time.time() - self._cache_timestamp < self._cache_ttl:
+                return list(self._remote_cache.values())
+        
+        # Fetch from API
+        repo_info = self.api.repo_info(
+            repo_id=self._get_target_repo(),
+            repo_type=self._get_target_repo_type()
+        )
+        
+        files = []
+        siblings = getattr(repo_info, 'siblings', [])
+        
+        for sibling in siblings:
+            if hasattr(sibling, 'rfilename'):
+                files.append({
+                    "path": sibling.rfilename,
+                    "size": getattr(sibling, 'size', 0),
+                    "blob_id": getattr(sibling, 'blob_id', ''),
+                    "is_local": False,
+                    "is_remote": True,
+                })
+        
+        # Update cache
+        self._remote_cache = {f["path"]: f for f in files}
+        self._cache_timestamp = time.time()
+        
+        return files
+    
+    # ─── Sync Operations ──────────────────────────────────────────────
+    
+    def sync_to_hub(
+        self,
+        path: str | None = None,
+        commit_message: str = "Sync workspace"
+    ) -> dict[str, Any]:
+        """
+        Sync local files to Hub.
+        
+        Args:
+            path: Specific file or directory to sync (None = all)
+            commit_message: Commit message for changes
+            
+        Returns:
+            Dict with sync results.
+        """
+        result = {
+            "success": False,
+            "uploaded": [],
+            "skipped": [],
+            "errors": []
+        }
+        
+        if not self.config.is_hub_enabled() and not self.config.is_dataset_storage_enabled():
+            result["errors"].append("Hub storage not configured")
+            return result
+        
+        try:
+            files_to_sync = []
+            
+            if path:
+                # Sync specific file/directory
+                full_path = self._local_workspace / path
+                if full_path.is_file():
+                    files_to_sync.append(path)
+                elif full_path.is_dir():
+                    for fp in full_path.rglob("*"):
+                        if fp.is_file():
+                            files_to_sync.append(str(fp.relative_to(self._local_workspace)))
+            else:
+                # Sync all local files
+                for fp in self._local_workspace.rglob("*"):
+                    if fp.is_file():
+                        rel = str(fp.relative_to(self._local_workspace))
+                        files_to_sync.append(rel)
+            
+            # Upload each file
+            for file_path in files_to_sync:
+                try:
+                    local_path = self._local_workspace / file_path
+                    content = local_path.read_text(encoding='utf-8')
+                    
+                    self._upload_to_hub(
+                        file_path,
+                        content,
+                        f"{commit_message}: {file_path}"
+                    )
+                    
+                    result["uploaded"].append(file_path)
+                    
+                except Exception as e:
+                    logger.error("Failed to sync %s: %s", file_path, e)
+                    result["errors"].append({"file": file_path, "error": str(e)})
+            
+            result["success"] = len(result["errors"]) == 0
+            
+            if self._on_sync:
+                self._on_sync(result)
+                
+        except Exception as e:
+            result["errors"].append({"file": "*", "error": str(e)})
+            logger.exception("Sync failed")
+        
+        return result
+    
+    def sync_from_hub(
+        self,
+        path: str | None = None
+    ) -> dict[str, Any]:
+        """
+        Sync files from Hub to local.
+        
+        Args:
+            path: Specific file to download (None = all)
+            
+        Returns:
+            Dict with sync results.
+        """
+        result = {
+            "success": False,
+            "downloaded": [],
+            "errors": []
+        }
+        
+        try:
+            if path:
+                # Download specific file
+                content = self._download_from_hub(path)
+                local_path = self._local_workspace / path
+                local_path.parent.mkdir(parents=True, exist_ok=True)
+                local_path.write_text(content, encoding='utf-8')
+                result["downloaded"].append(path)
+            else:
+                # Download all files
+                remote_files = self._list_hub_files(use_cache=False)
+                
+                for rf in remote_files:
+                    try:
+                        content = self._download_from_hub(rf["path"])
+                        local_path = self._local_workspace / rf["path"]
+                        local_path.parent.mkdir(parents=True, exist_ok=True)
+                        local_path.write_text(content, encoding='utf-8')
+                        result["downloaded"].append(rf["path"])
+                    except Exception as e:
+                        result["errors"].append({
+                            "file": rf["path"],
+                            "error": str(e)
+                        })
+            
+            result["success"] = len(result["errors"]) == len(remote_files) == 0 or len(remote_files) == 0
+            
+        except Exception as e:
+            result["errors"].append({"file": "*", "error": str(e)})
+            logger.exception("Sync from hub failed")
+        
+        return result
+    
+    def full_sync(self) -> dict[str, Any]:
+        """
+        Perform bidirectional sync between local and Hub.
+        
+        Merges changes from both sides.
+        """
+        result = {
+            "to_hub": {},
+            "from_hub": {}
+        }
+        
+        # Push local to hub
+        result["to_hub"] = self.sync_to_hub(commit_message="Full sync push")
+        
+        # Pull new/updated from hub
+        result["from_hub"] = self.sync_from_hub()
+        
+        return result
+    
+    # ─── Utility Methods ──────────────────────────────────────────────
+    
+    def _normalize_path(self, path: str) -> str:
+        """Normalize a file path."""
+        # Remove leading ./
+        if path.startswith("./"):
+            path = path[2:]
+        # Convert backslashes
+        path = path.replace("\\", "/")
+        # Remove double slashes
+        while "//" in path:
+            path = path.replace("//", "/")
+        # Security: prevent path traversal
+        if ".." in path:
+            raise HubStorageError(
+                "Path traversal detected",
+                operation="_normalize_path",
+                recoverable=False
+            )
+        return path.lstrip("/")
+    
+    def _cleanup_empty_dirs(self, dir_path: Path) -> None:
+        """Remove empty directories up to workspace root."""
+        try:
+            while dir_path != self._local_workspace:
+                if dir_path.is_dir() and not any(dir_path.iterdir()):
+                    dir_path.rmdir()
+                    dir_path = dir_path.parent
+                else:
+                    break
+        except Exception:
+            pass
+    
+    def get_storage_stats(self) -> dict[str, Any]:
+        """Get statistics about current storage."""
+        stats = {
+            "mode": self.config.mode,
+            "hub_enabled": self.config.is_hub_enabled(),
+            "dataset_enabled": self.config.is_dataset_storage_enabled(),
+            "local_files": 0,
+            "local_size_mb": 0,
+            "last_sync": self._cache_timestamp,
+        }
+        
+        # Count local files
+        if self._local_workspace.exists():
+            for f in self._local_workspace.rglob("*"):
+                if f.is_file():
+                    stats["local_files"] += 1
+                    stats["local_size_mb"] += f.stat().st_size / (1024 * 1024)
+        
+        stats["local_size_mb"] = round(stats["local_size_mb"], 2)
+        
+        return stats
+    
+    def clear_cache(self) -> None:
+        """Clear the remote file cache."""
+        self._remote_cache = {}
+        self._cache_timestamp = 0
+    
+    def set_callbacks(
+        self,
+        on_sync: Callable | None = None,
+        on_error: Callable | None = None
+    ) -> None:
+        """Set callback functions for events."""
+        self._on_sync = on_sync
+        self._on_error = on_error
+    
+    @contextmanager
+    def batch_operation(self, commit_message: str = "Batch update"):
+        """
+        Context manager for batch operations.
+        
+        All writes within the context are grouped into one commit.
+        
+        Usage:
+            with storage.batch_operation("Create API"):
+                storage.write_file("app.py", "...")
+                storage.write_file("config.py", "...")
+        """
+        original_auto_sync = self.config.auto_sync
+        self.config.auto_sync = False
+        
+        files_written = []
+        
+        try:
+            yield files_written
+            
+            # Commit all at once
+            if files_written:
+                self.sync_to_hub(commit_message=commit_message)
+        finally:
+            self.config.auto_sync = original_auto_sync
+
+
+# ─── Global Instance Management ────────────────────────────────────────
+
+_global_storage: Optional[HubStorage] = None
+
+
+def get_hub_storage(**kwargs) -> HubStorage:
+    """Get or create the global HubStorage instance."""
+    global _global_storage
+    if _global_storage is None:
+        _global_storage = HubStorage(**kwargs)
+    return _global_storage
+
+
+def init_hub_storage(
+    token: str = "",
+    repo_id: str = "",
+    dataset_repo_id: str = "",
+    mode: str = "auto",
+    **kwargs
+) -> HubStorage:
+    """
+    Initialize the global HubStorage instance.
+    
+    Auto-detects environment when running on HF Spaces.
+    
+    Args:
+        token: HF API token
+        repo_id: Main repository ID
+        dataset_repo_id: Dataset repository for workspace storage
+        mode: Storage mode ("local", "hub", "hybrid", "auto")
+        
+    Returns:
+        Initialized HubStorage instance.
+    """
+    global _global_storage
+    
+    # Auto-detect mode
+    if mode == "auto":
+        # Check if running on HF Spaces
+        if os.environ.get("SPACE_ID"):
+            mode = "hybrid" if token else "local"
+        else:
+            mode = "local"
+    
+    # Get token from env if not provided
+    if not token:
+        token = os.environ.get("HF_TOKEN", "") or os.environ.get("HF_AUTH_TOKEN", "")
+    
+    config = HubConfig(
+        token=token,
+        repo_id=repo_id,
+        dataset_repo_id=dataset_repo_id,
+        mode=mode,
+        **kwargs
+    )
+    
+    _global_storage = HubStorage(config=config)
+    _global_storage.initialize()
+    
+    return _global_storage
+
+
+def is_hub_available() -> bool:
+    """Check if Hub storage is available and configured."""
+    if _global_storage is None:
+        return False
+    return _global_storage.config.is_hub_enabled()
+
+
+def get_storage_mode() -> str:
+    """Get current storage mode."""
+    if _global_storage is None:
+        return "not_initialized"
+    return _global_storage.config.mode

code/server/routes.py

@@ -1662,3 +1662,360 @@ def handle_push_github(
         timeout=timeout_int,
     )
     yield json.dumps(result, default=str)
+
+
+# ═══════════════════════════════════════════════════════════════════════
+# HUB STORAGE API ENDPOINTS
+# ═══════════════════════════════════════════════════════════════════════
+
+@app.api(name="hub_status")
+def handle_hub_status() -> str:
+    """Get current Hub storage status and configuration.
+    
+    Returns:
+        JSON dict with: available, configured, mode, hub_enabled, 
+        dataset_enabled, local_files, local_size_mb, etc.
+    """
+    from code.tools.fs import get_hub_status
+    
+    result = get_hub_status()
+    yield json.dumps(result, default=str)
+
+
+@app.api(name="hub_init")
+def handle_hub_init(
+    token: str = "",
+    repo_id: str = "",
+    dataset_repo_id: str = "",
+    mode: str = "auto",
+) -> str:
+    """Initialize/configure Hub storage.
+    
+    This sets up the connection to Hugging Face Hub for cloud storage.
+    
+    Args:
+        token: HF API token (or use HF_TOKEN env var)
+        repo_id: Main repository ID (e.g., "user/repo")
+        dataset_repo_id: Dataset repo for workspace persistence
+        mode: "local", "hub", "hybrid", or "auto"
+        
+    Returns:
+        JSON dict with initialization status and config.
+    """
+    from code.hub_storage import init_hub_storage
+    
+    try:
+        storage = init_hub_storage(
+            token=token,
+            repo_id=repo_id,
+            dataset_repo_id=dataset_repo_id,
+            mode=mode,
+        )
+        
+        result = {
+            "success": True,
+            "message": "Hub storage initialized",
+            "config": {
+                "mode": storage.config.mode,
+                "repo_id": storage.config.repo_id,
+                "dataset_repo_id": storage.config.dataset_repo_id,
+                "hub_enabled": storage.config.is_hub_enabled(),
+                "dataset_enabled": storage.config.is_dataset_storage_enabled(),
+            },
+            "stats": storage.get_storage_stats(),
+        }
+    except Exception as e:
+        result = {
+            "success": False,
+            "error": str(e),
+        }
+    
+    yield json.dumps(result, default=str)
+
+
+@app.api(name="hub_sync_push", concurrency_limit=1)
+def handle_hub_sync_push(path: str = "") -> str:
+    """Sync local workspace files to HF Hub.
+    
+    Args:
+        path: Optional specific file/directory to sync
+        
+    Returns:
+        JSON dict with sync results.
+    """
+    from code.tools.fs import sync_to_hub
+    
+    try:
+        result = sync_to_hub(path=path if path else None)
+        yield json.dumps(result, default=str)
+    except Exception as e:
+        yield json.dumps({"success": False, "error": str(e)}, default=str)
+
+
+@app.api(name="hub_sync_pull", concurrency_limit=1)
+def handle_hub_sync_pull(path: str = "") -> str:
+    """Sync files from HF Hub to local workspace.
+    
+    Args:
+        path: Optional specific file to download
+        
+    Returns:
+        JSON dict with sync results.
+    """
+    from code.tools.fs import sync_from_hub
+    
+    try:
+        result = sync_from_hub(path=path if path else None)
+        yield json.dumps(result, default=str)
+    except Exception as e:
+        yield json.dumps({"success": False, "error": str(e)}, default=str)
+
+
+@app.api(name="hub_full_sync", concurrency_limit=1)
+def handle_hub_full_sync() -> str:
+    """Perform bidirectional sync between local and Hub.
+    
+    Merges changes from both sides.
+    
+    Returns:
+        JSON dict with both push and pull results.
+    """
+    from code.hub_storage import get_hub_storage, is_hub_available
+    
+    if not is_hub_available():
+        yield json.dumps({
+            "success": False,
+            "error": "Hub storage not configured"
+        }, default=str)
+        return
+    
+    try:
+        hub = get_hub_storage()
+        result = hub.full_sync()
+        yield json.dumps(result, default=str)
+    except Exception as e:
+        yield json.dumps({"success": False, "error": str(e)}, default=str)
+
+
+@app.api(name="hub_list_files")
+def handle_hub_list_files(path: str = ".", recursive: bool = True) -> str:
+    """List files in Hub storage.
+    
+    Args:
+        path: Directory path to list
+        recursive: Whether to list recursively
+        
+    Returns:
+        JSON dict with file listing.
+    """
+    from code.hub_storage import get_hub_storage, is_hub_available
+    
+    if not is_hub_available():
+        yield json.dumps({
+            "success": False,
+            "error": "Hub storage not configured",
+            "files": []
+        }, default_str=str)
+        return
+    
+    try:
+        hub = get_hub_storage()
+        files = hub.list_files(path=path, recursive=recursive)
+        yield json.dumps({
+            "success": True,
+            "files": files,
+            "count": len(files),
+        }, default=str)
+    except Exception as e:
+        yield json.dumps({"success": False, "error": str(e), "files": []}, default=str)
+
+
+@app.api(name="hub_delete_file")
+def handle_hub_delete_file(path: str) -> str:
+    """Delete a file from Hub storage.
+    
+    Args:
+        path: File path to delete
+        
+    Returns:
+        JSON dict with operation result.
+    """
+    from code.tools.fs import delete_from_hub
+    
+    try:
+        result = delete_from_hub(path)
+        yield json.dumps(result, default=str)
+    except Exception as e:
+        yield json.dumps({"success": False, "error": str(e)}, default=str)
+
+
+@app.api(name="hub_read_file")
+def handle_hub_read_file(path: str) -> str:
+    """Read a file specifically from Hub storage.
+    
+    Args:
+        path: File path to read
+        
+    Returns:
+        JSON dict with file content.
+    """
+    from code.hub_storage import get_hub_storage, is_hub_available
+    
+    if not is_hub_available():
+        yield json.dumps({
+            "success": False,
+            "error": "Hub storage not configured"
+        }, default_str=str)
+        return
+    
+    try:
+        hub = get_hub_storage()
+        content = hub.read_file(path)
+        yield json.dumps({
+            "success": True,
+            "path": path,
+            "content": content,
+            "size": len(content),
+        }, default_str=str)
+    except Exception as e:
+        yield json.dumps({"success": False, "error": str(e)}, default_str=str)
+
+
+@app.api(name="hub_write_file")
+def handle_hub_write_file(path: str, content: str, commit_message: str = "") -> str:
+    """Write a file directly to Hub storage.
+    
+    This allows the AI to write code that persists on HF Hub.
+    
+    Args:
+        path: Destination file path (e.g., "src/app.py")
+        content: File content
+        commit_message: Optional commit message
+        
+    Returns:
+        JSON dict with operation result and metadata.
+    """
+    from code.hub_storage import get_hub_storage, is_hub_available
+    
+    if not is_hub_available():
+        yield json.dumps({
+            "success": False,
+            "error": "Hub storage not configured. Initialize with /hub_init first."
+        }, default_str=str)
+        return
+    
+    try:
+        hub = get_hub_storage()
+        
+        # Use a descriptive commit message if not provided
+        if not commit_message:
+            commit_message = f"AI generated: {path}"
+        
+        metadata = hub.write_file(
+            path=path,
+            content=content,
+            commit_message=commit_message,
+            sync_to_hub=True  # Force immediate upload
+        )
+        
+        yield json.dumps({
+            "success": True,
+            "path": path,
+            "size": metadata.size,
+            "is_remote": metadata.is_remote,
+            "is_local": metadata.is_local,
+            "message": f"File written to {'Hub' if metadata.is_remote else 'local'} storage",
+        }, default_str=str)
+        
+    except Exception as e:
+        yield json.dumps({"success": False, "error": str(e)}, default_str=str)
+
+
+@app.api(name="hub_create_repo")
+def handle_hub_create_repo(
+    repo_id: str,
+    repo_type: str = "dataset",
+    private: bool = True,
+    token: str = "",
+) -> str:
+    """Create a new repository on HF Hub for storage.
+    
+    Args:
+        repo_id: Repository ID (e.g., "username/my-workspace")
+        repo_type: "space", "model", or "dataset"
+        private: Whether to make it private
+        token: HF API token
+        
+    Returns:
+        JSON dict with creation result.
+    """
+    from huggingface_hub import HfApi
+    
+    try:
+        api = HfApi(token=token) if token else HfApi()
+        
+        url = api.create_repo(
+            repo_id=repo_id,
+            repo_type=repo_type,
+            private=private,
+            exist_ok=True,
+        )
+        
+        yield json.dumps({
+            "success": True,
+            "repo_id": repo_id,
+            "repo_type": repo_type,
+            "private": private,
+            "url": url,
+            "message": f"Repository {repo_id} created/verified",
+        }, default_str=str)
+        
+    except Exception as e:
+        yield json.dumps({
+            "success": False,
+            "error": str(e),
+        }, default_str=str)
+
+
+@app.api(name="hub_stats")
+def handle_hub_stats() -> str:
+    """Get detailed statistics about current storage usage.
+    
+    Returns:
+        JSON dict with storage statistics for both local and Hub.
+    """
+    from code.hub_storage import get_hub_storage, is_hub_available
+    from code.tools.fs import snapshot_workspace
+    
+    stats = {
+        "timestamp": __import__("time").time(),
+        "local": {},
+        "hub": None,
+    }
+    
+    # Local stats
+    try:
+        workspace_snapshot = snapshot_workspace()
+        total_size = sum(len(c.encode('utf-8')) for c in workspace_snapshot.values())
+        stats["local"] = {
+            "file_count": len(workspace_snapshot),
+            "total_size_bytes": total_size,
+            "total_size_mb": round(total_size / (1024 * 1024), 2),
+        }
+    except Exception as e:
+        stats["local"]["error"] = str(e)
+    
+    # Hub stats
+    if is_hub_available():
+        try:
+            hub = get_hub_storage()
+            stats["hub"] = hub.get_storage_stats()
+            
+            # Get remote file count
+            hub_files = hub.list_files(use_cache=False)
+            stats["hub"]["remote_file_count"] = len(hub_files)
+            
+        except Exception as e:
+            stats["hub"] = {"error": str(e)}
+    
+    yield json.dumps(stats, default_str=str)

code/tools/fs.py

@@ -1,5 +1,10 @@
 """File system tools: read, write, edit, list, glob, grep.
 
+Enhanced with Hugging Face Hub cloud storage integration:
+- All file operations work seamlessly with both local and cloud storage
+- Automatic sync to HF Hub when configured
+- Transparent fallback between local and remote
+
 All tools are sandboxed to a configurable workspace root (default: ./workspace).
 They return JSON-serializable dicts so they can be exposed via the API.
 """
@@ -7,11 +12,14 @@ They return JSON-serializable dicts so they can be exposed via the API.
 from __future__ import annotations
 
 import fnmatch
+import logging
 import os
 import re
 import shutil
 from pathlib import Path
-from typing import Any
+from typing import Any, Optional
+
+logger = logging.getLogger(__name__)
 
 # ─── Workspace sandbox ──────────────────────────────────────────────────
 
@@ -21,6 +29,9 @@ _DEFAULT_WORKSPACE = os.environ.get(
     os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "workspace")),
 )
 
+# Hub storage instance (lazy initialized)
+_hub_storage = None
+
 
 def get_workspace_root() -> str:
     """Return the absolute path of the agent's workspace root."""
@@ -49,11 +60,30 @@ def _resolve_safe(path: str) -> str:
     return full
 
 
+def _get_hub_storage():
+    """Get the HubStorage instance if available."""
+    global _hub_storage
+    if _hub_storage is None:
+        try:
+            from code.hub_storage import get_hub_storage, is_hub_available
+            if is_hub_available():
+                _hub_storage = get_hub_storage()
+                logger.debug("Hub storage integration active")
+            else:
+                logger.debug("Hub storage not configured")
+        except ImportError:
+            logger.debug("huggingface_hub not installed")
+            _hub_storage = None
+    return _hub_storage
+
+
 # ─── read_file ──────────────────────────────────────────────────────────
 
 def read_file(path: str, offset: int = 0, limit: int | None = None) -> dict[str, Any]:
     """Read a text file from the workspace.
 
+    Enhanced: Falls back to Hub storage if file not found locally.
+
     Args:
         path: Relative path inside the workspace, or absolute within it.
         offset: 1-indexed line to start reading from.
@@ -64,32 +94,62 @@ def read_file(path: str, offset: int = 0, limit: int | None = None) -> dict[str,
     """
     try:
         full = _resolve_safe(path)
-        if not os.path.exists(full):
-            return {"success": False, "error": f"File not found: {path}"}
-        if os.path.isdir(full):
-            return {"success": False, "error": f"Path is a directory: {path}"}
-
-        with open(full, "r", encoding="utf-8", errors="replace") as f:
-            lines = f.readlines()
-
-        total = len(lines)
-        start = max(0, (offset - 1) if offset > 0 else 0)
-        end = (start + limit) if limit else total
-        selected = lines[start:end]
-
-        # Re-number for display
-        numbered = "".join(
-            f"{start + i + 1:6}\t{line}" for i, line in enumerate(selected)
-        )
+        
+        # Try local first
+        if os.path.exists(full) and os.path.isfile(full):
+            with open(full, "r", encoding="utf-8", errors="replace") as f:
+                lines = f.readlines()
+            
+            total = len(lines)
+            start = max(0, (offset - 1) if offset > 0 else 0)
+            end = (start + limit) if limit else total
+            selected = lines[start:end]
+
+            # Re-number for display
+            numbered = "".join(
+                f"{start + i + 1:6}\t{line}" for i, line in enumerate(selected)
+            )
 
-        return {
-            "success": True,
-            "path": path,
-            "content": numbered,
-            "line_count": total,
-            "returned_lines": len(selected),
-            "truncated": end < total,
-        }
+            return {
+                "success": True,
+                "path": path,
+                "content": numbered,
+                "line_count": total,
+                "returned_lines": len(selected),
+                "truncated": end < total,
+                "source": "local",
+            }
+        
+        # Try Hub storage
+        hub = _get_hub_storage()
+        if hub and hub.file_exists(path):
+            try:
+                content = hub.read_file(path)
+                lines = content.split('\n')
+                
+                total = len(lines)
+                start = max(0, (offset - 1) if offset > 0 else 0)
+                end = (start + limit) if limit else total
+                selected = lines[start:end]
+                
+                numbered = "".join(
+                    f"{start + i + 1:6}\t{line}" for i, line in enumerate(selected)
+                )
+                
+                return {
+                    "success": True,
+                    "path": path,
+                    "content": numbered,
+                    "line_count": total,
+                    "returned_lines": len(selected),
+                    "truncated": end < total,
+                    "source": "hub",
+                }
+            except Exception as e:
+                logger.warning("Failed to read %s from Hub: %s", path, e)
+        
+        return {"success": False, "error": f"File not found: {path}"}
+    
     except Exception as exc:
         return {"success": False, "error": str(exc)}
 
@@ -97,17 +157,52 @@ def read_file(path: str, offset: int = 0, limit: int | None = None) -> dict[str,
 # ─── write_file ─────────────────────────────────────────────────────────
 
 def write_file(path: str, content: str) -> dict[str, Any]:
-    """Write content to a file, creating parent directories as needed."""
+    """Write content to a file, creating parent directories as needed.
+
+    Enhanced: Automatically syncs to HF Hub when configured.
+    
+    Args:
+        path: Relative file path in workspace
+        content: File content string
+        
+    Returns:
+        dict with success status and metadata
+    """
     try:
         full = _resolve_safe(path)
         os.makedirs(os.path.dirname(full), exist_ok=True)
+        
+        # Write locally first
         with open(full, "w", encoding="utf-8") as f:
             f.write(content)
-        return {
+        
+        result = {
             "success": True,
             "path": path,
             "bytes_written": len(content.encode("utf-8")),
+            "source": "local",
         }
+        
+        # Sync to Hub if configured
+        hub = _get_hub_storage()
+        if hub:
+            try:
+                metadata = hub.write_file(
+                    path=path,
+                    content=content,
+                    commit_message=f"AI generated/updated: {path}"
+                )
+                result["hub_sync"] = True
+                result["hub_remote"] = metadata.is_remote
+                result["source"] = "synced"
+                logger.info("Synced %s to Hub", path)
+            except Exception as e:
+                logger.warning("Failed to sync %s to Hub: %s", path, e)
+                result["hub_sync"] = False
+                result["hub_error"] = str(e)
+        
+        return result
+    
     except Exception as exc:
         return {"success": False, "error": str(exc)}
 
@@ -120,15 +215,23 @@ def edit_file(
     new_str: str,
     replace_all: bool = False,
 ) -> dict[str, Any]:
-    """Replace occurrences of old_str with new_str in a file."""
+    """Replace occurrences of old_str with new_str in a file.
+
+    Enhanced: Syncs changes back to Hub after edit.
+    """
     try:
         full = _resolve_safe(path)
-        if not os.path.exists(full):
+        
+        # Read file (local or Hub)
+        hub = _get_hub_storage()
+        if os.path.exists(full):
+            with open(full, "r", encoding="utf-8") as f:
+                content = f.read()
+        elif hub and hub.file_exists(path):
+            content = hub.read_file(path)
+        else:
             return {"success": False, "error": f"File not found: {path}"}
 
-        with open(full, "r", encoding="utf-8") as f:
-            content = f.read()
-
         if old_str not in content:
             return {
                 "success": False,
@@ -152,28 +255,49 @@ def edit_file(
             old_str, new_str, 1
         )
 
+        # Write locally
+        os.makedirs(os.path.dirname(full), exist_ok=True)
         with open(full, "w", encoding="utf-8") as f:
             f.write(new_content)
 
-        return {
+        result = {
             "success": True,
             "path": path,
             "replacements": count,
         }
+        
+        # Sync to Hub
+        if hub:
+            try:
+                hub.write_file(path, new_content, commit_message=f"Edit: {path}")
+                result["hub_sync"] = True
+            except Exception as e:
+                logger.warning("Failed to sync edit to Hub: %s", e)
+
+        return result
+    
     except Exception as exc:
         return {"success": False, "error": str(exc)}
 
 
 def multi_edit(path: str, edits: list[dict[str, Any]]) -> dict[str, Any]:
-    """Apply multiple edits to a file atomically (all-or-nothing)."""
+    """Apply multiple edits to a file atomically (all-or-nothing).
+
+    Enhanced: Uses batch operation for Hub sync.
+    """
     try:
         full = _resolve_safe(path)
-        if not os.path.exists(full):
+        
+        # Read file
+        hub = _get_hub_storage()
+        if os.path.exists(full):
+            with open(full, "r", encoding="utf-8") as f:
+                content = f.read()
+        elif hub and hub.file_exists(path):
+            content = hub.read_file(path)
+        else:
             return {"success": False, "error": f"File not found: {path}"}
 
-        with open(full, "r", encoding="utf-8") as f:
-            content = f.read()
-
         applied = 0
         for edit in edits:
             old_str = edit.get("old_str", "")
@@ -196,10 +320,23 @@ def multi_edit(path: str, edits: list[dict[str, Any]]) -> dict[str, Any]:
             )
             applied += 1
 
+        # Write locally
+        os.makedirs(os.path.dirname(full), exist_ok=True)
         with open(full, "w", encoding="utf-8") as f:
             f.write(content)
 
-        return {"success": True, "path": path, "applied": applied}
+        result = {"success": True, "path": path, "applied": applied}
+        
+        # Sync to Hub
+        if hub:
+            try:
+                hub.write_file(path, content, commit_message=f"Multi-edit: {path}")
+                result["hub_sync"] = True
+            except Exception as e:
+                logger.warning("Failed to sync multi-edit to Hub: %s", e)
+
+        return result
+    
     except Exception as exc:
         return {"success": False, "error": str(exc)}
 
@@ -207,30 +344,56 @@ def multi_edit(path: str, edits: list[dict[str, Any]]) -> dict[str, Any]:
 # ─── list_dir ───────────────────────────────────────────────────────────
 
 def list_dir(path: str = ".") -> dict[str, Any]:
-    """List directory contents."""
+    """List directory contents.
+
+    Enhanced: Merges local and Hub files.
+    """
     try:
         full = _resolve_safe(path)
-        if not os.path.exists(full):
-            return {"success": False, "error": f"Path not found: {path}"}
-        if not os.path.isdir(full):
-            return {"success": False, "error": f"Not a directory: {path}"}
-
+        
         entries = []
-        for name in sorted(os.listdir(full)):
-            entry_path = os.path.join(full, name)
-            stat = os.stat(entry_path)
-            entries.append({
-                "name": name,
-                "type": "dir" if os.path.isdir(entry_path) else "file",
-                "size": stat.st_size,
-                "path": os.path.relpath(entry_path, get_workspace_root()),
-            })
+        seen_names = set()
+        
+        # List local files
+        if os.path.exists(full) and os.path.isdir(full):
+            for name in sorted(os.listdir(full)):
+                entry_path = os.path.join(full, name)
+                stat = os.stat(entry_path)
+                rel_path = os.path.relpath(entry_path, get_workspace_root())
+                entries.append({
+                    "name": name,
+                    "type": "dir" if os.path.isdir(entry_path) else "file",
+                    "size": stat.st_size,
+                    "path": rel_path.replace(os.sep, "/"),
+                    "source": "local",
+                })
+                seen_names.add(name)
+        
+        # Merge with Hub files
+        hub = _get_hub_storage()
+        if hub:
+            try:
+                hub_files = hub.list_files(path=path, recursive=False)
+                for hf in hub_files:
+                    hf_name = Path(hf["path"]).name
+                    if hf_name not in seen_names:
+                        entries.append({
+                            "name": hf_name,
+                            "type": "file",
+                            "size": hf.get("size", 0),
+                            "path": hf["path"],
+                            "source": "hub",
+                        })
+                        seen_names.add(hf_name)
+            except Exception as e:
+                logger.warning("Failed to list Hub files: %s", e)
 
         return {
             "success": True,
             "path": path,
-            "entries": entries,
+            "entries": sorted(entries, key=lambda x: x["name"]),
         }
+    
     except Exception as exc:
         return {"success": False, "error": str(exc)}
 
@@ -238,18 +401,43 @@ def list_dir(path: str = ".") -> dict[str, Any]:
 # ─── glob ───────────────────────────────────────────────────────────────
 
 def glob_paths(pattern: str, path: str = ".") -> dict[str, Any]:
-    """Glob file paths matching a pattern, recursively."""
+    """Glob file paths matching a pattern, recursively.
+
+    Enhanced: Searches both local and Hub storage.
+    """
     try:
         full = _resolve_safe(path)
-        matches: list[str] = []
-        for root_dir, _dirs, files in os.walk(full):
-            for fname in files:
-                if fnmatch.fnmatch(fname, pattern) or fnmatch.fnmatch(
-                    os.path.relpath(os.path.join(root_dir, fname), full), pattern
-                ):
-                    matches.append(os.path.relpath(os.path.join(root_dir, fname), get_workspace_root()))
-        matches.sort()
+        matches: list[dict[str, Any]] = []
+        seen_paths = set()
+        
+        # Local glob
+        if os.path.exists(full):
+            for root_dir, _dirs, files in os.walk(full):
+                for fname in files:
+                    if fnmatch.fnmatch(fname, pattern) or fnmatch.fnmatch(
+                        os.path.relpath(os.path.join(root_dir, fname), full), pattern
+                    ):
+                        rel_path = os.path.relpath(os.path.join(root_dir, fname), get_workspace_root()).replace(os.sep, "/")
+                        if rel_path not in seen_paths:
+                            matches.append({"path": rel_path, "source": "local"})
+                            seen_paths.add(rel_path)
+        
+        # Hub glob
+        hub = _get_hub_storage()
+        if hub:
+            try:
+                hub_files = hub.list_files(path=path, recursive=True)
+                for hf in hub_files:
+                    if fnmatch.fnmatch(Path(hf["path"]).name, pattern) or fnmatch.fnmatch(hf["path"], pattern):
+                        if hf["path"] not in seen_paths:
+                            matches.append({"path": hf["path"], "source": "hub"})
+                            seen_paths.add(hf["path"])
+            except Exception as e:
+                logger.warning("Failed to glob Hub files: %s", e)
+        
+        matches.sort(key=lambda x: x["path"])
         return {"success": True, "pattern": pattern, "matches": matches}
+    
     except Exception as exc:
         return {"success": False, "error": str(exc)}
 
@@ -263,26 +451,61 @@ def grep_search(
     ignore_case: bool = False,
     max_results: int = 100,
 ) -> dict[str, Any]:
-    """Search file contents with a regex pattern."""
+    """Search file contents with a regex pattern.
+
+    Enhanced: Searches both local and Hub files.
+    """
     try:
-        full = _resolve_safe(path)
         flags = re.IGNORECASE if ignore_case else 0
         regex = re.compile(pattern, flags)
 
         matches: list[dict[str, Any]] = []
-        for root_dir, _dirs, files in os.walk(full):
-            for fname in files:
-                if include and not fnmatch.fnmatch(fname, include):
-                    continue
-                fpath = os.path.join(root_dir, fname)
-                try:
-                    with open(fpath, "r", encoding="utf-8", errors="replace") as f:
-                        for lineno, line in enumerate(f, 1):
+        
+        # Search local files
+        full = _resolve_safe(path)
+        if os.path.exists(full):
+            for root_dir, _dirs, files in os.walk(full):
+                for fname in files:
+                    if include and not fnmatch.fnmatch(fname, include):
+                        continue
+                    fpath = os.path.join(root_dir, fname)
+                    try:
+                        with open(fpath, "r", encoding="utf-8", errors="replace") as f:
+                            for lineno, line in enumerate(f, 1):
+                                if regex.search(line):
+                                    matches.append({
+                                        "file": os.path.relpath(fpath, get_workspace_root()).replace(os.sep, "/"),
+                                        "line": lineno,
+                                        "text": line.rstrip()[:500],
+                                        "source": "local",
+                                    })
+                                    if len(matches) >= max_results:
+                                        return {
+                                            "success": True,
+                                            "pattern": pattern,
+                                            "matches": matches,
+                                            "truncated": True,
+                                        }
+                    except (UnicodeDecodeError, PermissionError):
+                        continue
+        
+        # Search Hub files
+        hub = _get_hub_storage()
+        if hub and len(matches) < max_results:
+            try:
+                hub_files = hub.list_files(path=path, recursive=True)
+                for hf in hub_files:
+                    if include and not fnmatch.fnmatch(Path(hf["path"]).name, include):
+                        continue
+                    try:
+                        content = hub.read_file(hf["path"])
+                        for lineno, line in enumerate(content.split('\n'), 1):
                             if regex.search(line):
                                 matches.append({
-                                    "file": os.path.relpath(fpath, get_workspace_root()),
+                                    "file": hf["path"],
                                     "line": lineno,
                                     "text": line.rstrip()[:500],
+                                    "source": "hub",
                                 })
                                 if len(matches) >= max_results:
                                     return {
@@ -291,8 +514,10 @@ def grep_search(
                                         "matches": matches,
                                         "truncated": True,
                                     }
-                except (UnicodeDecodeError, PermissionError):
-                    continue
+                    except Exception:
+                        continue
+            except Exception as e:
+                logger.warning("Failed to search Hub files: %s", e)
 
         return {
             "success": True,
@@ -300,6 +525,7 @@ def grep_search(
             "matches": matches,
             "truncated": False,
         }
+    
     except re.error as exc:
         return {"success": False, "error": f"Invalid regex: {exc}"}
     except Exception as exc:
@@ -309,7 +535,10 @@ def grep_search(
 # ─── Workspace management ───────────────────────────────────────────────
 
 def list_workspace_tree(max_depth: int = 3) -> dict[str, Any]:
-    """Return a tree view of the workspace."""
+    """Return a tree view of the workspace.
+
+    Enhanced: Includes Hub files.
+    """
     try:
         root = get_workspace_root()
 
@@ -333,13 +562,25 @@ def list_workspace_tree(max_depth: int = 3) -> dict[str, Any]:
             return {"name": os.path.basename(path), "type": "dir", "children": entries}
 
         tree = _walk(root, 0)
+        
+        # Add Hub info
+        hub = _get_hub_storage()
+        if hub:
+            stats = hub.get_storage_stats()
+            tree["hub_enabled"] = stats["hub_enabled"]
+            tree["storage_mode"] = stats["mode"]
+        
         return {"success": True, "tree": tree}
+    
     except Exception as exc:
         return {"success": False, "error": str(exc)}
 
 
 def reset_workspace() -> dict[str, Any]:
-    """Clear all files in the workspace (used by /new command)."""
+    """Clear all files in the workspace (used by /new command).
+
+    Note: Does NOT clear Hub storage - use reset_hub_workspace for that.
+    """
     try:
         root = get_workspace_root()
         if os.path.exists(root):
@@ -357,12 +598,14 @@ def reset_workspace() -> dict[str, Any]:
 def snapshot_workspace() -> dict[str, str]:
     """Return a dict of {relative_path: content} for all text files in the workspace.
 
+    Enhanced: Includes Hub files in snapshot.
     Used to package workspace files for ZIP/HF deploy.
     """
     root = get_workspace_root()
     files: dict[str, str] = {}
+    
+    # Local files
     for dirpath, _dirs, fnames in os.walk(root):
-        # Skip hidden dirs and node_modules / __pycache__
         parts = os.path.relpath(dirpath, root).split(os.sep)
         if any(p.startswith(".") or p in {"node_modules", "__pycache__", ".venv", "venv"} for p in parts):
             continue
@@ -375,4 +618,105 @@ def snapshot_workspace() -> dict[str, str]:
                     files[os.path.relpath(full, root)] = f.read()
             except (UnicodeDecodeError, PermissionError):
                 continue
+    
+    # Hub files (merge, don't overwrite local)
+    hub = _get_hub_storage()
+    if hub:
+        try:
+            hub_files = hub.list_files()
+            for hf in hub_files:
+                if hf["path"] not in files:
+                    try:
+                        content = hub.read_file(hf["path"])
+                        files[hf["path"]] = content
+                    except Exception:
+                        continue
+        except Exception as e:
+            logger.warning("Failed to snapshot Hub files: %s", e)
+    
     return files
+
+
+# ─── Hub-specific operations ────────────────────────────────────────────
+
+def sync_to_hub(path: str | None = None) -> dict[str, Any]:
+    """Manually trigger sync to HF Hub.
+    
+    Args:
+        path: Specific file/directory to sync (None = all)
+        
+    Returns:
+        Sync result dict
+    """
+    hub = _get_hub_storage()
+    if not hub:
+        return {"success": False, "error": "Hub storage not configured"}
+    
+    try:
+        return hub.sync_to_hub(path=path)
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+def sync_from_hub(path: str | None = None) -> dict[str, Any]:
+    """Manually trigger sync from HF Hub.
+    
+    Args:
+        path: Specific file to download (None = all)
+        
+    Returns:
+        Sync result dict
+    """
+    hub = _get_hub_storage()
+    if not hub:
+        return {"success": False, "error": "Hub storage not configured"}
+    
+    try:
+        return hub.sync_from_hub(path=path)
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+def get_hub_status() -> dict[str, Any]:
+    """Get current Hub storage status and configuration."""
+    hub = _get_hub_storage()
+    if not hub:
+        return {
+            "available": False,
+            "configured": False,
+            "mode": "local_only",
+        }
+    
+    try:
+        stats = hub.get_storage_stats()
+        return {
+            "available": True,
+            "configured": True,
+            **stats,
+        }
+    except Exception as e:
+        return {
+            "available": False,
+            "configured": True,
+            "error": str(e),
+        }
+
+
+def delete_from_hub(path: str) -> dict[str, Any]:
+    """Delete a file from Hub storage only (keeps local).
+    
+    Args:
+        path: File path to delete from Hub
+        
+    Returns:
+        Operation result
+    """
+    hub = _get_hub_storage()
+    if not hub:
+        return {"success": False, "error": "Hub storage not configured"}
+    
+    try:
+        success = hub.delete_file(path, commit_message=f"Delete: {path}")
+        return {"success": success, "path": path}
+    except Exception as e:
+        return {"success": False, "error": str(e)}