fix(api): eager archive errors, capped count-only listing, rollup tests

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

@@ -98,6 +98,10 @@ async def stream_mounts_archive(
     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.
     """
+    work = await mounts_service.build_archive_work_list(
+        project_id=project_id,
+        mounts=mounts,
+    )
 
     async def members():
         async for (
@@ -106,8 +110,7 @@ async def members():
             mtime,
             body,
         ) in mounts_service.iter_archive_members(
-            project_id=project_id,
-            mounts=mounts,
+            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

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

@@ -946,6 +946,33 @@ async def _count_children(sub_rel: str) -> Tuple[str, int]:
                 store_files, specs, truncated = await self._list_pruned_files(
                     base_prefix=list_prefix, mount_base=mount_base, cap=cap
                 )
+            elif cap is not None:
+                # RAW count-only: page until MORE than `cap` real files are known to exist (the UI
+                # then shows "N+") or the tree is exhausted, so a huge tree can't run away (matches
+                # the git-aware branch's bounded-count contract). `has_more` counts OBJECTS, not
+                # files, so truncation is decided on the file count alone — folder markers never
+                # inflate `total` into a false "N+" (they are assumed sparse; a marker-only tree is
+                # the one case still paged to exhaustion).
+                store_files = []
+                specs: List[Tuple[str, "pathspec.PathSpec"]] = []
+                truncated = False
+                start_after: Optional[str] = None
+                while len(store_files) <= cap:
+                    objs, has_more = await self.mounts_store.list_objects_page(
+                        bucket=self._bucket(),
+                        prefix=list_prefix,
+                        start_after=start_after,
+                        max_keys=max(cap, 200),
+                    )
+                    if not objs:
+                        break
+                    start_after = objs[-1].key
+                    store_files.extend(o for o in objs if not o.key.endswith("/"))
+                    if not has_more:
+                        break
+                if len(store_files) > cap:
+                    truncated = True
+                    store_files = store_files[:cap]
             else:
                 # RAW: every object under the prefix, no pruning (matches the plain-endpoint contract).
                 objects = await self.mounts_store.list_objects_v2(
@@ -1059,23 +1086,22 @@ async def read_file_bytes(
         key = self._storage_key(project_id=project_id, mount=mount, path=path)
         return await self.mounts_store.get_object(bucket=self._bucket(), key=key)
 
-    async def iter_archive_members(
+    async def build_archive_work_list(
         self,
         *,
         project_id: UUID,
         mounts: List[Tuple[UUID, str, str]],
-        concurrency: int = _ARCHIVE_READ_CONCURRENCY,
-    ) -> AsyncIterator[Tuple[str, int, Optional[int], bytes]]:
-        """Yield ``(zip_path, size, mtime, raw_bytes)`` for the files in the given mounts, in order —
-        the basis for a STREAMING archive. Each mount is a ``(mount_id, zip_prefix, source_path)``:
+    ) -> List[Tuple[str, str, int, Optional[int]]]:
+        """Build the ordered archive work list for the given mounts.
+
+        Each mount is a ``(mount_id, zip_prefix, source_path)``:
         ``source_path`` scopes it to a FOLDER within the mount ("" = the whole mount, for "download
         all"); ``zip_prefix`` places its files under ``prefix/`` in the zip (e.g. "agent-files" for
-        the folded agent mount). Folder markers are skipped. Reads up to ``concurrency`` files AHEAD
-        (bounded ordered prefetch) — never buffering the zip whole nor hammering the store.
+        the folded agent mount). Folder markers are skipped. Each work item is a
+        ``(zip_path, storage_key, size, mtime)`` tuple.
         """
         bucket = self._bucket()
 
-        # Full ordered work list across mounts: (zip_path, storage_key, size, mtime).
         work: List[Tuple[str, str, int, Optional[int]]] = []
         for mount_id, prefix, source_path in mounts:
             mount = await self._resolve_mount(project_id=project_id, mount_id=mount_id)
@@ -1101,6 +1127,17 @@ async def iter_archive_members(
                 zip_path = f"{pfx}/{rel}" if pfx else rel
                 work.append((zip_path, obj.key, obj.size or 0, obj.mtime))
 
+        return work
+
+    async def iter_archive_members(
+        self,
+        *,
+        work: List[Tuple[str, str, int, Optional[int]]],
+        concurrency: int = _ARCHIVE_READ_CONCURRENCY,
+    ) -> AsyncIterator[Tuple[str, int, Optional[int], bytes]]:
+        """Yield ``(zip_path, size, mtime, raw_bytes)`` with bounded ordered prefetch."""
+        bucket = self._bucket()
+
         # Ordered bounded-concurrency prefetch: keep ~`concurrency` reads in flight, yield in order.
         inflight: deque = deque()
         cursor = 0

api/oss/src/core/store/storage.py

@@ -335,6 +335,36 @@ async def list_objects_v2(
             )
         return results
 
+    async def list_objects_page(
+        self,
+        *,
+        bucket: str,
+        prefix: str,
+        start_after: Optional[str] = None,
+        max_keys: int = 500,
+    ) -> "Tuple[List[StoreObject], bool]":
+        """One PAGE of objects under `prefix`, in the store's lexicographic (= path) order, resuming
+        strictly AFTER `start_after`. Returns `(objects, has_more)`. Streams lazily and stops after
+        `max_keys` — so it NEVER enumerates a huge subtree; this is the basis for bounded counting
+        (the caller carries the last key forward as `start_after`). One extra element is pulled to
+        detect `has_more`, then dropped."""
+        client = self._client()
+        results: List[StoreObject] = []
+        async for obj in client.list_objects(
+            bucket, prefix=prefix, recursive=True, start_after=start_after or None
+        ):
+            if obj is None:
+                continue
+            # We already have a full page — the presence of one more object means more remain.
+            if len(results) >= max_keys:
+                return results, True
+            last_modified = getattr(obj, "last_modified", None)
+            mtime = int(last_modified.timestamp() * 1000) if last_modified else None
+            results.append(
+                StoreObject(key=obj.object_name, size=obj.size or 0, mtime=mtime)
+            )
+        return results, False
+
     async def list_objects_shallow(
         self,
         *,

api/oss/tests/pytest/unit/test_mounts_file_ops.py

@@ -17,11 +17,17 @@
 
 import pytest
 
-from oss.src.core.mounts.dtos import Mount
-from oss.src.core.mounts.service import MountsService, validate_file_path
+from oss.src.core.mounts import service as mounts_service_module
+from oss.src.core.mounts.dtos import Mount, MountFile
+from oss.src.core.mounts.service import (
+    MountsService,
+    _rollup_recent_entries,
+    validate_file_path,
+)
 from oss.src.core.store.dtos import StoreObject
 from oss.src.core.mounts.types import (
     MountFileNotFound,
+    MountNotFound,
     MountPathInvalid,
 )
 
@@ -62,6 +68,73 @@ def test_rejects_special_chars(self):
             validate_file_path("file;rm -rf")
 
 
+class TestRollupRecentEntries:
+    def test_clone_then_edit_collapses_to_top_directory(self):
+        files = [
+            MountFile(path="repo/a.txt", mtime=1000, size=1),
+            MountFile(path="repo/web/b.txt", mtime=1001, size=1),
+            MountFile(path="repo/web/c.txt", mtime=1002, size=1),
+            MountFile(path="note.txt", mtime=5000, size=1),
+        ]
+
+        by_path = {entry.path: entry for entry in _rollup_recent_entries(files, None)}
+
+        assert by_path["repo"].is_folder is True
+        assert "note.txt" in by_path
+        assert not by_path["note.txt"].is_folder
+        assert not any(path.startswith("repo/") for path in by_path)
+
+    def test_old_plus_fresh_directory_does_not_collapse(self):
+        files = [
+            MountFile(path="dir/old.txt", mtime=100, size=1),
+            MountFile(path="dir/new.txt", mtime=5000, size=1),
+            MountFile(path="recent.txt", mtime=5001, size=1),
+        ]
+
+        by_path = {entry.path: entry for entry in _rollup_recent_entries(files, None)}
+
+        assert "dir" not in by_path
+        assert "dir/old.txt" in by_path
+        assert "dir/new.txt" in by_path
+
+    def test_untimed_leaf_blocks_collapse(self):
+        files = [
+            MountFile(path="batch/a.txt", mtime=1000, size=1),
+            MountFile(path="batch/b.txt", mtime=None, size=1),
+            MountFile(path="later.txt", mtime=5000, size=1),
+        ]
+
+        by_path = {entry.path: entry for entry in _rollup_recent_entries(files, None)}
+
+        assert "batch" not in by_path
+        assert "batch/a.txt" in by_path
+        assert "batch/b.txt" in by_path
+
+    def test_single_batch_history_produces_no_rollup(self):
+        files = [
+            MountFile(path="repo/a.txt", mtime=1000, size=1),
+            MountFile(path="repo/b.txt", mtime=1001, size=1),
+        ]
+
+        result = _rollup_recent_entries(files, None)
+
+        assert all(not entry.is_folder for entry in result)
+        assert {entry.path for entry in result} == {"repo/a.txt", "repo/b.txt"}
+
+    def test_shallow_to_deep_resolution_picks_repo_over_repo_web(self):
+        files = [
+            MountFile(path="repo/a.txt", mtime=1000, size=1),
+            MountFile(path="repo/web/b.txt", mtime=1001, size=1),
+            MountFile(path="repo/web/c.txt", mtime=1002, size=1),
+            MountFile(path="outside.txt", mtime=9000, size=1),
+        ]
+
+        result = _rollup_recent_entries(files, None)
+        folder_paths = {entry.path for entry in result if entry.is_folder}
+
+        assert folder_paths == {"repo"}
+
+
 # ---------------------------------------------------------------------------
 # In-memory fake storage (same interface as ObjectStore)
 # ---------------------------------------------------------------------------
@@ -80,6 +153,21 @@ async def list_objects_v2(self, *, bucket: str, prefix: str) -> List[StoreObject
             if k.startswith(prefix)
         ]
 
+    async def list_objects_page(
+        self, *, bucket: str, prefix: str, start_after=None, max_keys: int = 500
+    ) -> Tuple[List[StoreObject], bool]:
+        # One bounded page in the store's UTF-8 byte order, resuming strictly after `start_after`.
+        b = self._store.get(bucket, {})
+        keys = sorted(
+            (k for k in b if k.startswith(prefix)), key=lambda k: k.encode("utf-8")
+        )
+        if start_after is not None:
+            sa = start_after.encode("utf-8")
+            keys = [k for k in keys if k.encode("utf-8") > sa]
+        page = keys[:max_keys]
+        has_more = len(keys) > max_keys
+        return [StoreObject(key=k, size=len(b[k])) for k in page], has_more
+
     async def list_objects_shallow(self, *, bucket: str, prefix: str):
         # One level under `prefix` (delimiter "/"): immediate files + immediate subdir prefixes.
         # Mirrors the real store: a trailing-slash key is a folder marker / common-prefix (a subdir),
@@ -137,6 +225,11 @@ async def fetch_mount(self, *, project_id, mount_id):
         return self._mount
 
 
+class _MissingMountDAO:
+    async def fetch_mount(self, *, project_id, mount_id):
+        return None
+
+
 def _make_mount() -> Mount:
     return Mount(
         id=uuid4(),
@@ -154,6 +247,49 @@ def _make_service(mount: Mount) -> Tuple[MountsService, UUID, UUID]:
     return service, mount.project_id, mount.id
 
 
+# ---------------------------------------------------------------------------
+# Archive work list
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+class TestArchiveWorkList:
+    async def test_missing_mount_raises_during_work_list_build(self):
+        pid = uuid4()
+        service = MountsService(
+            mounts_dao=_MissingMountDAO(),
+            mounts_store=FakeMountStorage(),
+            bucket=_BUCKET,
+        )
+
+        with pytest.raises(MountNotFound):
+            await service.build_archive_work_list(
+                project_id=pid,
+                mounts=[(uuid4(), "", "")],
+            )
+
+    async def test_zip_paths_include_mount_prefix(self):
+        mount = _make_mount()
+        service, pid, mid = _make_service(mount)
+        for path in ["one.txt", "nested/two.txt"]:
+            await service.write_file(
+                project_id=pid,
+                mount_id=mid,
+                path=path,
+                content=b"x",
+            )
+
+        work = await service.build_archive_work_list(
+            project_id=pid,
+            mounts=[(mid, "prefix", "")],
+        )
+
+        assert {zip_path for zip_path, _key, _size, _mtime in work} == {
+            "prefix/one.txt",
+            "prefix/nested/two.txt",
+        }
+
+
 # ---------------------------------------------------------------------------
 # Roundtrip
 # ---------------------------------------------------------------------------
@@ -274,6 +410,70 @@ async def test_count_only_returns_total_no_files(self):
         assert listing.total_capped is False
         assert listing.files == []
 
+    async def test_raw_count_only_caps_total(self, monkeypatch):
+        monkeypatch.setattr(mounts_service_module, "_COUNT_CAP", 3)
+        mount = _make_mount()
+        service, pid, mid = _make_service(mount)
+        for path in ["a.txt", "b.txt", "c.txt", "d.txt", "e.txt"]:
+            await service.write_file(
+                project_id=pid, mount_id=mid, path=path, content=b"x"
+            )
+
+        listing = await service.list_files(
+            project_id=pid, mount_id=mid, limit=0, git_aware=False
+        )
+
+        assert listing.total == 3
+        assert listing.total_capped is True
+        assert listing.files == []
+
+    async def test_raw_count_only_reports_uncapped_total_below_cap(self, monkeypatch):
+        monkeypatch.setattr(mounts_service_module, "_COUNT_CAP", 3)
+        mount = _make_mount()
+        service, pid, mid = _make_service(mount)
+        for path in ["a.txt", "b.txt"]:
+            await service.write_file(
+                project_id=pid, mount_id=mid, path=path, content=b"x"
+            )
+
+        listing = await service.list_files(
+            project_id=pid, mount_id=mid, limit=0, git_aware=False
+        )
+
+        assert listing.total == 2
+        assert listing.total_capped is False
+        assert listing.files == []
+
+    async def test_raw_count_only_ignores_trailing_folder_markers(self, monkeypatch):
+        # Exactly `_COUNT_CAP` real files followed only by folder markers is an EXACT count, not a
+        # floor: the object-level `has_more` (markers still to page) must not report a false "N+".
+        monkeypatch.setattr(mounts_service_module, "_COUNT_CAP", 3)
+        mount = _make_mount()
+        storage = FakeMountStorage()
+        service = MountsService(
+            mounts_dao=_StubDAO(mount),
+            mounts_store=storage,
+            bucket=_BUCKET,
+        )
+        pid, mid = mount.project_id, mount.id
+        for path in ["a.txt", "b.txt", "c.txt"]:
+            await service.write_file(
+                project_id=pid, mount_id=mid, path=path, content=b"x"
+            )
+        # More markers than one store page (max(cap, 200)) so the first page reports has_more=True.
+        mount_base = service._storage_key(project_id=pid, mount=mount)
+        bucket_store = storage._store.setdefault(_BUCKET, {})
+        for i in range(250):
+            bucket_store[f"{mount_base}zzz{i:04}/"] = b""
+
+        listing = await service.list_files(
+            project_id=pid, mount_id=mid, limit=0, git_aware=False
+        )
+
+        assert listing.total == 3
+        assert listing.total_capped is False
+        assert listing.files == []
+
     async def test_shallow_depth_lists_top_level_only(self):
         # depth=1 → ONE delimiter level: top-level files + folders, no descent into subtrees. The
         # nested `sub/c.txt` surfaces its parent `sub` as a folder (never the deep file), and a