Python: fix(core): prevent path collisions between files and directories in InMemoryAgentFileStore

python/packages/core/agent_framework/_harness/_file_access.py

@@ -1067,12 +1067,24 @@ def __init__(self) -> None:
         # wrote, mirroring how :class:`FileSystemAgentFileStore` preserves the
         # on-disk casing.
         self._files: dict[str, tuple[str, str]] = {}
+        self._dir_counts: dict[str, int] = {}
+        self._explicit_dirs: dict[str, str] = {}
         self._lock = asyncio.Lock()
 
     @staticmethod
     def _key(path: str) -> str:
         return _normalize_relative_path(path).lower()
 
+    @staticmethod
+    def _ancestor_keys(key: str) -> list[str]:
+        """Return all strict ancestor directory keys in top-down order in linear time."""
+        ancestors: list[str] = []
+        idx = key.find("/")
+        while idx != -1:
+            ancestors.append(key[:idx])
+            idx = key.find("/", idx + 1)
+        return ancestors
+
     async def write(self, path: str, content: str, *, overwrite: bool = True) -> None:
         """Write ``content`` to the file at ``path``.
 
@@ -1083,9 +1095,19 @@ async def write(self, path: str, content: str, *, overwrite: bool = True) -> Non
         display = _normalize_relative_path(path)
         key = display.lower()
         async with self._lock:
+            ancestors = self._ancestor_keys(key)
+            for ancestor in ancestors:
+                if ancestor in self._files:
+                    raise NotADirectoryError(f"Not a directory: {path!r}")
+            if self._dir_counts.get(key, 0) > 0 or key in self._explicit_dirs:
+                raise IsADirectoryError(f"Is a directory: {path!r}")
             if not overwrite and key in self._files:
                 raise FileExistsError(f"File already exists: {path!r}")
+            is_new_file = key not in self._files
             self._files[key] = (display, content)
+            if is_new_file:
+                for ancestor in ancestors:
+                    self._dir_counts[ancestor] = self._dir_counts.get(ancestor, 0) + 1
 
     async def read(self, path: str) -> str | None:
         """Return the file content, or ``None`` if the file does not exist."""
@@ -1095,10 +1117,33 @@ async def read(self, path: str) -> str | None:
         return entry[1] if entry is not None else None
 
     async def delete(self, path: str) -> bool:
-        """Delete the file and return whether anything was removed."""
+        """Delete the file and return whether anything was removed.
+
+        An explicitly created directory can also be deleted, but only when it is
+        empty: nothing is stored under it and no other explicit directory is
+        nested below it. This gives session-scoped directories a lifecycle instead
+        of staying in the shared map forever.
+        """
         key = self._key(path)
         async with self._lock:
-            return self._files.pop(key, None) is not None
+            removed = self._files.pop(key, None)
+            if removed is not None:
+                for ancestor in self._ancestor_keys(key):
+                    new_count = self._dir_counts[ancestor] - 1
+                    if new_count <= 0:
+                        del self._dir_counts[ancestor]
+                    else:
+                        self._dir_counts[ancestor] = new_count
+                return True
+            if key in self._explicit_dirs:
+                if self._dir_counts.get(key, 0) > 0:
+                    return False
+                prefix = key + "/"
+                if any(other != key and other.startswith(prefix) for other in self._explicit_dirs):
+                    return False
+                del self._explicit_dirs[key]
+                return True
+            return False
 
     async def list_children(self, directory: str = "") -> list[FileStoreEntry]:
         """Return the direct child files and subdirectories of ``directory``.
@@ -1118,11 +1163,12 @@ async def list_children(self, directory: str = "") -> list[FileStoreEntry]:
             prefix += "/"
         prefix_depth = prefix.count("/")
         async with self._lock:
-            entries = [(key, display) for key, (display, _) in self._files.items()]
+            file_entries = [(key, display) for key, (display, _) in self._files.items()]
+            dir_entries = list(self._explicit_dirs.items())
         files: list[str] = []
         directories: list[str] = []
         seen_dirs: set[str] = set()
-        for key, display in entries:
+        for key, display in file_entries:
             if not key.startswith(prefix):
                 continue
             # Unicode lowercasing can change character counts, so key offsets
@@ -1137,6 +1183,17 @@ async def list_children(self, directory: str = "") -> list[FileStoreEntry]:
                     continue
                 seen_dirs.add(segment_key)
                 directories.append(segment)
+        for key, display in dir_entries:
+            if not key.startswith(prefix) or key == prefix.rstrip("/"):
+                continue
+            remainder = display.split("/", prefix_depth)[-1]
+            segment, _, _ = remainder.partition("/")
+            if segment:
+                segment_key = segment.lower()
+                if segment_key in seen_dirs:
+                    continue
+                seen_dirs.add(segment_key)
+                directories.append(segment)
         results: list[FileStoreEntry] = [FileStoreEntry(name, FileStoreEntry.DIRECTORY) for name in directories]
         results.extend(FileStoreEntry(name, FileStoreEntry.FILE) for name in files)
         return results
@@ -1194,8 +1251,30 @@ def scan() -> list[FileSearchResult]:
         return await _run_search_with_timeout(asyncio.to_thread(scan))
 
     async def create_directory(self, path: str) -> None:
-        """No-op: directories are implicit from file paths in the in-memory store."""
-        del path
+        """Ensure the directory at ``path`` exists."""
+        display = _normalize_relative_path(path, is_directory=True)
+        key = display.lower()
+        if not key:
+            return
+        async with self._lock:
+            ancestors = self._ancestor_keys(key)
+            for ancestor in ancestors:
+                if ancestor in self._files:
+                    raise FileExistsError(f"File already exists: {ancestor!r}")
+            if key in self._files:
+                raise FileExistsError(f"File already exists: {key!r}")
+            if key not in self._explicit_dirs:
+                idx_display = 0
+                for ancestor in ancestors:
+                    next_slash = display.find("/", idx_display)
+                    if next_slash != -1:
+                        ancestor_display = display[:next_slash]
+                        idx_display = next_slash + 1
+                    else:
+                        ancestor_display = display
+                    if ancestor not in self._explicit_dirs:
+                        self._explicit_dirs[ancestor] = ancestor_display
+                self._explicit_dirs[key] = display
 
 
 @experimental(feature_id=ExperimentalFeature.HARNESS)

python/packages/core/tests/core/test_harness_file_access.py

@@ -313,6 +313,115 @@ async def test_in_memory_store_normalizes_paths() -> None:
             await store.write(bad, "boom")
 
 
+async def test_in_memory_store_rejects_file_ancestor_collision() -> None:
+    """Writing a file whose ancestor is already a file must raise NotADirectoryError."""
+    store = InMemoryAgentFileStore()
+    await store.write("Reports", "file content")
+
+    with pytest.raises(NotADirectoryError):
+        await store.write("Reports/q1.txt", "nested content")
+
+    with pytest.raises(NotADirectoryError):
+        await store.write("reports/deep/q2.txt", "deep nested content")
+
+    assert await store.read("Reports") == "file content"
+    children = await store.list_children("")
+    assert len(children) == 1
+    assert children[0].name == "Reports"
+    assert children[0].type == FileStoreEntry.FILE
+
+
+async def test_in_memory_store_rejects_descendant_directory_collision() -> None:
+    """Writing a file whose path is already a directory must raise IsADirectoryError."""
+    store = InMemoryAgentFileStore()
+    await store.write("Reports/q1.txt", "nested content")
+
+    with pytest.raises(IsADirectoryError):
+        await store.write("Reports", "file content")
+
+    with pytest.raises(IsADirectoryError):
+        await store.write("reports", "case insensitive collision")
+
+    assert await store.read("Reports/q1.txt") == "nested content"
+    assert await store.read("Reports") is None
+    children = await store.list_children("")
+    assert len(children) == 1
+    assert children[0].name == "Reports"
+    assert children[0].type == FileStoreEntry.DIRECTORY
+
+
+async def test_in_memory_store_create_directory_rejects_existing_file() -> None:
+    """Creating a directory that collides with an existing file must raise FileExistsError."""
+    store = InMemoryAgentFileStore()
+    await store.write("Reports", "file content")
+
+    with pytest.raises(FileExistsError):
+        await store.create_directory("Reports")
+
+    with pytest.raises(FileExistsError):
+        await store.create_directory("reports/nested")
+
+
+async def test_in_memory_store_explicit_directory_reserves_path_against_write() -> None:
+    """Explicitly created directory must reserve the path and reject file writes."""
+    store = InMemoryAgentFileStore()
+    await store.create_directory("Reports")
+
+    with pytest.raises(IsADirectoryError):
+        await store.write("Reports", "file content")
+
+    with pytest.raises(IsADirectoryError):
+        await store.write("reports", "case insensitive collision")
+
+    children = await store.list_children("")
+    assert len(children) == 1
+    assert children[0].name == "Reports"
+    assert children[0].type == FileStoreEntry.DIRECTORY
+
+    nested_children = await store.list_children("Reports")
+    assert len(nested_children) == 0
+
+
+async def test_in_memory_store_explicit_nested_directory_creation() -> None:
+    """Explicit nested directory creation registers intermediate directories and preserves casing."""
+    store = InMemoryAgentFileStore()
+    await store.create_directory("Reports/2026/Q1")
+
+    root_children = await store.list_children("")
+    assert len(root_children) == 1
+    assert root_children[0].name == "Reports"
+    assert root_children[0].type == FileStoreEntry.DIRECTORY
+
+    reports_children = await store.list_children("Reports")
+    assert len(reports_children) == 1
+    assert reports_children[0].name == "2026"
+    assert reports_children[0].type == FileStoreEntry.DIRECTORY
+
+    q1_children = await store.list_children("Reports/2026")
+    assert len(q1_children) == 1
+    assert q1_children[0].name == "Q1"
+    assert q1_children[0].type == FileStoreEntry.DIRECTORY
+
+    with pytest.raises(IsADirectoryError):
+        await store.write("reports/2026", "collision")
+
+
+async def test_in_memory_store_delete_file_cleans_up_implicit_directory_counts() -> None:
+    """Deleting a file unreserves implicit parent directories when no other files remain."""
+    store = InMemoryAgentFileStore()
+    await store.write("data/temp/file.txt", "file content")
+
+    with pytest.raises(IsADirectoryError):
+        await store.write("data/temp", "should fail while file exists")
+
+    deleted = await store.delete("data/temp/file.txt")
+    assert deleted is True
+
+    # Now that the implicit descendant is gone and temp was never explicitly created, writing succeeds
+    await store.write("data/temp", "now a file")
+    assert await store.read("data/temp") == "now a file"
+
+
 async def test_filesystem_store_round_trips_files(tmp_path: Path) -> None:
     """The filesystem store should round-trip files on disk and create parents on write."""
     store = FileSystemAgentFileStore(tmp_path)
@@ -2610,3 +2719,20 @@ async def test_scope_alone_enables_scoped_mode(chat_client_base: SupportsChatGet
     provider = FileAccessProvider(store=store, scope="tenant-1")
     context = SessionContext(session_id=None, input_messages=[])
     await provider.before_run(agent=None, session=AgentSession(), context=context, state={})
+
+
+async def test_in_memory_store_delete_removes_empty_explicit_directory() -> None:
+    """An empty explicitly created directory can be deleted; a used one cannot."""
+    store = InMemoryAgentFileStore()
+    await store.create_directory("sessions/s1")
+
+    assert await store.delete("sessions/s1") is True
+    children = await store.list_children("sessions")
+    assert all(entry.name != "s1" for entry in children)
+
+    await store.create_directory("sessions/s2")
+    await store.write("sessions/s2/state.txt", "data")
+    assert await store.delete("sessions/s2") is False
+    assert await store.delete("sessions/s2/state.txt") is True
+    assert await store.delete("sessions/s2") is True
+    assert await store.delete("sessions") is True

python/packages/core/tests/core/test_harness_file_memory.py

@@ -307,8 +307,11 @@ async def test_tools_reject_nested_paths() -> None:
 
     saved = await tools["file_memory_write"].invoke(arguments={"file_name": "notes/plan.md", "content": "x"})
     assert "subdirectory" in _text(saved)
-    # Nothing should have been written for the nested name.
-    assert await store.list_children("") == []
+    # Nothing should have been written for the nested name. The session
+    # directory created during before_run is the only root entry, and the
+    # rejected write leaves it empty (no "notes" child).
+    assert [entry.name for entry in await store.list_children("")] == ["session-1"]
+    assert await store.list_children("session-1") == []
 
     # Backslash separators are normalized to "/" and rejected the same way.
     saved_backslash = await tools["file_memory_write"].invoke(arguments={"file_name": "notes\\plan.md", "content": "x"})