From 05f1554d6486fb9c4c6825c9ed0c1bb36e62c729 Mon Sep 17 00:00:00 2001
From: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com>
Date: Wed, 23 Sep 2026 18:49:36 +0530
Subject: [PATCH 1/6] Python: fix(core): prevent path collisions between files
 and directories in InMemoryAgentFileStore

Signed-off-by: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com>
---
 .../agent_framework/_harness/_file_access.py  | 22 ++++++++-
 .../tests/core/test_harness_file_access.py    | 49 +++++++++++++++++++
 2 files changed, 69 insertions(+), 2 deletions(-)

diff --git a/python/packages/core/agent_framework/_harness/_file_access.py b/python/packages/core/agent_framework/_harness/_file_access.py
index 6191bb98f7d..2b0e885d143 100644
--- a/python/packages/core/agent_framework/_harness/_file_access.py
+++ b/python/packages/core/agent_framework/_harness/_file_access.py
@@ -1079,7 +1079,16 @@ async def write(self, path: str, content: str, *, overwrite: bool = True) -> Non
         """
         display = _normalize_relative_path(path)
         key = display.lower()
+        dir_prefix = f"{key}/"
         async with self._lock:
+            parts = key.split("/")
+            for i in range(1, len(parts)):
+                ancestor = "/".join(parts[:i])
+                if ancestor in self._files:
+                    raise NotADirectoryError(f"Not a directory: {path!r}")
+            for existing_key in self._files:
+                if existing_key.startswith(dir_prefix):
+                    raise IsADirectoryError(f"Is a directory: {path!r}")
             if not overwrite and key in self._files:
                 raise FileExistsError(f"File already exists: {path!r}")
             self._files[key] = (display, content)
@@ -1191,8 +1200,17 @@ 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:
+            parts = key.split("/")
+            for i in range(1, len(parts) + 1):
+                ancestor = "/".join(parts[:i])
+                if ancestor in self._files:
+                    raise FileExistsError(f"File already exists: {ancestor!r}")
 
 
 @experimental(feature_id=ExperimentalFeature.HARNESS)
diff --git a/python/packages/core/tests/core/test_harness_file_access.py b/python/packages/core/tests/core/test_harness_file_access.py
index 8c01b1c6146..f0f97a64a55 100644
--- a/python/packages/core/tests/core/test_harness_file_access.py
+++ b/python/packages/core/tests/core/test_harness_file_access.py
@@ -308,6 +308,55 @@ 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_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)

From 9b05a0f91616add523e40fc92b920ea8a31617bf Mon Sep 17 00:00:00 2001
From: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com>
Date: Fri, 25 Sep 2026 18:43:36 +0530
Subject: [PATCH 2/6] Index directory prefixes and persist explicit directories
 in InMemoryAgentFileStore

Signed-off-by: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com>
---
 .../agent_framework/_harness/_file_access.py  | 73 +++++++++++++++----
 .../tests/core/test_harness_file_access.py    | 60 +++++++++++++++
 2 files changed, 120 insertions(+), 13 deletions(-)

diff --git a/python/packages/core/agent_framework/_harness/_file_access.py b/python/packages/core/agent_framework/_harness/_file_access.py
index 2b0e885d143..5c256365a45 100644
--- a/python/packages/core/agent_framework/_harness/_file_access.py
+++ b/python/packages/core/agent_framework/_harness/_file_access.py
@@ -1064,12 +1064,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``.
 
@@ -1079,19 +1091,20 @@ async def write(self, path: str, content: str, *, overwrite: bool = True) -> Non
         """
         display = _normalize_relative_path(path)
         key = display.lower()
-        dir_prefix = f"{key}/"
         async with self._lock:
-            parts = key.split("/")
-            for i in range(1, len(parts)):
-                ancestor = "/".join(parts[:i])
+            ancestors = self._ancestor_keys(key)
+            for ancestor in ancestors:
                 if ancestor in self._files:
                     raise NotADirectoryError(f"Not a directory: {path!r}")
-            for existing_key in self._files:
-                if existing_key.startswith(dir_prefix):
-                    raise IsADirectoryError(f"Is 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."""
@@ -1104,7 +1117,16 @@ async def delete(self, path: str) -> bool:
         """Delete the file and return whether anything was removed."""
         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
+            return False
 
     async def list_children(self, directory: str = "") -> list[FileStoreEntry]:
         """Return the direct child files and subdirectories of ``directory``.
@@ -1124,11 +1146,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
@@ -1143,6 +1166,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
@@ -1206,11 +1240,24 @@ async def create_directory(self, path: str) -> None:
         if not key:
             return
         async with self._lock:
-            parts = key.split("/")
-            for i in range(1, len(parts) + 1):
-                ancestor = "/".join(parts[:i])
+            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)
diff --git a/python/packages/core/tests/core/test_harness_file_access.py b/python/packages/core/tests/core/test_harness_file_access.py
index f0f97a64a55..dd30a181168 100644
--- a/python/packages/core/tests/core/test_harness_file_access.py
+++ b/python/packages/core/tests/core/test_harness_file_access.py
@@ -357,6 +357,66 @@ async def test_in_memory_store_create_directory_rejects_existing_file() -> None:
         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)

From 92785b5d75728d4eaf8b3f8a568aa3cf211d0317 Mon Sep 17 00:00:00 2001
From: Manohar Paturi <manoharpaturi@users.noreply.github.com>
Date: Fri, 2 Oct 2026 15:44:19 +0530
Subject: [PATCH 3/6] test(harness): account for session dir in nested-path
 rejection test

The upstream #8507 assertion expected list_children("") to stay empty,
which held while create_directory was a no-op. This PR makes explicit
directories listable, so the session-1 directory created during
before_run now appears at the root. The invariant the test protects,
that a rejected nested write leaves nothing behind, is preserved: the
session directory exists and stays empty, with no notes child.
---
 .../packages/core/tests/core/test_harness_file_memory.py   | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/python/packages/core/tests/core/test_harness_file_memory.py b/python/packages/core/tests/core/test_harness_file_memory.py
index 4f2e810c858..c5985df4032 100644
--- a/python/packages/core/tests/core/test_harness_file_memory.py
+++ b/python/packages/core/tests/core/test_harness_file_memory.py
@@ -305,8 +305,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"})

From b4bf58ee38a52ebe2a147021b78c7bdc7cdcbf62 Mon Sep 17 00:00:00 2001
From: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com>
Date: Wed, 14 Oct 2026 14:03:01 +0530
Subject: [PATCH 4/6] Python: let the in-memory store delete empty explicit
 directories

Explicitly created directories had no deletion path, so session-scoped
runs kept accumulating entries in the shared map. delete() now removes
an explicit directory once nothing is stored under it and no other
explicit directory is nested below, giving session directories a
lifecycle. Non-empty directories stay put, matching the filesystem
store's file-only deletes.

Signed-off-by: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com>
---
 .../agent_framework/_harness/_file_access.py    | 16 +++++++++++++++-
 .../core/tests/core/test_harness_file_access.py | 17 +++++++++++++++++
 2 files changed, 32 insertions(+), 1 deletion(-)

diff --git a/python/packages/core/agent_framework/_harness/_file_access.py b/python/packages/core/agent_framework/_harness/_file_access.py
index 57633afeb15..b98e029c267 100644
--- a/python/packages/core/agent_framework/_harness/_file_access.py
+++ b/python/packages/core/agent_framework/_harness/_file_access.py
@@ -1117,7 +1117,13 @@ 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:
             removed = self._files.pop(key, None)
@@ -1129,6 +1135,14 @@ async def delete(self, path: str) -> bool:
                     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]:
diff --git a/python/packages/core/tests/core/test_harness_file_access.py b/python/packages/core/tests/core/test_harness_file_access.py
index 14b26ba50cd..6f007812f7d 100644
--- a/python/packages/core/tests/core/test_harness_file_access.py
+++ b/python/packages/core/tests/core/test_harness_file_access.py
@@ -2719,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

From a0fdd60c8adb0e205b9876fcc2c645ee0402529c Mon Sep 17 00:00:00 2001
From: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com>
Date: Thu, 22 Oct 2026 15:48:21 +0000
Subject: [PATCH 5/6] Python: keep delete file-only and add delete_directory to
 AgentFileStore

Signed-off-by: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com>
---
 .../agent_framework/_harness/_file_access.py  | 93 ++++++++++++++-----
 .../tests/core/test_harness_file_access.py    | 74 +++++++++++++--
 2 files changed, 140 insertions(+), 27 deletions(-)

diff --git a/python/packages/core/agent_framework/_harness/_file_access.py b/python/packages/core/agent_framework/_harness/_file_access.py
index b98e029c267..b0139827a3b 100644
--- a/python/packages/core/agent_framework/_harness/_file_access.py
+++ b/python/packages/core/agent_framework/_harness/_file_access.py
@@ -1048,6 +1048,23 @@ def scan_batch(pending: list[tuple[str, str]]) -> list[FileSearchResult]:
     async def create_directory(self, path: str) -> None:
         """Ensure ``path`` exists as a directory, creating it if necessary."""
 
+    async def delete_directory(self, path: str) -> bool:
+        """Delete the empty directory at ``path``.
+
+        Args:
+            path: The relative path of the directory to delete.
+
+        Returns:
+            ``True`` if an empty directory was removed; ``False`` if it did not
+            exist, was not a directory, or still held entries. Backends that
+            cannot remove directories return ``False`` rather than raising.
+
+        ``delete`` stays file-only so backends remain interchangeable; this is
+        the supported way to release an explicitly created directory (for
+        example a session-scoped working folder) once it is empty.
+        """
+        return False
+
 
 @experimental(feature_id=ExperimentalFeature.HARNESS)
 class InMemoryAgentFileStore(AgentFileStore):
@@ -1119,31 +1136,46 @@ async def read(self, path: str) -> str | None:
     async def delete(self, path: str) -> bool:
         """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.
+        Deleting is file-only, matching :class:`FileSystemAgentFileStore`: a
+        path that names a directory (explicitly created or holding stored
+        descendants) is never removed and returns ``False``. Use
+        :meth:`delete_directory` to release an empty explicitly created
+        directory.
         """
         key = self._key(path)
         async with self._lock:
+            if key in self._explicit_dirs or self._dir_counts.get(key, 0) > 0:
+                return False
             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
+            if removed is None:
+                return False
+            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
+
+    async def delete_directory(self, path: str) -> bool:
+        """Delete the empty explicitly created directory at ``path``.
+
+        The directory is removed only when it is empty: nothing is stored under
+        it and no other explicit directory is nested below it. This keeps the
+        file-only ``delete`` contract while giving session-scoped directories a
+        lifecycle instead of staying in the shared map forever.
+        """
+        key = self._key(_normalize_relative_path(path, is_directory=True))
+        async with self._lock:
+            if key not in self._explicit_dirs:
+                return False
+            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
 
     async def list_children(self, directory: str = "") -> list[FileStoreEntry]:
         """Return the direct child files and subdirectories of ``directory``.
@@ -1523,6 +1555,25 @@ def _delete_file_sync(cls, full_path: Path) -> bool:
                 return False
         return True
 
+    async def delete_directory(self, path: str) -> bool:
+        """Delete the empty directory at ``path``.
+
+        Returns ``True`` only when an empty directory was removed; a directory
+        that is missing, is a file, or still holds entries returns ``False``
+        instead of raising, matching :class:`InMemoryAgentFileStore`.
+        """
+        full_path = self._resolve_safe_directory_path(path)
+        return await asyncio.to_thread(self._delete_directory_sync, full_path)
+
+    @staticmethod
+    def _delete_directory_sync(full_path: Path) -> bool:
+        try:
+            full_path.rmdir()
+        except OSError:
+            # Missing path, a file, or a directory that still holds entries.
+            return False
+        return True
+
     async def list_children(self, directory: str = "") -> list[FileStoreEntry]:
         """Return the direct child files and subdirectories of ``directory``.
 
diff --git a/python/packages/core/tests/core/test_harness_file_access.py b/python/packages/core/tests/core/test_harness_file_access.py
index 6f007812f7d..b089717145e 100644
--- a/python/packages/core/tests/core/test_harness_file_access.py
+++ b/python/packages/core/tests/core/test_harness_file_access.py
@@ -2721,18 +2721,80 @@ async def test_scope_alone_enables_scoped_mode(chat_client_base: SupportsChatGet
     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."""
+async def test_filesystem_delete_is_file_only_and_delete_directory_matches_in_memory(
+    tmp_path: Path,
+) -> None:
+    """FileSystemAgentFileStore.delete must reject directories exactly like the
+    in-memory backend, and delete_directory must release an empty directory."""
+
+    async def in_memory_mirror() -> tuple[InMemoryAgentFileStore, FileSystemAgentFileStore]:
+        memory_store = InMemoryAgentFileStore()
+        fs_root = tmp_path / "root"
+        fs_store = FileSystemAgentFileStore(fs_root)
+        for store in (memory_store, fs_store):
+            await store.create_directory("sessions/s1")
+            await store.write("sessions/s1/state.txt", "data")
+        return memory_store, fs_store
+
+    memory_store, fs_store = await in_memory_mirror()
+
+    # Backend parity: deleting a directory returns False on both backends.
+    assert await fs_store.delete("sessions/s1") is False
+    assert await memory_store.delete("sessions/s1") is False
+
+    # Deleting the file first leaves the (now empty) directory in place.
+    assert await fs_store.delete("sessions/s1/state.txt") is True
+    assert await memory_store.delete("sessions/s1/state.txt") is True
+    for store in (fs_store, memory_store):
+        assert [entry.name for entry in await store.list_children("sessions/s1")] == []
+
+    # delete_directory releases the empty directory on both backends.
+    assert await fs_store.delete_directory("sessions/s1") is True
+    assert await memory_store.delete_directory("sessions/s1") is True
+    for store in (fs_store, memory_store):
+        assert await store.list_children("sessions") == []
+
+    # A second release of the same path reports False on both backends.
+    assert await fs_store.delete_directory("sessions/s1") is False
+    assert await memory_store.delete_directory("sessions/s1") is False
+
+
+async def test_in_memory_store_delete_is_file_only() -> None:
+    """delete() must stay file-only: a directory path is never removed, matching
+    FileSystemAgentFileStore.delete on the same input (backend parity)."""
     store = InMemoryAgentFileStore()
     await store.create_directory("sessions/s1")
 
-    assert await store.delete("sessions/s1") is True
+    assert await store.delete("sessions/s1") is False
+    assert await store.delete("sessions/s1/state.txt") is False
     children = await store.list_children("sessions")
-    assert all(entry.name != "s1" for entry in children)
+    assert [entry.name for entry in children] == ["s1"]
 
     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
+    assert await store.delete("sessions/s2") is False
+    assert await store.delete("sessions") is False
+
+    # The emptied directory stays listed, exactly like an empty directory on a filesystem.
+    entries = await store.list_children("sessions")
+    assert [entry.name for entry in entries] == ["s1", "s2"]
+
+
+async def test_in_memory_store_delete_directory_releases_empty_explicit_directory() -> None:
+    """delete_directory() releases an empty explicitly created directory so a
+    session-scoped working folder does not stay in the shared map forever."""
+    store = InMemoryAgentFileStore()
+    await store.create_directory("sessions/s1")
+
+    assert await store.delete_directory("sessions/s1") is True
+    children = await store.list_children("sessions")
+    assert children == []
+
+    await store.create_directory("sessions/s2")
+    await store.write("sessions/s2/state.txt", "data")
+    assert await store.delete_directory("sessions/s2") is False
+    assert await store.delete("sessions/s2/state.txt") is True
+    assert await store.delete_directory("sessions/s2") is True
+    assert await store.delete_directory("sessions") is True

From dacf0062a1adec382d27773e3769cca01e775311 Mon Sep 17 00:00:00 2001
From: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com>
Date: Sat, 24 Oct 2026 11:48:10 +0530
Subject: [PATCH 6/6] Python: bound store path complexity before materializing
 ancestor prefixes

The in-memory store's directory index retains the full key of every ancestor directory, so an unbounded path let a single write hold quadratic space in the path depth (an 8000 deep path retained roughly 64 million characters of prefix strings). The shared normalizer now refuses paths beyond 255 segments or 4096 characters with ENAMETOOLONG, the same errno the filesystem backend's OS layer returns, so the bound is checked before any prefix is allocated and is identical across backends. Deep but realistic paths keep working, and regression tests cover the rejection, the empty index afterward, and backend parity on both sides of the bound.

Signed-off-by: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com>
---
 .../agent_framework/_harness/_file_access.py  | 16 +++++
 .../tests/core/test_harness_file_access.py    | 68 +++++++++++++++++++
 2 files changed, 84 insertions(+)

diff --git a/python/packages/core/agent_framework/_harness/_file_access.py b/python/packages/core/agent_framework/_harness/_file_access.py
index 97c1bb7a244..f9cc46a2c4a 100644
--- a/python/packages/core/agent_framework/_harness/_file_access.py
+++ b/python/packages/core/agent_framework/_harness/_file_access.py
@@ -90,6 +90,15 @@
 # regex match when building a result snippet.
 _SEARCH_SNIPPET_RADIUS = 50
 
+# Path complexity bounds shared by every store backend. Real filesystems cap
+# these too (PATH_MAX and practical directory depth); the in-memory store's
+# directory index materializes the full key of every ancestor directory, so an
+# unbounded path would let a single write retain quadratic space in the path
+# depth. Bounding here keeps that retention constant regardless of input, and
+# rejects the same extremes the filesystem backend's OS layer would refuse.
+_MAX_PATH_SEGMENTS: Final[int] = 255
+_MAX_PATH_LENGTH: Final[int] = 4096
+
 # Hard cap on the length of a user-supplied search regex. The cap is a coarse bound on
 # how much work a single pattern can describe; it does not stop a short pathological
 # pattern, which is what :class:`_BoundedSearchPattern` is for.
@@ -296,6 +305,10 @@ def _normalize_relative_path(path: str, *, is_directory: bool = False) -> str:
         ValueError: When the path is rooted, starts with a drive letter, contains
             ``.``/``..`` segments, is empty for a file path, or ends with a
             separator for a file path.
+        OSError: With :data:`errno.ENAMETOOLONG` when the normalized path
+            exceeds ``_MAX_PATH_SEGMENTS`` segments or ``_MAX_PATH_LENGTH``
+            characters, matching what a real filesystem raises for the same
+            input.
     """
     if not path or not path.strip():
         if not is_directory:
@@ -333,6 +346,9 @@ def _normalize_relative_path(path: str, *, is_directory: bool = False) -> str:
     result = "/".join(clean_segments)
     if not is_directory and not result:
         raise ValueError(f"Invalid path: {path!r}. A file path must not be empty.")
+    if len(clean_segments) > _MAX_PATH_SEGMENTS or len(result) > _MAX_PATH_LENGTH:
+        # Check before any caller materializes ancestor prefixes of this path.
+        raise OSError(errno.ENAMETOOLONG, "File name too long", path)
     return result
 
 
diff --git a/python/packages/core/tests/core/test_harness_file_access.py b/python/packages/core/tests/core/test_harness_file_access.py
index b089717145e..7ee996016b1 100644
--- a/python/packages/core/tests/core/test_harness_file_access.py
+++ b/python/packages/core/tests/core/test_harness_file_access.py
@@ -2798,3 +2798,71 @@ async def test_in_memory_store_delete_directory_releases_empty_explicit_director
     assert await store.delete("sessions/s2/state.txt") is True
     assert await store.delete_directory("sessions/s2") is True
     assert await store.delete_directory("sessions") is True
+
+
+async def test_in_memory_store_rejects_paths_beyond_the_complexity_bound() -> None:
+    """A path deeper than the shared bound is rejected before any ancestor
+    prefix is materialized, so one write cannot retain quadratic memory in the
+    path depth. The error matches what the OS raises for the same input on the
+    filesystem backend (ENAMETOOLONG)."""
+    import errno
+
+    store = InMemoryAgentFileStore()
+    path = "/".join(["a"] * 8000 + ["f"])
+
+    with pytest.raises(OSError) as exc_info:
+        await store.write(path, "x")
+
+    assert exc_info.value.errno == errno.ENAMETOOLONG
+    # Nothing was stored and no ancestor prefixes were retained.
+    assert store._files == {}
+    assert store._dir_counts == {}
+    # Reads of the same path hit the same validation, never the index.
+    with pytest.raises(OSError) as read_info:
+        await store.read(path)
+    assert read_info.value.errno == errno.ENAMETOOLONG
+
+    # A total length past the bound is refused the same way even when shallow.
+    with pytest.raises(OSError) as long_info:
+        await store.write("/".join(["b" * 3000, "c" * 3000]), "x")
+    assert long_info.value.errno == errno.ENAMETOOLONG
+    assert store._files == {}
+
+
+async def test_in_memory_store_accepts_deep_paths_within_the_bound() -> None:
+    """Legitimate deep paths (deeper than any realistic hierarchy) still work,
+    and the retention of the directory index stays linear in the stored path."""
+    store = InMemoryAgentFileStore()
+    path = "/".join(["d"] * 200 + ["f"])
+
+    await store.write(path, "x")
+
+    assert await store.read(path) == "x"
+    # Every retained ancestor key is a prefix of the one stored path, so the
+    # index is bounded by a small constant times the path length.
+    retained = sum(len(k) for k in store._dir_counts)
+    assert retained < 200 * len(path)
+
+
+async def test_filesystem_store_matches_in_memory_complexity_bound(tmp_path) -> None:
+    """Both backends refuse the same over-bound path with the same errno, so
+    backend substitution is not observable on the bound either."""
+    import errno
+
+    fs_store = FileSystemAgentFileStore(str(tmp_path))
+    mem_store = InMemoryAgentFileStore()
+    path = "/".join(["a"] * 8000 + ["f"])
+
+    with pytest.raises(OSError) as fs_info:
+        await fs_store.write(path, "x")
+    with pytest.raises(OSError) as mem_info:
+        await mem_store.write(path, "x")
+
+    assert fs_info.value.errno == errno.ENAMETOOLONG
+    assert mem_info.value.errno == errno.ENAMETOOLONG
+
+    within = "/".join(["d"] * 200 + ["f"])
+    await fs_store.write(within, "x")
+    await mem_store.write(within, "x")
+    assert await fs_store.read(within) == "x"
+    assert await mem_store.read(within) == "x"