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

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

@@ -1,4 +1,5 @@
 from typing import List, Optional
+from uuid import UUID
 
 from pydantic import BaseModel, Field
 
@@ -36,6 +37,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: UUID
+    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
 # ---------------------------------------------------------------------------
@@ -58,6 +75,12 @@ class MountsResponse(BaseModel):
 
 class MountFileListResponse(BaseModel):
     count: int = 0
+    # Entries this view would return BEFORE any limit — so a limited "latest N" listing still reports
+    # the true total (the UI badge). Its unit follows the view: leaf files only in the recency listing
+    # (order/limit set), files-plus-folders in the shallow (depth=1) and browse modes.
+    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)
 
 

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

@@ -1,5 +1,5 @@
 from functools import wraps
-from typing import Optional
+from typing import Literal, Optional
 from uuid import UUID
 
 from fastapi import APIRouter, HTTPException, Query, Request, UploadFile, status
@@ -10,6 +10,7 @@
 from oss.src.core.access.permissions.service import check_action_access
 from oss.src.apis.fastapi.shared.exceptions import FORBIDDEN_EXCEPTION
 
+from oss.src.core.mounts.dtos import MountArchiveSource
 from oss.src.core.mounts.service import MountsService
 from oss.src.core.mounts.types import (
     MountArtifactIdInvalid,
@@ -27,6 +28,7 @@
 
 from oss.src.apis.fastapi.mounts.models import (
     AgentMountQueryRequest,
+    MountArchiveRequest,
     MountCreateRequest,
     MountCredentialsResponse,
     MountEditRequest,
@@ -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 the "/{mount_id}/..." routes so this literal path isn't captured as an
+        # operation on a mount named "files".
+        self.router.add_api_route(
+            "/files/export",
+            self.export_mount_files,
+            methods=["POST"],
+            operation_id="export_mount_files",
+            response_model=None,
+            status_code=status.HTTP_200_OK,
+        )
         self.router.add_api_route(
             "/{mount_id}/archive",
             self.archive_mount,
@@ -487,6 +500,14 @@ async def get_mount_files(
         *,
         path: Optional[str] = Query(default=None),
         read: Optional[str] = Query(default=None),
+        order: Optional[Literal["recent", "name", "path"]] = Query(default=None),
+        limit: Optional[int] = Query(default=None, ge=0),
+        # Only depth==1 is implemented (the shallow one-level summary); reject other values loudly
+        # instead of silently falling through to the most expensive full-tree branch.
+        depth: Optional[int] = Query(default=None, ge=1, le=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,9 +526,17 @@ 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,
         )
 
@@ -589,6 +618,30 @@ async def download_mount_file(
             path=path,
         )
 
+    @intercept_exceptions()
+    @handle_mount_exceptions()
+    async def export_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=[
+                MountArchiveSource(
+                    mount_id=m.mount_id,
+                    archive_prefix=m.prefix,
+                    source_path=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,46 @@
+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
+from urllib.parse import quote
 from uuid import UUID
 
 from fastapi import Response, UploadFile
-
-from oss.src.core.mounts.dtos import MountCredentials, MountFileWritten, MountQuery
+from fastapi.responses import StreamingResponse
+from stream_zip import ZIP_AUTO, async_stream_zip
+
+from oss.src.core.mounts.dtos import (
+    MountArchiveSource,
+    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
+
+
+def _content_disposition_attachment(filename: str) -> str:
+    """Build a safe `Content-Disposition: attachment` header value (RFC 6266).
+
+    `filename` is client-supplied (a download path's basename or the archive name), so it must never
+    be interpolated raw: a `"` or control char would break out of the quoted parameter and inject
+    further header directives. The quoted `filename` is stripped to an ASCII, printable, quote-free
+    fallback (the header is latin-1 encoded); `filename*` carries the exact (percent-encoded) value for
+    clients that honour it.
+    """
+    safe = (
+        "".join(
+            c for c in filename if c.isascii() and c.isprintable() and c not in '"\\'
+        )
+        or "download"
+    )
+    return (
+        f"attachment; filename=\"{safe}\"; filename*=UTF-8''{quote(filename, safe='')}"
+    )
+
 
 async def upload_mount_file(
     *,
@@ -58,7 +91,64 @@ async def download_mount_file(
     return Response(
         content=body,
         media_type=media_type,
-        headers={"Content-Disposition": f'attachment; filename="{name}"'},
+        headers={"Content-Disposition": _content_disposition_attachment(name)},
+    )
+
+
+async def stream_mounts_archive(
+    *,
+    mounts_service: MountsService,
+    project_id: UUID,
+    mounts: List[MountArchiveSource],
+    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 entry
+    by size; very large individual files are out of product scope for "download all".
+    """
+    work = await mounts_service.build_archive_work_list(
+        project_id=project_id,
+        mounts=mounts,
+    )
+
+    async def members():
+        async for (
+            zip_path,
+            _size,
+            mtime,
+            body,
+        ) in mounts_service.iter_archive_members(
+            work=work,
+        ):
+            # `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": _content_disposition_attachment(filename)},
     )
 
 

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

@@ -63,10 +63,32 @@ 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)
+    # Entries this view would return BEFORE any limit — so a limited "latest N" listing still reports
+    # the true total (the UI badge) without shipping the whole tree. Its unit follows the view: leaf
+    # files only in the recency listing (order/limit set), files-plus-folders in the shallow (depth=1)
+    # and browse modes.
+    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 MountArchiveSource(BaseModel):
+    """One mount to include in a download-all archive: which mount, which folder within it
+    (`source_path`; "" = the whole mount), and the prefix its files sit under in the zip (the folded
+    drive layout)."""
+
+    mount_id: UUID
+    source_path: str = ""
+    archive_prefix: str = ""
 
 
 class MountFileContent(BaseModel):