Python: fix(core): give each file memory its own description sidecar

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

@@ -95,15 +95,43 @@
 def _description_file_name(file_name: str) -> str:
     """Return the companion description file name for ``file_name``.
 
-    The suffix replaces the original extension when present (so ``notes.md``
-    becomes ``notes_description.md``); otherwise it is appended.
+    The suffix is appended to the full file name, extension included, so
+    ``notes.md`` becomes ``notes.md_description.md``. Keeping the extension
+    gives every memory file its own sidecar: ``notes.md`` and ``notes.json``
+    must not share one, or writing or deleting either would clobber the
+    other's description.
+    """
+    return f"{file_name}{_DESCRIPTION_SUFFIX}"
+
+
+def _legacy_description_file_name(file_name: str) -> str:
+    """Return the sidecar name used before the extension was kept.
+
+    Earlier versions replaced the extension (``notes.md`` ->
+    ``notes_description.md``). Sidecars written under that scheme are still
+    read so existing memories keep their descriptions; new sidecars always use
+    :func:`_description_file_name`.
     """
     dot_index = file_name.rfind(".")
     if dot_index > 0:
         return f"{file_name[:dot_index]}{_DESCRIPTION_SUFFIX}"
     return f"{file_name}{_DESCRIPTION_SUFFIX}"
 
 
+def _legacy_sidecar_is_shared(file_name: str, file_names: list[str]) -> bool:
+    """Return whether another memory file maps to the same legacy sidecar as ``file_name``.
+
+    Under the legacy scheme ``notes.md`` and ``notes.json`` both map to
+    ``notes_description.md``, so that sidecar may only be removed on behalf of
+    ``file_name`` when no other file still claims it.
+    """
+    legacy_name = _legacy_description_file_name(file_name)
+    return any(
+        other != file_name and not _is_internal_file(other) and _legacy_description_file_name(other) == legacy_name
+        for other in file_names
+    )
+
+
 def _is_internal_file(file_name: str) -> bool:
     """Return whether ``file_name`` is an internal file hidden from the agent.
 
@@ -338,8 +366,9 @@ async def _rebuild_index(self, working_folder: str) -> None:
         sorted_files = sorted((name for name in file_names if not _is_internal_file(name)), key=str.lower)
 
         lines = ["# Memory Index", ""]
+        available = set(file_names)
         for file_name in sorted_files[:_MAX_INDEX_ENTRIES]:
-            description = await self.store.read(_combine_paths(working_folder, _description_file_name(file_name)))
+            description = await self._read_description(working_folder, file_name, available)
             if description and description.strip():
                 lines.append(f"- **{file_name}**: {description.strip()}")
             else:
@@ -348,6 +377,30 @@ async def _rebuild_index(self, working_folder: str) -> None:
         index_path = _combine_paths(working_folder, _MEMORY_INDEX_FILE_NAME)
         await self.store.write(index_path, "\n".join(lines) + "\n")
 
+    async def _read_description(self, working_folder: str, file_name: str, available: set[str]) -> str | None:
+        """Return the description stored for ``file_name``, or ``None`` when it has none.
+
+        The current sidecar name is preferred; a sidecar written under the
+        legacy extension-replacing scheme is used when no current one exists.
+        """
+        for sidecar in (_description_file_name(file_name), _legacy_description_file_name(file_name)):
+            if sidecar in available:
+                return await self.store.read(_combine_paths(working_folder, sidecar))
+        return None
+
+    async def _delete_description(self, working_folder: str, file_name: str) -> None:
+        """Remove the description sidecar(s) of ``file_name``.
+
+        The legacy sidecar is only removed when no other memory file still maps
+        to it, since the legacy naming is shared between files that differ only
+        by extension.
+        """
+        await self.store.delete(_combine_paths(working_folder, _description_file_name(file_name)))
+        entries = await self.store.list_children(working_folder)
+        file_names = [entry.name for entry in entries if entry.type == FileStoreEntry.FILE]
+        if not _legacy_sidecar_is_shared(file_name, file_names):
+            await self.store.delete(_combine_paths(working_folder, _legacy_description_file_name(file_name)))
+
     async def before_run(
         self,
         *,
@@ -388,7 +441,7 @@ async def file_memory_write(file_name: str, content: str, description: str | Non
                     if description and description.strip():
                         await self.store.write(desc_path, description)
                     else:
-                        await self.store.delete(desc_path)
+                        await self._delete_description(working_folder, normalized)
                     await self._rebuild_index(working_folder)
                 except ValueError as exc:
                     return f"Could not write file '{file_name}': {exc}"
@@ -426,12 +479,12 @@ async def file_memory_delete(file_name: str) -> str:
                 return f"File '{file_name}' not found."
 
             path = _combine_paths(working_folder, normalized)
-            desc_path = _combine_paths(working_folder, _description_file_name(normalized))
             async with self._write_lock:
                 try:
                     deleted = await self.store.delete(path)
-                    await self.store.delete(desc_path)
-                    await self._rebuild_index(working_folder)
+                    if deleted:
+                        await self._delete_description(working_folder, normalized)
+                        await self._rebuild_index(working_folder)
                 except ValueError as exc:
                     return f"Could not delete file '{file_name}': {exc}"
                 except OSError as exc:
@@ -454,10 +507,7 @@ async def file_memory_ls(glob_pattern: str | None = None) -> list[dict[str, Any]
                     continue
                 if not _matches_glob(file_name, glob_pattern):
                     continue
-                description: str | None = None
-                desc_file_name = _description_file_name(file_name)
-                if desc_file_name in available:
-                    description = await self.store.read(_combine_paths(working_folder, desc_file_name))
+                description = await self._read_description(working_folder, file_name, available)
                 results.append({"name": file_name, "type": "file", "description": description})
             return results
 

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

@@ -23,6 +23,8 @@
     _combine_paths,
     _description_file_name,
     _is_internal_file,
+    _legacy_description_file_name,
+    _legacy_sidecar_is_shared,
 )
 from agent_framework._sessions import SessionContext
 
@@ -53,13 +55,31 @@ async def _prepare(
     return context, tools
 
 
-def test_description_file_name_replaces_extension() -> None:
-    """The description sidecar replaces a known extension and appends otherwise."""
-    assert _description_file_name("notes.md") == "notes_description.md"
-    assert _description_file_name("data.json") == "data_description.md"
+def test_description_file_name_keeps_extension() -> None:
+    """The description sidecar appends the suffix to the whole name, extension included."""
+    assert _description_file_name("notes.md") == "notes.md_description.md"
+    assert _description_file_name("data.json") == "data.json_description.md"
     assert _description_file_name("noext") == "noext_description.md"
-    # Leading-dot files have no stem, so the suffix is appended.
     assert _description_file_name(".hidden") == ".hidden_description.md"
+    # Files that differ only by extension must not share a sidecar.
+    assert _description_file_name("notes.md") != _description_file_name("notes.json")
+
+
+def test_legacy_description_file_name_replaces_extension() -> None:
+    """The legacy scheme replaced a known extension and appended otherwise."""
+    assert _legacy_description_file_name("notes.md") == "notes_description.md"
+    assert _legacy_description_file_name("data.json") == "data_description.md"
+    assert _legacy_description_file_name("noext") == "noext_description.md"
+    assert _legacy_description_file_name(".hidden") == ".hidden_description.md"
+
+
+def test_legacy_sidecar_is_shared_only_with_other_visible_files() -> None:
+    """A legacy sidecar is shared when another non-internal file maps to it."""
+    assert _legacy_sidecar_is_shared("notes.md", ["notes.md", "notes.json"])
+    assert not _legacy_sidecar_is_shared("notes.md", ["notes.md", "plan.md"])
+    assert not _legacy_sidecar_is_shared("notes.md", ["notes.md"])
+    # The sidecar itself and the index are internal and never count as owners.
+    assert not _legacy_sidecar_is_shared("notes.md", ["notes.md", "notes_description.md", "memories.md"])
 
 
 def test_is_internal_file_detects_sidecars_and_index() -> None:
@@ -150,15 +170,15 @@ async def test_description_sidecar_is_written_and_listed() -> None:
     )
     assert "with description" in _text(result)
 
-    sidecar = await store.read(_combine_paths("user-1", "arch_description.md"))
+    sidecar = await store.read(_combine_paths("user-1", "arch.md_description.md"))
     assert sidecar == "system architecture"
 
     listed = json.loads(_text(await list_files.invoke()))
     assert listed == [{"name": "arch.md", "type": "file", "description": "system architecture"}]
 
     # Re-saving without a description removes the sidecar.
     await save.invoke(arguments={"file_name": "arch.md", "content": "big content"})
-    assert await store.read(_combine_paths("user-1", "arch_description.md")) is None
+    assert await store.read(_combine_paths("user-1", "arch.md_description.md")) is None
     listed_again = json.loads(_text(await list_files.invoke()))
     assert listed_again == [{"name": "arch.md", "type": "file", "description": None}]
 
@@ -170,10 +190,75 @@ async def test_delete_removes_sidecar() -> None:
     _, tools = await _prepare(provider)
 
     await tools["file_memory_write"].invoke(arguments={"file_name": "arch.md", "content": "x", "description": "desc"})
-    assert await store.read(_combine_paths("user-1", "arch_description.md")) == "desc"
+    assert await store.read(_combine_paths("user-1", "arch.md_description.md")) == "desc"
 
     await tools["file_memory_delete"].invoke(arguments={"file_name": "arch.md"})
-    assert await store.read(_combine_paths("user-1", "arch_description.md")) is None
+    assert await store.read(_combine_paths("user-1", "arch.md_description.md")) is None
+
+
+async def test_files_differing_only_by_extension_keep_separate_descriptions() -> None:
+    """``notes.md`` and ``notes.json`` each own their description; neither write nor delete crosses over."""
+    store = InMemoryAgentFileStore()
+    provider = FileMemoryProvider(store=store, scope="user-1")
+    _, tools = await _prepare(provider)
+    save = tools["file_memory_write"]
+    delete = tools["file_memory_delete"]
+    list_files = tools["file_memory_ls"]
+
+    await save.invoke(arguments={"file_name": "notes.md", "content": "md", "description": "markdown notes"})
+
+    # Writing a sibling without a description must not drop the markdown file's description ...
+    await save.invoke(arguments={"file_name": "notes.json", "content": "{}"})
+    listed = {entry["name"]: entry["description"] for entry in json.loads(_text(await list_files.invoke()))}
+    assert listed == {"notes.md": "markdown notes", "notes.json": None}
+
+    # ... nor may the sibling's own description leak onto the markdown file.
+    await save.invoke(arguments={"file_name": "notes.json", "content": "{}", "description": "json notes"})
+    listed = {entry["name"]: entry["description"] for entry in json.loads(_text(await list_files.invoke()))}
+    assert listed == {"notes.md": "markdown notes", "notes.json": "json notes"}
+    index = await store.read(_combine_paths("user-1", _MEMORY_INDEX_FILE_NAME))
+    assert index is not None
+    assert "- **notes.json**: json notes" in index
+    assert "- **notes.md**: markdown notes" in index
+
+    # Deleting one sibling (or a missing one) leaves the other's description alone.
+    assert "not found" in _text(await delete.invoke(arguments={"file_name": "notes.txt"}))
+    assert "deleted" in _text(await delete.invoke(arguments={"file_name": "notes.json"}))
+    listed = json.loads(_text(await list_files.invoke()))
+    assert listed == [{"name": "notes.md", "type": "file", "description": "markdown notes"}]
+    assert await store.read(_combine_paths("user-1", "notes.json_description.md")) is None
+
+
+async def test_legacy_description_sidecar_is_still_read_and_cleaned_up() -> None:
+    """Sidecars written under the old extension-replacing name keep working."""
+    store = InMemoryAgentFileStore()
+    await store.write("user-1/notes.md", "md")
+    await store.write("user-1/notes_description.md", "legacy description")
+    provider = FileMemoryProvider(store=store, scope="user-1")
+    _, tools = await _prepare(provider)
+    save = tools["file_memory_write"]
+    delete = tools["file_memory_delete"]
+    list_files = tools["file_memory_ls"]
+
+    listed = json.loads(_text(await list_files.invoke()))
+    assert listed == [{"name": "notes.md", "type": "file", "description": "legacy description"}]
+
+    # A new description supersedes the legacy sidecar without touching it.
+    await save.invoke(arguments={"file_name": "notes.md", "content": "md", "description": "new description"})
+    listed = json.loads(_text(await list_files.invoke()))
+    assert listed == [{"name": "notes.md", "type": "file", "description": "new description"}]
+    assert await store.read("user-1/notes_description.md") == "legacy description"
+
+    # While a sibling still maps to the legacy sidecar, deleting one file keeps it for the other.
+    await save.invoke(arguments={"file_name": "notes.json", "content": "{}"})
+    await delete.invoke(arguments={"file_name": "notes.md"})
+    assert await store.read("user-1/notes_description.md") == "legacy description"
+    listed = json.loads(_text(await list_files.invoke()))
+    assert listed == [{"name": "notes.json", "type": "file", "description": "legacy description"}]
+
+    # Once the last owner is deleted, the legacy sidecar goes with it.
+    await delete.invoke(arguments={"file_name": "notes.json"})
+    assert await store.read("user-1/notes_description.md") is None
 
 
 async def test_index_is_rebuilt_and_injected_on_next_run() -> None: