From d4302c01e5226c61a8fe0ff463042740001e5e7f Mon Sep 17 00:00:00 2001 From: ww2283 Date: Wed, 19 Aug 2026 18:46:48 -0400 Subject: [PATCH] fix(cli): copy chunk metadata before id assignment to break dict aliasing (#7) --- packages/leann-core/src/leann/cli.py | 4 ++-- tests/test_incremental_build.py | 34 ++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 2 deletions(-) diff --git a/packages/leann-core/src/leann/cli.py b/packages/leann-core/src/leann/cli.py index f37e1ca9..58e81faf 100644 --- a/packages/leann-core/src/leann/cli.py +++ b/packages/leann-core/src/leann/cli.py @@ -2081,7 +2081,7 @@ def _assign_chunk_ids(chunks: list[dict]) -> None: for path_key, path_chunks in by_path.items(): for idx, c in enumerate(path_chunks): sid = hashlib.sha256(f"{path_key}:{idx}".encode()).hexdigest()[:16] - c.setdefault("metadata", {})["id"] = sid + c["metadata"] = {**c.get("metadata", {}), "id": sid} c["id"] = sid @staticmethod @@ -2089,7 +2089,7 @@ def _assign_unique_chunk_ids(chunks: list[dict]) -> None: """Assign unique IDs for incremental (avoids collision when path lookup misses some old ids).""" for c in chunks: sid = uuid.uuid4().hex[:16] - c.setdefault("metadata", {})["id"] = sid + c["metadata"] = {**c.get("metadata", {}), "id": sid} c["id"] = sid def _chunks_for_paths(self, all_texts: list[dict], paths: set[str]) -> list[dict]: diff --git a/tests/test_incremental_build.py b/tests/test_incremental_build.py index 63ec0ea2..e9cb366a 100644 --- a/tests/test_incremental_build.py +++ b/tests/test_incremental_build.py @@ -371,3 +371,37 @@ def test_ivf_multiple_incremental_no_duplicates(tmp_path): assert len(stale_ids) == 0, ( f"passages.jsonl has {len(stale_ids)} stale entries not in offset_map: {stale_ids[:5]}" ) + + +def test_assign_unique_chunk_ids_with_shared_metadata_dict(): + from leann.cli import LeannCLI + + # Arrange + shared = {"file_path": "a.md"} + chunks = [{"text": t, "metadata": shared} for t in ("one", "two", "three")] + + # Act + LeannCLI._assign_unique_chunk_ids(chunks) + + # Assert + metadata_ids = [c["metadata"]["id"] for c in chunks] + assert len(set(metadata_ids)) == len(chunks) + for c in chunks: + assert c["metadata"]["id"] == c["id"] + + +def test_assign_chunk_ids_with_shared_metadata_dict(): + from leann.cli import LeannCLI + + # Arrange + shared = {"file_path": "a.md"} + chunks = [{"text": t, "metadata": shared} for t in ("one", "two", "three")] + + # Act + LeannCLI._assign_chunk_ids(chunks) + + # Assert + metadata_ids = [c["metadata"]["id"] for c in chunks] + assert len(set(metadata_ids)) == len(chunks) + for c in chunks: + assert c["metadata"]["id"] == c["id"]