fix(api): resolve zip-namespace collisions when merging drive mounts

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

@@ -54,6 +54,12 @@
 # The single session-bound mount: the agent's durable working directory.
 _SESSION_CWD_NAME = "cwd"
 
+# The name the runner symlinks the agent mount into the cwd as (runner: `AGENT_FILES_LINK_NAME`).
+# The runner unlinks whatever else sits at that path on every attach, so the name is reserved AT THE
+# CWD MOUNT ROOT and nowhere else. geesefs degrades the symlink to a plain object of the same name in
+# the cwd store; that object is a runner artifact, never user content.
+_AGENT_FILES_LINK_NAME = "agent-files"
+
 # Default TTL (seconds) for signed mount credentials. Covers the mount lifetime for a
 # turn; geesefs holds the creds without refresh, so a turn outliving this hits ExpiredToken.
 _CREDENTIALS_TTL_SECONDS = 3600
@@ -151,6 +157,50 @@ def _safe_zip_segments(path: str) -> List[str]:
     return [s for s in _zip_segments(path) if s not in ("", ".", "..")]
 
 
+def _is_session_cwd_mount(mount: Mount) -> bool:
+    """The session's durable working directory — the one mount the runner links `agent-files` into."""
+    return mount.session_id is not None and mount.name == _SESSION_CWD_NAME
+
+
+def _resolve_zip_namespace(
+    work: List[Tuple[str, str, int, Optional[int]]],
+) -> List[Tuple[str, str, int, Optional[int]]]:
+    """Drop members that cannot coexist in ONE zip namespace.
+
+    The archive carries no directory entries: `a/b` IMPLIES the directory `a`, so a member that is a
+    FILE named `a` blocks it and everything under `a/` is lost on extraction. An object store holds
+    `a` and `a/b` side by side happily, and the drive merges several mounts into one zip (cwd at the
+    root, the agent mount under `agent-files/`), so the collision is reachable across sources — it is
+    how the degraded `agent-files` symlink swallowed the agent's files (#5482). Directories win; a
+    duplicate path keeps the first member.
+    """
+    directories: set[str] = set()
+    for zip_path, *_rest in work:
+        segments = zip_path.split("/")
+        for i in range(1, len(segments)):
+            directories.add("/".join(segments[:i]))
+
+    resolved: List[Tuple[str, str, int, Optional[int]]] = []
+    seen: set[str] = set()
+    for member in work:
+        zip_path = member[0]
+        if zip_path in directories:
+            log.warning(
+                "mounts.archive: skipping file member shadowed by a directory",
+                zip_path=zip_path,
+            )
+            continue
+        if zip_path in seen:
+            log.warning(
+                "mounts.archive: skipping duplicate member",
+                zip_path=zip_path,
+            )
+            continue
+        seen.add(zip_path)
+        resolved.append(member)
+    return resolved
+
+
 def _is_internal_mount_path(path: str) -> bool:
     """Runner-owned runtime artifacts written into the durable cwd — hidden from flat file listings.
     Mirrors the web `isInternalDrivePath`: the whole `agents/` namespace plus `.agenta-*` markers."""
@@ -1200,6 +1250,10 @@ async def build_archive_work_list(
         all"); ``zip_prefix`` places its files under ``prefix/`` in the zip (e.g. "agent-files" for
         the folded agent mount). Folder markers are skipped. Each work item is a
         ``(zip_path, storage_key, size, mtime)`` tuple.
+
+        The sources share ONE zip namespace, so the merged list is passed through
+        ``_resolve_zip_namespace`` — a member the store allows but a zip cannot carry alongside its
+        neighbours (a file shadowing a directory, a duplicate path) is dropped there.
         """
         bucket = self._bucket()
 
@@ -1213,6 +1267,7 @@ async def build_archive_work_list(
                 project_id=project_id, mount_id=source.mount_id
             )
             mount_base = self._storage_key(project_id=project_id, mount=mount)
+            is_session_cwd = _is_session_cwd_mount(mount)
             pfx_segments = _safe_zip_segments(source.archive_prefix)
             src = source.source_path.strip("/")
             # Scope the listing to a folder when `source_path` is set (folder download); the
@@ -1229,6 +1284,11 @@ async def build_archive_work_list(
                     if obj.key.startswith(mount_base)
                     else obj.key
                 )
+                # The runner's `agent-files` link, degraded by geesefs into an object in the cwd
+                # store. A runner artifact, not user content (the runner unlinks anything else at
+                # that path), and the drive hides it — so it has no place in the zip either.
+                if is_session_cwd and rel.strip("/") == _AGENT_FILES_LINK_NAME:
+                    continue
                 rel_segments = _zip_segments(rel)
                 # Store keys come from signed-credential writers, so `rel` can carry `..` or a
                 # backslash. Don't REWRITE such a key — `a/../report.txt` would collapse onto a real
@@ -1242,7 +1302,7 @@ async def build_archive_work_list(
                 zip_path = "/".join([*pfx_segments, *rel_segments])
                 work.append((zip_path, obj.key, obj.size or 0, obj.mtime))
 
-        return work
+        return _resolve_zip_namespace(work)
 
     async def iter_archive_members(
         self,

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

@@ -23,6 +23,7 @@
 from oss.src.core.mounts.service import (
     MountsService,
     _rollup_recent_entries,
+    mint_session_slug,
     validate_file_path,
 )
 from oss.src.core.store.dtos import StoreObject
@@ -276,19 +277,44 @@ async def fetch_mount(self, *, project_id, mount_id):
         return self._mount
 
 
+class _MultiMountDAO:
+    """Resolves several mounts by id — an archive folds the cwd and agent mounts into one zip."""
+
+    def __init__(self, *mounts: Mount):
+        self._by_id = {mount.id: mount for mount in mounts}
+
+    async def fetch_mount(self, *, project_id, mount_id):
+        return self._by_id.get(mount_id)
+
+
 class _MissingMountDAO:
     async def fetch_mount(self, *, project_id, mount_id):
         return None
 
 
-def _make_mount() -> Mount:
+_PROJECT_ID = UUID("00000000-0000-0000-0000-0000000000a9")
+
+
+def _make_mount(project_id: UUID = _PROJECT_ID) -> Mount:
     return Mount(
         id=uuid4(),
-        project_id=uuid4(),
+        project_id=project_id,
         slug="m",
     )
 
 
+def _make_cwd_mount(project_id: UUID = _PROJECT_ID) -> Mount:
+    """The session's durable working directory — where the runner links `agent-files` in."""
+    session_id = str(uuid4())
+    return Mount(
+        id=uuid4(),
+        project_id=project_id,
+        session_id=session_id,
+        slug=mint_session_slug(session_id=session_id, name="cwd"),
+        name="cwd",
+    )
+
+
 def _make_service(mount: Mount) -> Tuple[MountsService, UUID, UUID]:
     service = MountsService(
         mounts_dao=_StubDAO(mount),
@@ -377,6 +403,124 @@ async def test_traversal_key_does_not_alias_a_real_entry(self):
         assert zip_paths == ["a/report.txt"]
 
 
+@pytest.mark.asyncio
+class TestArchiveZipNamespace:
+    """One zip namespace across sources: a FILE member may not shadow a DIRECTORY, and a path may
+    not appear twice. The store allows both; extraction does not."""
+
+    async def _work(self, sources, contents):
+        # `contents`: [(mount, {rel_path: bytes})]; `sources`: the archive request, in order.
+        storage = FakeMountStorage()
+        service = MountsService(
+            mounts_dao=_MultiMountDAO(*(mount for mount, _files in contents)),
+            mounts_store=storage,
+            bucket=_BUCKET,
+        )
+        bucket_store = storage._store.setdefault(_BUCKET, {})
+        for mount, files in contents:
+            base = service._storage_key(project_id=_PROJECT_ID, mount=mount)
+            for rel, body in files.items():
+                bucket_store[f"{base}{rel}"] = body
+
+        return await service.build_archive_work_list(
+            project_id=_PROJECT_ID,
+            mounts=sources,
+        )
+
+    async def test_degraded_agent_files_link_does_not_block_the_folded_mount(self):
+        # The real "download all" shape (#5482): the cwd source carries the runner's `agent-files`
+        # link — which geesefs degraded into a plain 0-byte object — while the agent source folds its
+        # content in under the same name. The link must not ship; the agent's files must.
+        cwd, agent = _make_cwd_mount(), _make_mount()
+
+        work = await self._work(
+            sources=[
+                MountArchiveSource(mount_id=cwd.id),
+                MountArchiveSource(mount_id=agent.id, archive_prefix="agent-files"),
+            ],
+            contents=[
+                (cwd, {"notes.md": b"notes", "agent-files": b""}),
+                (agent, {"plan.md": b"plan", "sub/deep.md": b"deep"}),
+            ],
+        )
+
+        assert {zip_path for zip_path, *_rest in work} == {
+            "notes.md",
+            "agent-files/plan.md",
+            "agent-files/sub/deep.md",
+        }
+
+    async def test_degraded_link_is_dropped_even_with_an_empty_agent_mount(self):
+        # Nothing collides here, so only the reserved-at-the-cwd-root rule can drop the link. It
+        # still must: it is a runner artifact the drive hides, not a file the user put there.
+        cwd, agent = _make_cwd_mount(), _make_mount()
+
+        work = await self._work(
+            sources=[
+                MountArchiveSource(mount_id=cwd.id),
+                MountArchiveSource(mount_id=agent.id, archive_prefix="agent-files"),
+            ],
+            contents=[(cwd, {"notes.md": b"notes", "agent-files": b""}), (agent, {})],
+        )
+
+        assert {zip_path for zip_path, *_rest in work} == {"notes.md"}
+
+    async def test_agent_files_is_not_a_reserved_name_outside_the_cwd_root(self):
+        # The runner only clobbers `agent-files` at the CWD ROOT. A real file of that name anywhere
+        # else is user content and must survive — including at the root of the agent mount itself,
+        # which the drive presents as `agent-files/agent-files`.
+        cwd, agent = _make_cwd_mount(), _make_mount()
+
+        work = await self._work(
+            sources=[
+                MountArchiveSource(mount_id=cwd.id),
+                MountArchiveSource(mount_id=agent.id, archive_prefix="agent-files"),
+            ],
+            contents=[
+                (cwd, {"docs/agent-files": b"about the agent files"}),
+                (agent, {"agent-files": b"a real file"}),
+            ],
+        )
+
+        assert {zip_path for zip_path, *_rest in work} == {
+            "docs/agent-files",
+            "agent-files/agent-files",
+        }
+
+    async def test_file_shadowing_a_directory_is_dropped_whatever_its_name(self):
+        # Nothing here is agent-files-specific: the store holds `report` and `report/q1.csv` side by
+        # side, but a zip cannot — the file would block the directory on extraction.
+        mount = _make_mount()
+
+        work = await self._work(
+            sources=[MountArchiveSource(mount_id=mount.id)],
+            contents=[(mount, {"report": b"stale", "report/q1.csv": b"rows"})],
+        )
+
+        assert {zip_path for zip_path, *_rest in work} == {"report/q1.csv"}
+
+    async def test_duplicate_zip_path_keeps_the_first_source(self):
+        # Two sources landing on one path would write the member twice; extractors silently keep the
+        # last. Ship the first and drop the rest so the zip stays deterministic.
+        first, second = _make_mount(), _make_mount()
+
+        work = await self._work(
+            sources=[
+                MountArchiveSource(mount_id=first.id),
+                MountArchiveSource(mount_id=second.id),
+            ],
+            contents=[
+                (first, {"notes.md": b"first"}),
+                (second, {"notes.md": b"second"}),
+            ],
+        )
+
+        assert len(work) == 1
+        assert work[0][1].startswith(f"mounts/{_PROJECT_ID}/{first.id}/"), (
+            "the first source's key must win"
+        )
+
+
 # ---------------------------------------------------------------------------
 # Roundtrip
 # ---------------------------------------------------------------------------