Repository: Significant-Gravitas/Auto-GPT
File: autogpt_platform/backend/backend/util/workspace_storage.py (Python)

Code before:
        pass

    @abstractmethod
    async def get_download_url(self, storage_path: str, expires_in: int = 3600) -> str:
        """
        Get URL for downloading the file.

        Args:
            storage_path: The storage path
            expires_in: URL expiration time in seconds (default 1 hour)

        Returns:
            Download URL (signed URL for GCS, direct API path for local)
        """
        pass


class GCSWorkspaceStorage(WorkspaceStorageBackend):
    """Google Cloud Storage implementation for workspace storage.

    Event-loop-aware: each event loop gets its own aiohttp session and GCS
    client.  This is required because aiohttp.ClientSession (and the
    asyncio.timeout() calls it makes internally) are bound to the loop they
    were created on.  In the copilot executor every worker thread runs its
    own event loop, so a single shared session would fail with
    "Timeout context manager should be used inside a task".
    """

    def __init__(self, bucket_name: str):
        self.bucket_name = bucket_name
        # Keyed by id(event_loop) → per-loop client/session
        self._async_clients: dict[int, async_gcs_storage.Storage] = {}
        self._sessions: dict[int, aiohttp.ClientSession] = {}
        self._sync_client: Optional[gcs_storage.Client] = None

    async def _get_async_client(self) -> async_gcs_storage.Storage:
        """Get or create an async GCS client for the current event loop."""
        import asyncio

        loop_id = id(asyncio.get_running_loop())
        if loop_id not in self._async_clients:
            session = aiohttp.ClientSession(
                connector=aiohttp.TCPConnector(limit=100, force_close=False)
            )
            self._sessions[loop_id] = session
            self._async_clients[loop_id] = async_gcs_storage.Storage(session=session)
        return self._async_clients[loop_id]

    def _get_sync_client(self) -> gcs_storage.Client:
        """Get or create sync GCS client (for signed URLs)."""
        if self._sync_client is None:

Review comment (bug):
```suggestion
    ## Event-loop-aware
    Each event loop gets its own `aiohttp` session and GCS
    client. This is required because `aiohttp.ClientSession` (and the
    `asyncio.timeout()` calls it makes internally) are bound to the loop they
    were created on. In the copilot executor every worker thread runs its
    own event loop, so a single shared session would fail with
    "Timeout context manager should be used inside a task".
```

Code after:
        pass

    @abstractmethod
    async def get_download_url(self, storage_path: str, expires_in: int = 3600) -> str:
        """
        Get URL for downloading the file.

        Args:
            storage_path: The storage path
            expires_in: URL expiration time in seconds (default 1 hour)

        Returns:
            Download URL (signed URL for GCS, direct API path for local)
        """
        pass


class GCSWorkspaceStorage(WorkspaceStorageBackend):
    """Google Cloud Storage implementation for workspace storage.

    Each instance owns a single ``aiohttp.ClientSession`` and GCS async
    client.  Because ``ClientSession`` is bound to the event loop on which it
    was created, callers that run on separate loops (e.g. copilot executor
    worker threads) **must** obtain their own ``GCSWorkspaceStorage`` instance
    via :func:`get_workspace_storage` which is event-loop-aware.
    """

    def __init__(self, bucket_name: str):
        self.bucket_name = bucket_name
        self._async_client: Optional[async_gcs_storage.Storage] = None
        self._sync_client: Optional[gcs_storage.Client] = None
        self._session: Optional[aiohttp.ClientSession] = None

    async def _get_async_client(self) -> async_gcs_storage.Storage:
        """Get or create async GCS client."""
        if self._async_client is None:
            self._session = aiohttp.ClientSession(
                connector=aiohttp.TCPConnector(limit=100, force_close=False)
            )
            self._async_client = async_gcs_storage.Storage(session=self._session)
        return self._async_client

    def _get_sync_client(self) -> gcs_storage.Client:
        """Get or create sync GCS client (for signed URLs)."""
        if self._sync_client is None:
            self._sync_client = gcs_storage.Client()
        return self._sync_client

    async def close(self) -> None:
        """Close all client connections."""
        if self._async_client is not None: