From dddaf0f84cf3c94c5ad158666660111e52535942 Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Sun, 26 Jul 2026 18:45:48 +0300 Subject: [PATCH] Remove unreachable code and fix two defects it surfaced Delete repository methods with no callers (SettingsRepository CRUD, ChunkRepository.update/delete/get_chunks_in_range), the DataFrame branch of _process_search_results whose only caller always passes a query, and guards that cannot be reached from their call sites: the rebuild mode=None default, the staging drop already performed by _resolve_rebuild_recovery, the empty batch skip, two context fast paths, the doctor prefix guard, the poller _task attribute that is never assigned, and a docling caption fallback for a field name no item class defines. set_haiku_version built a recreated settings row from the process-global Config rather than the store's own, so a store opened with a custom config stamped global settings into the database. check_source_accessible called urlparse outside its try block, so a stored URI with a malformed IPv6 host raised ValueError instead of reporting the source as inaccessible, aborting the whole rebuild sweep. --- CHANGELOG.md | 9 ++ haiku_rag_slim/haiku/rag/client/documents.py | 10 ++- haiku_rag_slim/haiku/rag/client/rebuild.py | 9 +- haiku_rag_slim/haiku/rag/context.py | 4 - haiku_rag_slim/haiku/rag/doctor.py | 2 - .../haiku/rag/ingester/pollers/base.py | 4 - haiku_rag_slim/haiku/rag/store/engine.py | 2 +- .../haiku/rag/store/models/document_item.py | 4 - .../haiku/rag/store/repositories/chunk.py | 82 +------------------ .../haiku/rag/store/repositories/settings.py | 41 ---------- tests/test_client.py | 25 ++++++ tests/test_settings.py | 25 ++++++ 12 files changed, 71 insertions(+), 146 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bba53050..cc59c816 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,15 @@ # Changelog ## [Unreleased] +### Fixed + +- `Store.set_haiku_version` stamps the store's own config into a recreated settings row instead of the process-global `Config`. +- `check_source_accessible` returns `False` for a URI that fails to parse instead of raising `ValueError`. + +### Removed + +- `SettingsRepository.create`, `get_by_id`, `update`, `delete`, `list_all` and `ChunkRepository.update`, `delete`, `get_chunks_in_range`. + ## [0.70.0] - 2026-07-25 ### Added diff --git a/haiku_rag_slim/haiku/rag/client/documents.py b/haiku_rag_slim/haiku/rag/client/documents.py index 41e25599..6dca344e 100644 --- a/haiku_rag_slim/haiku/rag/client/documents.py +++ b/haiku_rag_slim/haiku/rag/client/documents.py @@ -906,13 +906,17 @@ async def update_document( def check_source_accessible(uri: str) -> bool: - """Check if a document's source URI is accessible.""" - parsed_url = urlparse(uri) + """Check if a document's source URI is accessible. + + A stored URI that no longer parses (``urlparse`` rejects malformed IPv6 + hosts) counts as inaccessible rather than aborting the caller's sweep. + """ try: + parsed_url = urlparse(uri) if parsed_url.scheme == "file": return Path(parsed_url.path).exists() elif parsed_url.scheme in ("http", "https", "s3"): return True return False - except Exception: + except ValueError: return False diff --git a/haiku_rag_slim/haiku/rag/client/rebuild.py b/haiku_rag_slim/haiku/rag/client/rebuild.py index fb7f5b5a..0ff37cb5 100644 --- a/haiku_rag_slim/haiku/rag/client/rebuild.py +++ b/haiku_rag_slim/haiku/rag/client/rebuild.py @@ -66,7 +66,7 @@ class _StagingMarkerRecord(LanceModel): async def rebuild_database( - client: "HaikuRAG", mode: "RebuildMode | None" = None + client: "HaikuRAG", mode: "RebuildMode" ) -> AsyncGenerator[str, None]: """Rebuild the database with the specified mode. @@ -79,9 +79,6 @@ async def rebuild_database( """ from haiku.rag.client import RebuildMode - if mode is None: - mode = RebuildMode.FULL - async with client.store._rebuild_lock: if mode == RebuildMode.SET_EMBEDDER: await _set_embedder(client) @@ -287,8 +284,6 @@ async def _populate_staging_table(client: "HaikuRAG") -> None: """ db = client.store.db tables = (await db.list_tables()).tables - if _STAGING_TABLE_NAME in tables: - await db.drop_table(_STAGING_TABLE_NAME) staging = await db.create_table(_STAGING_TABLE_NAME, schema=_StagingChunkRecord) if "chunks" not in tables: @@ -301,8 +296,6 @@ async def _populate_staging_table(client: "HaikuRAG") -> None: ) async for batch in stream: rows = batch.to_pylist() - if not rows: - continue records = [ _StagingChunkRecord( id=r["id"], diff --git a/haiku_rag_slim/haiku/rag/context.py b/haiku_rag_slim/haiku/rag/context.py index 79ccb516..51e8375d 100644 --- a/haiku_rag_slim/haiku/rag/context.py +++ b/haiku_rag_slim/haiku/rag/context.py @@ -101,8 +101,6 @@ def _evidence_anchors(content: str, max_chars: int) -> list[str]: mid = len(content) // 2 start = max(0, mid - target // 2) anchors.append(content[start : start + target]) - if not anchors: - anchors.append(content[:max_chars] if max_chars > 0 else content) return anchors @@ -289,8 +287,6 @@ def _add_input_pages_for_surviving_refs( ) -> None: """Fill missing item-table pages from inputs whose own refs all survived.""" surviving = set(refs) - if not surviving: - return for result in original_results: if not result.page_numbers or not result.doc_item_refs: continue diff --git a/haiku_rag_slim/haiku/rag/doctor.py b/haiku_rag_slim/haiku/rag/doctor.py index f6bd51fd..c885dc68 100644 --- a/haiku_rag_slim/haiku/rag/doctor.py +++ b/haiku_rag_slim/haiku/rag/doctor.py @@ -330,8 +330,6 @@ def _common_path_prefix(labels: list[str]) -> str: Returns "" unless the shared prefix is long enough to be worth factoring out of every line (deep URI trees are otherwise unreadable). """ - if len(labels) < 2: - return "" lo, hi = min(labels), max(labels) end = 0 while end < len(lo) and lo[end] == hi[end]: diff --git a/haiku_rag_slim/haiku/rag/ingester/pollers/base.py b/haiku_rag_slim/haiku/rag/ingester/pollers/base.py index e6700b8f..26e992fb 100644 --- a/haiku_rag_slim/haiku/rag/ingester/pollers/base.py +++ b/haiku_rag_slim/haiku/rag/ingester/pollers/base.py @@ -55,7 +55,6 @@ class BasePoller: self._sync = sync_repo self._breaker = breaker or CircuitBreaker(config.circuit_breaker) self._stop = asyncio.Event() - self._task: asyncio.Task | None = None self._last_polled_at: datetime | None = None self._last_skip_reason: str | None = None self._default_max_attempts = default_max_attempts @@ -84,9 +83,6 @@ class BasePoller: async def stop(self) -> None: self._stop.set() - if self._task is not None: - await asyncio.gather(self._task, return_exceptions=True) - self._task = None async def _stagger_start(self) -> bool: """Sleep a random fraction of the interval so pollers sharing an diff --git a/haiku_rag_slim/haiku/rag/store/engine.py b/haiku_rag_slim/haiku/rag/store/engine.py index 1c5cd926..cc177ad8 100644 --- a/haiku_rag_slim/haiku/rag/store/engine.py +++ b/haiku_rag_slim/haiku/rag/store/engine.py @@ -847,7 +847,7 @@ class Store: ) else: # Create new settings record - settings_data = Config.model_dump(mode="json") + settings_data = self._config.model_dump(mode="json") settings_data["version"] = version await self.settings_table.add( [SettingsRecord(id="settings", settings=json.dumps(settings_data))] diff --git a/haiku_rag_slim/haiku/rag/store/models/document_item.py b/haiku_rag_slim/haiku/rag/store/models/document_item.py index e05506cc..f86ec44b 100644 --- a/haiku_rag_slim/haiku/rag/store/models/document_item.py +++ b/haiku_rag_slim/haiku/rag/store/models/document_item.py @@ -105,10 +105,6 @@ def extract_item_text( except Exception: pass - if caption := getattr(item, "caption", None): - if hasattr(caption, "text"): - return caption.text - return None diff --git a/haiku_rag_slim/haiku/rag/store/repositories/chunk.py b/haiku_rag_slim/haiku/rag/store/repositories/chunk.py index 1960b5c9..09cc3498 100644 --- a/haiku_rag_slim/haiku/rag/store/repositories/chunk.py +++ b/haiku_rag_slim/haiku/rag/store/repositories/chunk.py @@ -4,7 +4,6 @@ from typing import TYPE_CHECKING from uuid import uuid4 if TYPE_CHECKING: - import pandas as pd from lancedb.query import AsyncQueryBase from lancedb.index import FTS @@ -153,40 +152,6 @@ class ChunkRepository: order=chunk_record.order, ) - async def update(self, entity: Chunk) -> Chunk: - """Update an existing chunk. - - Chunk must have embedding set before calling this method. - """ - self.store._assert_writable() - assert entity.id, "Chunk ID is required for update" - assert entity.embedding is not None, "Chunk must have an embedding" - - await self.store.chunks_table.update( - { - "document_id": entity.document_id, - "content": entity.content, - "content_fts": self._contextualize_content(entity), - "metadata": json.dumps( - {k: v for k, v in entity.metadata.items() if k != "order"} - ), - "order": int(entity.order), - "vector": entity.embedding, - }, - where=f"id = '{entity.id}'", - ) - return entity - - async def delete(self, entity_id: str) -> bool: - """Delete a chunk by its ID.""" - self.store._assert_writable() - chunk = await self.get_by_id(entity_id) - if chunk is None: - return False - - await self.store.chunks_table.delete(f"id = '{entity_id}'") - return True - async def list_all( self, limit: int | None = None, offset: int | None = None ) -> list[Chunk]: @@ -417,46 +382,10 @@ class ChunkRepository: ) return len(df) - async def get_chunks_in_range( - self, document_id: str, min_order: int, max_order: int - ) -> list[Chunk]: - """Get chunks for a document within an order range. - - Args: - document_id: The document ID to get chunks for. - min_order: Minimum order value (inclusive). - max_order: Maximum order value (inclusive). - - Returns: - List of chunks within the order range. - """ - where = ( - f"document_id = '{document_id}'" - f" AND `order` >= {min_order}" - f" AND `order` <= {max_order}" - ) - results = await query_to_pydantic( - self.store.chunks_table.query().where(where), self.store.ChunkRecord - ) - return [ - Chunk( - id=rec.id, - document_id=rec.document_id, - content=rec.content, - metadata=json.loads(rec.metadata), - order=rec.order, - ) - for rec in results - ] - async def _process_search_results( - self, query_result: "pd.DataFrame | AsyncQueryBase" + self, query_result: "AsyncQueryBase" ) -> list[tuple[Chunk, float]]: - """Process search results into chunks with document info and scores. - - Args: - query_result: Either a pandas DataFrame or a LanceDB async query result - """ + """Process search results into chunks with document info and scores.""" import pandas as pd def extract_scores(df: pd.DataFrame) -> list[float]: @@ -473,12 +402,7 @@ class ChunkRepository: else: raise ValueError("Unknown search result format, cannot extract scores") - # Convert everything to DataFrame for uniform processing - if isinstance(query_result, pd.DataFrame): - df = query_result - else: - # Convert LanceDB query result to DataFrame - df = await query_result.to_pandas() + df = await query_result.to_pandas() # Extract scores scores = extract_scores(df) diff --git a/haiku_rag_slim/haiku/rag/store/repositories/settings.py b/haiku_rag_slim/haiku/rag/store/repositories/settings.py index 5d7fd2e1..d7b2986c 100644 --- a/haiku_rag_slim/haiku/rag/store/repositories/settings.py +++ b/haiku_rag_slim/haiku/rag/store/repositories/settings.py @@ -18,47 +18,6 @@ class SettingsRepository: def __init__(self, store: Store) -> None: self.store = store - async def create(self, entity: dict) -> dict: - """Create settings in the database.""" - settings_record = SettingsRecord(id="settings", settings=json.dumps(entity)) - await self.store.settings_table.add([settings_record]) - return entity - - async def get_by_id(self, entity_id: str) -> dict | None: - """Get settings by ID.""" - results = await query_to_pydantic( - self.store.settings_table.query().where(f"id = '{entity_id}'").limit(1), - SettingsRecord, - ) - - if not results: - return None - - return json.loads(results[0].settings) if results[0].settings else {} - - async def update(self, entity: dict) -> dict: - """Update existing settings.""" - await self.store.settings_table.update( - {"settings": json.dumps(entity)}, where="id = 'settings'" - ) - return entity - - async def delete(self, entity_id: str) -> bool: - """Delete settings by ID.""" - await self.store.settings_table.delete(f"id = '{entity_id}'") - return True - - async def list_all( - self, limit: int | None = None, offset: int | None = None - ) -> list[dict]: - """List all settings.""" - results = await query_to_pydantic( - self.store.settings_table.query(), SettingsRecord - ) - return [ - json.loads(record.settings) if record.settings else {} for record in results - ] - async def get_current_settings(self) -> dict: """Get the current settings.""" results = await query_to_pydantic( diff --git a/tests/test_client.py b/tests/test_client.py index 19b6fa6c..73012dc5 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -14,6 +14,7 @@ from haiku.rag.client.documents import ( DocumentImport, _prepare_document_from_docling, _write_fetch_body, + check_source_accessible, ) from haiku.rag.config import Config from haiku.rag.store.compression import decompress_json @@ -2254,3 +2255,27 @@ async def test_metadata_only_update_waits_for_write_lock(temp_db_path): assert not task.done() updated = await task assert updated.metadata == {"k": "v"} + + +@pytest.mark.parametrize( + "uri,expected", + [ + ("https://example.com/doc.pdf", True), + ("s3://bucket/key", True), + ("mem://not-a-source", False), + # urlparse rejects a malformed IPv6 host; a stored URI that no longer + # parses must not abort the caller's rebuild sweep. + ("http://[::1", False), + ], + ids=["https", "s3", "unknown_scheme", "unparseable"], +) +def test_check_source_accessible(uri, expected): + assert check_source_accessible(uri) is expected + + +def test_check_source_accessible_file_uri(tmp_path): + existing = tmp_path / "there.txt" + existing.write_text("x") + + assert check_source_accessible(existing.as_uri()) is True + assert check_source_accessible((tmp_path / "gone.txt").as_uri()) is False diff --git a/tests/test_settings.py b/tests/test_settings.py index b4e1f478..b2f11626 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -44,6 +44,31 @@ async def test_settings_save_and_retrieve(temp_db_path): Config.processing.chunk_size = original_chunk_size +@pytest.mark.asyncio +async def test_set_haiku_version_recreates_row_from_store_config(temp_db_path): + """Recreating a missing settings row stamps the store's own config, not the + process-global one.""" + from haiku.rag.store.engine import Store + from haiku.rag.store.repositories.settings import SettingsRepository + + config = AppConfig() + config.processing.chunk_size = Config.processing.chunk_size + 512 + + async with Store(temp_db_path, config=config, create=True) as store: + settings_repo = SettingsRepository(store) + + await store.settings_table.delete("id = 'settings'") + assert await settings_repo.get_current_settings() == {} + assert await store.get_haiku_version() == "0.0.0" + + await store.set_haiku_version("1.2.3") + + recreated = await settings_repo.get_current_settings() + assert recreated["version"] == "1.2.3" + assert recreated["processing"]["chunk_size"] == config.processing.chunk_size + assert await store.get_haiku_version() == "1.2.3" + + class TestValidateConfigCompatibility: """Tests for validate_config_compatibility method."""