[AGE-3965] refactor(drive): virtualize surfaces + unify Files drawer

api/oss/src/apis/fastapi/mounts/models.py

@@ -36,6 +36,22 @@ class AgentMountQueryRequest(BaseModel):
     name: str = "default"
 
 
+class ArchiveMount(BaseModel):
+    """One mount to include in an archive. `path` scopes it to a folder within the mount ("" = the
+    whole mount); `prefix` places its files under `prefix/` in the zip (the folded drive layout)."""
+
+    mount_id: str
+    prefix: str = ""
+    path: str = ""
+
+
+class MountArchiveRequest(BaseModel):
+    """Zip several mounts into ONE archive (the drive folds cwd + agent-files into one tree)."""
+
+    mounts: List[ArchiveMount] = Field(default_factory=list)
+    filename: str = "files.zip"
+
+
 # ---------------------------------------------------------------------------
 # Response models
 # ---------------------------------------------------------------------------
@@ -57,8 +73,23 @@ class MountsResponse(BaseModel):
 
 
 class MountFileListResponse(BaseModel):
+    count: int = 0
+    # Full file count matching the request before any limit — lets a limited "latest N" listing
+    # still report the true total (the UI badge). Equals `count` for an unlimited listing.
+    total: int = 0
+    # `total` is a FLOOR (the count-only scan hit its cap) — the UI shows "N+". False when exact.
+    total_capped: bool = False
+    files: List[MountFile] = Field(default_factory=list)
+
+
+class MountFilePageResponse(BaseModel):
+    """One cursor PAGE of a mount's flat (recursive, path-sorted) file listing — the basis for the
+    Files drawer's infinite-scroll flat view. `next_cursor` is an opaque token for the following page;
+    None means the listing is exhausted."""
+
     count: int = 0
     files: List[MountFile] = Field(default_factory=list)
+    next_cursor: Optional[str] = None
 
 
 class MountFileContentResponse(BaseModel):

api/oss/src/apis/fastapi/mounts/router.py

@@ -27,12 +27,14 @@
 
 from oss.src.apis.fastapi.mounts.models import (
     AgentMountQueryRequest,
+    MountArchiveRequest,
     MountCreateRequest,
     MountCredentialsResponse,
     MountEditRequest,
     MountFileContentResponse,
     MountFileDeletedResponse,
     MountFileListResponse,
+    MountFilePageResponse,
     MountFileWrittenResponse,
     MountFolderCreatedResponse,
     MountQueryRequest,
@@ -43,6 +45,7 @@
     download_mount_file,
     merge_mount_query,
     sign_mount_credentials,
+    stream_mounts_archive,
     upload_mount_file,
 )
 
@@ -189,6 +192,16 @@ def __init__(
             response_model_exclude_none=True,
             status_code=status.HTTP_200_OK,
         )
+        # Registered BEFORE "/{mount_id}/archive" so `POST /files/archive` (download-all zip) isn't
+        # captured as archiving a mount literally named "files".
+        self.router.add_api_route(
+            "/files/archive",
+            self.archive_mount_files,
+            methods=["POST"],
+            operation_id="archive_mount_files",
+            response_model=None,
+            status_code=status.HTTP_200_OK,
+        )
         self.router.add_api_route(
             "/{mount_id}/archive",
             self.archive_mount,
@@ -236,6 +249,16 @@ def __init__(
             response_model=None,
             status_code=status.HTTP_200_OK,
         )
+        # Registered before "/{mount_id}/files" so `/files/page` isn't swallowed by the browse route.
+        self.router.add_api_route(
+            "/{mount_id}/files/page",
+            self.get_mount_files_page,
+            methods=["GET"],
+            operation_id="get_mount_files_page",
+            response_model=MountFilePageResponse,
+            response_model_exclude_none=True,
+            status_code=status.HTTP_200_OK,
+        )
         self.router.add_api_route(
             "/{mount_id}/files",
             self.get_mount_files,
@@ -487,6 +510,12 @@ async def get_mount_files(
         *,
         path: Optional[str] = Query(default=None),
         read: Optional[str] = Query(default=None),
+        order: Optional[str] = Query(default=None),
+        limit: Optional[int] = Query(default=None, ge=0),
+        depth: Optional[int] = Query(default=None, ge=1),
+        with_counts: bool = Query(default=False),
+        git_aware: bool = Query(default=False),
+        include_gitignored: bool = Query(default=False),
     ):
         await self._check(request, Permission.VIEW_MOUNTS)
 
@@ -505,12 +534,53 @@ async def get_mount_files(
             project_id=UUID(request.state.project_id),
             mount_id=mount_id,
             path=path,
+            order=order,
+            limit=limit,
+            depth=depth,
+            with_counts=with_counts,
+            git_aware=git_aware,
+            include_gitignored=include_gitignored,
         )
         return MountFileListResponse(
             count=len(listing.files),
+            total=listing.total,
+            total_capped=listing.total_capped,
             files=listing.files,
         )
 
+    @intercept_exceptions()
+    @handle_mount_exceptions()
+    async def get_mount_files_page(
+        self,
+        request: Request,
+        mount_id: UUID,
+        *,
+        path: Optional[str] = Query(default=None),
+        cursor: Optional[str] = Query(default=None),
+        limit: int = Query(default=100, ge=1, le=1000),
+        git_aware: bool = Query(default=False),
+        include_gitignored: bool = Query(default=False),
+    ) -> MountFilePageResponse:
+        """One cursor page of the flat (recursive, path-sorted) file listing under `path` — the Files
+        drawer's infinite-scroll flat view. Never enumerates the whole subtree, so it's fast on any
+        mount size; carry `next_cursor` back to fetch the next page."""
+        await self._check(request, Permission.VIEW_MOUNTS)
+
+        files, next_cursor = await self.mounts_service.list_files_page(
+            project_id=UUID(request.state.project_id),
+            mount_id=mount_id,
+            path=path,
+            cursor=cursor,
+            limit=limit,
+            git_aware=git_aware,
+            include_gitignored=include_gitignored,
+        )
+        return MountFilePageResponse(
+            count=len(files),
+            files=files,
+            next_cursor=next_cursor,
+        )
+
     @intercept_exceptions()
     @handle_mount_exceptions()
     async def write_mount_file(
@@ -589,6 +659,25 @@ async def download_mount_file(
             path=path,
         )
 
+    @intercept_exceptions()
+    @handle_mount_exceptions()
+    async def archive_mount_files(
+        self,
+        request: Request,
+        *,
+        archive_request: MountArchiveRequest,
+    ):
+        await self._check(request, Permission.VIEW_MOUNTS)
+
+        return await stream_mounts_archive(
+            mounts_service=self.mounts_service,
+            project_id=UUID(request.state.project_id),
+            mounts=[
+                (UUID(m.mount_id), m.prefix, m.path) for m in archive_request.mounts
+            ],
+            filename=archive_request.filename,
+        )
+
     @intercept_exceptions()
     @handle_mount_exceptions()
     async def delete_mount_file(

api/oss/src/apis/fastapi/mounts/utils.py

@@ -1,13 +1,20 @@
+from datetime import datetime, timezone
 from mimetypes import guess_type
 from posixpath import basename
-from typing import Optional
+from stat import S_IFREG
+from typing import List, Optional, Tuple
 from uuid import UUID
 
 from fastapi import Response, UploadFile
+from fastapi.responses import StreamingResponse
+from stream_zip import ZIP_AUTO, async_stream_zip
 
 from oss.src.core.mounts.dtos import MountCredentials, MountFileWritten, MountQuery
 from oss.src.core.mounts.service import MountsService
 
+# Regular-file mode for archive members (owner rw, group/other r).
+_ARCHIVE_FILE_MODE = S_IFREG | 0o644
+
 
 async def upload_mount_file(
     *,
@@ -62,6 +69,60 @@ async def download_mount_file(
     )
 
 
+async def stream_mounts_archive(
+    *,
+    mounts_service: MountsService,
+    project_id: UUID,
+    mounts: List[Tuple[UUID, str, str]],
+    filename: str = "files.zip",
+) -> StreamingResponse:
+    """STREAM a zip of EVERY file across the given mounts as a binary download ("download all").
+
+    The drive folds cwd + agent-files into one tree, so each ``(mount_id, prefix)`` is placed under
+    ``prefix/`` in the zip. The archive is streamed member-by-member (never buffered whole), and the
+    service prefetches file bodies with bounded concurrency. ``ZIP_AUTO`` picks zip32/zip64 per file
+    by size, so large drives and >4 GB archives are handled.
+    """
+
+    async def members():
+        async for (
+            zip_path,
+            _size,
+            mtime,
+            body,
+        ) in mounts_service.iter_archive_members(
+            project_id=project_id,
+            mounts=mounts,
+        ):
+            # `mtime` is the store's LastModified as epoch MILLISECONDS (see StoreObject.mtime);
+            # `datetime.fromtimestamp` wants SECONDS — passing ms overflows to a year out of range
+            # and RAISES mid-stream (after 200 headers are sent), truncating the zip to 0 bytes.
+            modified_at = (
+                datetime.fromtimestamp(mtime / 1000, tz=timezone.utc)
+                if mtime
+                else datetime.now(tz=timezone.utc)
+            )
+
+            async def _data(_body=body):
+                yield _body
+
+            # Size from the actual bytes (not the pre-read listing) so a file changed between list
+            # and read can't desync the zip entry; ZIP_AUTO then picks zip32/zip64 accordingly.
+            yield (
+                zip_path,
+                modified_at,
+                _ARCHIVE_FILE_MODE,
+                ZIP_AUTO(len(body)),
+                _data(),
+            )
+
+    return StreamingResponse(
+        async_stream_zip(members()),
+        media_type="application/zip",
+        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
+    )
+
+
 async def sign_mount_credentials(
     *,
     mounts_service: MountsService,

api/oss/src/core/mounts/dtos.py

@@ -63,10 +63,20 @@ class MountFile(BaseModel):
     # Object-store LastModified as epoch milliseconds; None when the store omits it. Lets the UI
     # order files by recency regardless of how they were created (bash, Write tool, upload).
     mtime: Optional[int] = None
+    # Direct-child count for a folder entry — only set when the recency view rolls a whole
+    # freshly-written directory (e.g. a `git clone`) up into ONE folder row instead of flooding the
+    # "recent files" list with its leaves. None for real files.
+    item_count: Optional[int] = None
 
 
 class MountFileList(BaseModel):
     files: List[MountFile] = Field(default_factory=list)
+    # Total real files matching the request BEFORE any limit — so a limited "latest N" listing can
+    # still report the true file count (the UI badge) without shipping the whole tree.
+    total: int = 0
+    # `total` is a FLOOR, not exact — the count-only scan hit its cap on a very large tree, so the UI
+    # shows "N+". False for an exhaustive count.
+    total_capped: bool = False
 
 
 class MountFileContent(BaseModel):