fix(kb): single-sentence chunks skip subclass metadata (#14467)

autobot-backend/utils/semantic_chunker_base.py

@@ -256,16 +256,25 @@ def _create_chunks_with_boundaries(
 
     # ---- Single-sentence + metadata helpers ---------------------------
 
-    def _create_single_sentence_chunk(
-        self, text: str, sentences: List[str], metadata: Dict[str, Any] | None
-    ) -> SemanticChunk:
+    def _create_single_sentence_chunk(self, text: str, sentences: List[str]) -> SemanticChunk:
+        """Zero/one-sentence fallback chunk (Issue #14467).
+
+        Carries only the chunk-shape marker; subclass and pipeline metadata
+        (`optimization_version`, `chunk_index`, ...) is applied afterwards by
+        `_enrich_chunks_with_metadata()` in `chunk_text()` — the same call the
+        multi-sentence path uses — so both paths agree on one enrichment point.
+        """
+        chunk_metadata: Dict[str, Any] = {"single_sentence": True}
+        if not sentences:
+            chunk_metadata["empty_input"] = True
+
         return SemanticChunk(
             content=text,
             start_index=0,
             end_index=1,
             sentences=sentences,
             semantic_score=1.0,
-            metadata=metadata or {"single_sentence": True},
+            metadata=chunk_metadata,
         )
 
     def _build_chunk_metadata(
@@ -348,18 +357,21 @@ async def chunk_text(self, text: str, metadata: Dict[str, Any] | None = None) ->
 
             sentences = self._split_into_sentences(text)
             if len(sentences) <= 1:
-                return [self._create_single_sentence_chunk(text, sentences, metadata)]
+                chunks = [self._create_single_sentence_chunk(text, sentences)]
+            else:
+                logger.debug("Split text into %d sentences", len(sentences))
 
-            logger.debug("Split text into %d sentences", len(sentences))
+                await self._initialize_model()
+                embeddings = await self._compute_embeddings(sentences)
+                distances = self._compute_semantic_distances(embeddings)
+                boundaries = self._find_chunk_boundaries(distances)
 
-            await self._initialize_model()
-            embeddings = await self._compute_embeddings(sentences)
-            distances = self._compute_semantic_distances(embeddings)
-            boundaries = self._find_chunk_boundaries(distances)
+                logger.debug("Found %d semantic boundaries", len(boundaries))
 
-            logger.debug("Found %d semantic boundaries", len(boundaries))
+                chunks = self._create_chunks_with_boundaries(sentences, boundaries, distances)
 
-            chunks = self._create_chunks_with_boundaries(sentences, boundaries, distances)
+            # Issue #14467: a single trailing call so neither branch above — nor a
+            # future one — can reintroduce the defect by forgetting to enrich.
             self._enrich_chunks_with_metadata(chunks, metadata)
 
             if chunks:

autobot-backend/utils/semantic_chunker_gpu_test.py

@@ -36,6 +36,27 @@
 
 _TEXT = "First sentence here. Second sentence follows now. Third one too right now."
 
+# Issue #14467: input shapes that exercise `chunk_text()`'s zero/one-sentence
+# fallback (`_create_single_sentence_chunk`) instead of the boundary-detection
+# path. "Hi." is filtered to zero sentences by the >=3-word rule in
+# `_split_into_sentences` (#380); the other has exactly one qualifying sentence.
+_EMPTY_SENTENCE_TEXT = "Hi."
+_ONE_SENTENCE_TEXT = "This single sentence is deliberately long enough to survive filtering."
+
+# Keys `_build_chunk_metadata()`/`_extra_chunk_metadata()` add on every chunk,
+# regardless of which internal path produced it.
+_SUBCLASS_METADATA_KEYS = frozenset(
+    {
+        "chunk_index",
+        "total_chunks",
+        "source_metadata",
+        "embedding_model",
+        "chunking_method",
+        "gpu_batch_size",
+        "optimization_version",
+    }
+)
+
 
 @pytest.mark.asyncio
 class TestOptimizationMetadataReachesTheProductionSingleton:
@@ -74,3 +95,49 @@ async def test_a_fresh_instance_behaves_the_same_as_the_singleton(self):
 
         assert chunks
         assert all(c.metadata.get("optimization_version") == "rtx4070_gpu" for c in chunks)
+
+
+@pytest.mark.asyncio
+class TestShortInputChunksCarryTheSameMetadataAsLongInput:
+    """Issue #14467: `_create_single_sentence_chunk` used to return before
+    `_enrich_chunks_with_metadata()` ran, so `chunk_text()` on zero- or
+    one-sentence input silently dropped `optimization_version` and the rest
+    of `_build_chunk_metadata()`'s keys, while identical multi-sentence input
+    kept them. All three assertions below enter through the real, public
+    `chunk_text()` — never `_create_single_sentence_chunk` directly — so a fix
+    that only special-cases the private method without wiring it into the
+    public path would still fail here.
+
+    Pre-fix: the single- and empty-input cases fail (missing every key in
+    `_SUBCLASS_METADATA_KEYS`). Post-fix: all three pass.
+    """
+
+    async def _chunk_text(self, chunker: GPUSemanticChunker, text: str):
+        with (
+            patch.object(chunker, "_initialize_model", AsyncMock(return_value=None)),
+            patch.object(chunker, "_compute_embeddings", AsyncMock(return_value=np.zeros((3, 4)))),
+            patch.object(chunker, "_compute_semantic_distances", return_value=[0.1, 0.1]),
+        ):
+            return await chunker.chunk_text(text)
+
+    async def test_single_sentence_input_matches_multi_sentence_metadata_keys(self):
+        chunker = GPUSemanticChunker(gpu_batch_size=10, enable_gpu_memory_pool=False)
+
+        multi_chunks = await self._chunk_text(chunker, _TEXT)
+        single_chunks = await self._chunk_text(chunker, _ONE_SENTENCE_TEXT)
+
+        assert multi_chunks and single_chunks
+        for chunk in multi_chunks + single_chunks:
+            missing = _SUBCLASS_METADATA_KEYS - chunk.metadata.keys()
+            assert not missing, f"chunk missing keys {missing}: {sorted(chunk.metadata.keys())}"
+
+    async def test_empty_input_matches_multi_sentence_metadata_keys(self):
+        chunker = GPUSemanticChunker(gpu_batch_size=10, enable_gpu_memory_pool=False)
+
+        empty_chunks = await self._chunk_text(chunker, _EMPTY_SENTENCE_TEXT)
+
+        assert empty_chunks
+        for chunk in empty_chunks:
+            missing = _SUBCLASS_METADATA_KEYS - chunk.metadata.keys()
+            assert not missing, f"chunk missing keys {missing}: {sorted(chunk.metadata.keys())}"
+            assert chunk.metadata.get("empty_input") is True