From 4fb0c7d08d475fefbbab7db5d3ef869c3b721174 Mon Sep 17 00:00:00 2001 From: alt-glitch Date: Sat, 18 Apr 2026 15:54:54 +0530 Subject: [PATCH] refactor(workspace): remove indexer.py and search.py, use DefaultIndexer everywhere Delete the old workspace/indexer.py and workspace/search.py modules. All test imports now use workspace.default.DefaultIndexer directly. Backwards-compat re-exports removed from workspace/__init__.py. --- tests/workspace/test_indexer_pipeline.py | 37 +- tests/workspace/test_workspace_edge_cases.py | 25 +- workspace/__init__.py | 7 - workspace/indexer.py | 540 ------------------- workspace/search.py | 37 -- 5 files changed, 33 insertions(+), 613 deletions(-) delete mode 100644 workspace/indexer.py delete mode 100644 workspace/search.py diff --git a/tests/workspace/test_indexer_pipeline.py b/tests/workspace/test_indexer_pipeline.py index f4272b21f8..efb55d8102 100644 --- a/tests/workspace/test_indexer_pipeline.py +++ b/tests/workspace/test_indexer_pipeline.py @@ -20,8 +20,7 @@ import sys import textwrap from pathlib import Path -from workspace.indexer import index_workspace -from workspace.search import search_workspace +from workspace.default import DefaultIndexer from workspace.store import SQLiteFTS5Store @@ -45,7 +44,7 @@ def test_markdown_pipeline_emits_clean_metadata_per_modality( "More prose.\n", ) - summary = index_workspace(cfg) + summary = DefaultIndexer(cfg).index() assert summary.files_indexed == 1 assert summary.files_errored == 0 @@ -107,7 +106,7 @@ def test_small_markdown_file_is_split_into_modalities( "# Tiny\n\nShort intro.\n\n```python\nprint('hi')\n```\n", ) - summary = index_workspace(cfg) + summary = DefaultIndexer(cfg).index() assert summary.files_indexed == 1 with SQLiteFTS5Store(cfg.workspace_root) as store: @@ -149,7 +148,7 @@ def test_overlap_context_propagates_and_is_prefix_of_next_chunk( cfg.workspace_root / "notes" / "long.txt", "\n".join(sentences) + "\n" ) - summary = index_workspace(cfg) + summary = DefaultIndexer(cfg).index() assert summary.files_indexed == 1 with SQLiteFTS5Store(cfg.workspace_root) as store: @@ -201,7 +200,7 @@ def test_deprecated_strategy_and_threshold_keys_are_silently_ignored( # And indexing works end-to-end with the legacy-keyed config. write_file(cfg.workspace_root / "docs" / "readme.md", "# Hi\n\nSome prose.\n") - summary = index_workspace(cfg) + summary = DefaultIndexer(cfg).index() assert summary.files_indexed == 1 assert summary.files_errored == 0 @@ -215,18 +214,18 @@ def test_config_signature_change_invalidates_existing_index( cfg = make_workspace_config({"knowledgebase": {"chunking": {"chunk_size": 512}}}) write_file(cfg.workspace_root / "docs" / "a.md", "# A\n\nContent A.\n") - first = index_workspace(cfg) + first = DefaultIndexer(cfg).index() assert first.files_indexed == 1 assert first.files_skipped == 0 # Same config → second run skips. - second = index_workspace(cfg) + second = DefaultIndexer(cfg).index() assert second.files_indexed == 0 assert second.files_skipped == 1 # Changed chunk_size → third run re-indexes. cfg2 = make_workspace_config({"knowledgebase": {"chunking": {"chunk_size": 256}}}) - third = index_workspace(cfg2) + third = DefaultIndexer(cfg2).index() assert third.files_indexed == 1 assert third.files_skipped == 0 @@ -276,11 +275,11 @@ def test_concurrent_index_does_not_crash( sys.path.insert(0, {str(project_root)!r}) from workspace.config import WorkspaceConfig - from workspace.indexer import index_workspace + from workspace.default import DefaultIndexer hermes_home = Path({str(hermes_home)!r}) cfg = WorkspaceConfig(workspace_root=hermes_home / "workspace") - summary = index_workspace(cfg) + summary = DefaultIndexer(cfg).index() sys.exit(0) """ ), @@ -330,22 +329,20 @@ def test_search_path_prefix_resolves_symlinks( {"knowledgebase": {"roots": [{"path": str(linked), "recursive": True}]}}, ) - summary = index_workspace(cfg) + summary = DefaultIndexer(cfg).index() assert summary.files_indexed == 2 - # Via the symlink path — must still return hits, because search_workspace + # Via the symlink path — must still return hits, because search # resolves path_prefix before the byte-prefix compare. - via_symlink = search_workspace( + via_symlink = DefaultIndexer(cfg).search( "document", - cfg, path_prefix=str(linked), ) - assert len(via_symlink) > 0, "search_workspace must resolve symlinked path_prefix" + assert len(via_symlink) > 0, "search must resolve symlinked path_prefix" # Via the resolved real path — the counts must match. - via_resolved = search_workspace( + via_resolved = DefaultIndexer(cfg).search( "document", - cfg, path_prefix=str(real_docs.resolve()), ) assert len(via_symlink) == len(via_resolved) @@ -367,7 +364,7 @@ def test_hermesignore_never_indexed(make_workspace_config, write_file): # Plus a legitimate markdown file so the index has something in it. write_file(cfg.workspace_root / "docs" / "ok.md", "# Ok\n\nSome prose.\n") - summary = index_workspace(cfg) + summary = DefaultIndexer(cfg).index() assert summary.files_errored == 0 with SQLiteFTS5Store(cfg.workspace_root) as store: @@ -407,7 +404,7 @@ def test_summary_reports_filtered_empty_and_oversized( # One real file that should be indexed. write_file(cfg.workspace_root / "docs" / "real.md", "# Real\n\nActual content.\n") - summary = index_workspace(cfg) + summary = DefaultIndexer(cfg).index() assert summary.files_indexed == 1 assert summary.files_skipped == 3 diff --git a/tests/workspace/test_workspace_edge_cases.py b/tests/workspace/test_workspace_edge_cases.py index 33bbee3f66..27b0ee1cc6 100644 --- a/tests/workspace/test_workspace_edge_cases.py +++ b/tests/workspace/test_workspace_edge_cases.py @@ -10,7 +10,7 @@ from pathlib import Path import pytest from workspace.commands import workspace_command -from workspace.indexer import index_workspace +from workspace.default import DefaultIndexer from workspace.store import SQLiteFTS5Store @@ -63,7 +63,7 @@ def second_block(): """, ) - summary = index_workspace(cfg) + summary = DefaultIndexer(cfg).index() assert summary.files_indexed == 1 assert summary.files_errored == 0 @@ -107,7 +107,7 @@ def test_failed_reindex_keeps_previous_committed_rows( file_a = write_file(cfg.workspace_root / "docs" / "a.txt", "stable old content\n") file_b = write_file(cfg.workspace_root / "docs" / "b.txt", "other old content\n") - initial = index_workspace(cfg) + initial = DefaultIndexer(cfg).index() assert initial.files_indexed == 2 with SQLiteFTS5Store(cfg.workspace_root) as store: @@ -127,7 +127,7 @@ def test_failed_reindex_keeps_previous_committed_rows( file_a.write_text("stable new content that should roll back\n", encoding="utf-8") file_b.write_text("other new content that should succeed\n", encoding="utf-8") - summary = index_workspace(cfg) + summary = DefaultIndexer(cfg).index() assert summary.files_indexed == 1 assert summary.files_errored == 1 @@ -147,7 +147,9 @@ def test_failed_reindex_keeps_previous_committed_rows( assert "stable new content" not in combined -def test_missing_root_skips_stale_prune(tmp_path: Path, make_workspace_config, write_file): +def test_missing_root_skips_stale_prune( + tmp_path: Path, make_workspace_config, write_file +): external_root = tmp_path / "external" external_root.mkdir() @@ -159,15 +161,17 @@ def test_missing_root_skips_stale_prune(tmp_path: Path, make_workspace_config, w }, ) - local_file = write_file(cfg.workspace_root / "docs" / "local.txt", "local content\n") + local_file = write_file( + cfg.workspace_root / "docs" / "local.txt", "local content\n" + ) external_file = write_file(external_root / "external.txt", "external content\n") - first = index_workspace(cfg) + first = DefaultIndexer(cfg).index() assert first.files_indexed == 2 shutil.rmtree(external_root) - second = index_workspace(cfg) + second = DefaultIndexer(cfg).index() assert second.files_pruned == 0 with SQLiteFTS5Store(cfg.workspace_root) as store: @@ -231,7 +235,10 @@ def test_workspace_command_wraps_unexpected_errors_as_json( @pytest.mark.parametrize( ("argv", "expected"), [ - (["hermes", "workspace", "search", "needle", "--human"], ("search", None, True)), + ( + ["hermes", "workspace", "search", "needle", "--human"], + ("search", None, True), + ), (["hermes", "workspace", "roots", "list", "--human"], ("roots", "list", True)), ], ) diff --git a/workspace/__init__.py b/workspace/__init__.py index 0e54f23d15..5da6ccec0a 100644 --- a/workspace/__init__.py +++ b/workspace/__init__.py @@ -36,19 +36,12 @@ def get_indexer(config: WorkspaceConfig | None = None) -> BaseIndexer: return cls(config) -# Backwards-compat re-exports (indexer.py and search.py still exist for now) -from workspace.indexer import ensure_workspace_dirs, index_workspace # noqa: E402, F401 -from workspace.search import search_workspace # noqa: E402, F401 - __all__ = [ "BaseIndexer", "DefaultIndexer", "WorkspaceConfig", "load_workspace_config", "get_indexer", - "index_workspace", - "search_workspace", - "ensure_workspace_dirs", "IndexingError", "IndexSummary", "SearchResult", diff --git a/workspace/indexer.py b/workspace/indexer.py deleted file mode 100644 index 6d0848de2b..0000000000 --- a/workspace/indexer.py +++ /dev/null @@ -1,540 +0,0 @@ -"""Workspace indexing pipeline. - -Discovers files → checks content hash + config signature → dispatches to the -appropriate `chonkie.Pipeline` (markdown / code / plain) → iterates the -pipeline's modality-specific output into ChunkRecords → stores in SQLite FTS5. - -One Pipeline per file kind is built per `index_workspace` call. Chonkie caches -component instances keyed by init kwargs, so components are fully reused across -files of the same kind within a run. -""" - -import dataclasses -import hashlib -import json -import logging -import re -import time -import uuid -from datetime import datetime, timezone -from pathlib import Path -from typing import Callable, Literal - -try: - from chonkie import Pipeline - from chonkie.types import Document, MarkdownDocument -except ImportError as e: - raise ImportError( - "Chonkie is required for workspace indexing. " - "Install it with: pip install hermes-agent[workspace]" - ) from e - -from workspace.config import ChunkingConfig, WorkspaceConfig -from workspace.constants import ( - CHUNKING_PLAN_VERSION, - CODE_SUFFIXES, - MARKDOWN_SUFFIXES, - WORKSPACE_SUBDIRS, - get_index_dir, -) -from workspace.files import discover_workspace_files, seed_hermesignore -from workspace.store import SQLiteFTS5Store -from workspace.types import ChunkRecord, FileRecord, IndexingError, IndexSummary - -PipelineKind = Literal["markdown", "code", "plain"] - -log = logging.getLogger(__name__) - -_replace = dataclasses.replace - -ProgressCallback = Callable[[int, int, str], None] - -_HEADING_RE = re.compile(r"^(#{1,6})\s+(.+)$", re.MULTILINE) - -_MAX_ERRORS = 50 - - -def index_workspace( - config: WorkspaceConfig, - *, - progress: ProgressCallback | None = None, -) -> IndexSummary: - start = time.monotonic() - ensure_workspace_dirs(config) - config_sig = _config_signature(config) - - files_indexed = 0 - files_skipped = 0 - files_errored = 0 - chunks_created = 0 - errors: list[IndexingError] = [] - - discovery = discover_workspace_files(config) - files_skipped += discovery.filtered_count - all_files = discovery.files - total = len(all_files) - disk_paths: set[str] = set() - - pipelines = _build_pipelines(config.knowledgebase.chunking) - - with SQLiteFTS5Store(config.workspace_root) as store: - for i, (root_path, file_path) in enumerate(all_files): - abs_path = str(file_path.resolve()) - disk_paths.add(abs_path) - write_started = False - - if progress: - progress(i + 1, total, abs_path) - - try: - content_hash = _file_hash(file_path) - existing = store.get_file_record(abs_path) - if ( - existing - and existing.content_hash == content_hash - and existing.config_signature == config_sig - ): - files_skipped += 1 - continue - - text = _read_file_text(file_path) - if text is None: - files_errored += 1 - _append_error( - errors, - IndexingError( - path=abs_path, - stage="read", - error_type="EncodingError", - message="Could not decode file with sufficient confidence", - ), - ) - continue - - if not text.strip(): - files_skipped += 1 - continue - - suffix = file_path.suffix.lower() - chunk_records = _process_file(abs_path, text, suffix, pipelines) - - stat = file_path.stat() - record = FileRecord( - abs_path=abs_path, - root_path=root_path, - content_hash=content_hash, - config_signature=config_sig, - size_bytes=stat.st_size, - modified_at=datetime.fromtimestamp( - stat.st_mtime, tz=timezone.utc - ).isoformat(), - indexed_at=datetime.now(tz=timezone.utc).isoformat(), - chunk_count=len(chunk_records), - ) - - # Replace a file's rows atomically so a failed rebuild never - # destroys the previously indexed version of that file. - store.conn.execute("SAVEPOINT workspace_file_update") - write_started = True - store.delete_chunks_for_file(abs_path) - store.upsert_file(record) - if chunk_records: - store.insert_chunks(chunk_records) - store.conn.execute("RELEASE SAVEPOINT workspace_file_update") - store.commit() - write_started = False - - files_indexed += 1 - chunks_created += len(chunk_records) - - except Exception as exc: - if write_started: - try: - store.conn.execute( - "ROLLBACK TO SAVEPOINT workspace_file_update" - ) - store.conn.execute("RELEASE SAVEPOINT workspace_file_update") - except Exception: - log.warning( - "Failed to roll back workspace update for %s", - abs_path, - exc_info=True, - ) - files_errored += 1 - stage = "read" if isinstance(exc, FileNotFoundError) else "store" - _append_error( - errors, - IndexingError( - path=abs_path, - stage=stage, - error_type=type(exc).__name__, - message=str(exc), - ), - ) - log.warning("Failed to index %s: %s", abs_path, exc, exc_info=True) - continue - - if discovery.complete: - pruned = _prune_stale(store, disk_paths) - else: - pruned = 0 - log.warning( - "Workspace discovery was incomplete; skipping stale prune for this run" - ) - store.commit() - - elapsed = time.monotonic() - start - return IndexSummary( - files_indexed=files_indexed, - files_skipped=files_skipped, - files_pruned=pruned, - files_errored=files_errored, - chunks_created=chunks_created, - duration_seconds=elapsed, - errors=errors, - errors_truncated=files_errored > _MAX_ERRORS, - ) - - -def _append_error(errors: list[IndexingError], error: IndexingError) -> None: - if len(errors) < _MAX_ERRORS: - errors.append(error) - - -def _read_file_text(path: Path) -> str | None: - raw = path.read_bytes() - try: - return raw.decode("utf-8") - except UnicodeDecodeError: - pass - try: - from charset_normalizer import from_bytes - - result = from_bytes(raw).best() - if result is None or result.encoding is None: - return None - if result.coherence < 0.5: - return None - return str(result) - except ImportError: - log.debug("charset-normalizer not installed, skipping non-UTF8 file: %s", path) - return None - - -def ensure_workspace_dirs(config: WorkspaceConfig) -> None: - root = config.workspace_root - root.mkdir(parents=True, exist_ok=True) - for sub in WORKSPACE_SUBDIRS: - (root / sub).mkdir(exist_ok=True) - get_index_dir(root).mkdir(parents=True, exist_ok=True) - seed_hermesignore(root) - - -# --------------------------------------------------------------------------- -# Pipeline construction -# --------------------------------------------------------------------------- - - -def _build_pipelines(ch: ChunkingConfig) -> dict[PipelineKind, Pipeline]: - """Build one Pipeline per file kind, sharing overlap-refinery config. - - Chonkie's Pipeline caches component instances internally keyed by init - kwargs, so constructing a pipeline once per indexing run is enough to - get full reuse across files of the same kind. - """ - overlap_kwargs = dict( - tokenizer="word", - context_size=ch.overlap, - mode="token", - method="suffix", - merge=False, - ) - return { - "markdown": ( - Pipeline() - .process_with("markdown", tokenizer="word") - .chunk_with("recursive", tokenizer="word", chunk_size=ch.chunk_size) - .refine_with("overlap", **overlap_kwargs) - ), - "code": ( - Pipeline() - .chunk_with( - "code", - tokenizer="word", - chunk_size=ch.chunk_size, - language="auto", - ) - .refine_with("overlap", **overlap_kwargs) - ), - "plain": ( - Pipeline() - .chunk_with("recursive", tokenizer="word", chunk_size=ch.chunk_size) - .refine_with("overlap", **overlap_kwargs) - ), - } - - -# --------------------------------------------------------------------------- -# File processing pipeline -# --------------------------------------------------------------------------- - - -def _process_file( - abs_path: str, - text: str, - suffix: str, - pipelines: dict[PipelineKind, Pipeline], -) -> list[ChunkRecord]: - if suffix in MARKDOWN_SUFFIXES: - return _process_markdown(abs_path, text, pipelines) - elif suffix in CODE_SUFFIXES: - return _process_code(abs_path, text, pipelines) - else: - return _process_plain(abs_path, text, pipelines) - - -def _process_markdown( - abs_path: str, - text: str, - pipelines: dict[PipelineKind, Pipeline], -) -> list[ChunkRecord]: - result = pipelines["markdown"].run(texts=text) - assert isinstance(result, MarkdownDocument), ( - f"markdown pipeline returned {type(result).__name__}" - ) - doc = result - - headings = _scan_headings(text) - line_offsets = _build_line_offsets(text) - candidates: list[ChunkRecord] = [] - - for chunk in doc.chunks: - if not chunk.text.strip(): - continue - sc, ec = chunk.start_index, chunk.end_index - candidates.append( - ChunkRecord( - chunk_id=_make_id(), - abs_path=abs_path, - chunk_index=0, - content=chunk.text, - token_count=chunk.token_count, - start_line=_offset_to_line(line_offsets, sc), - end_line=_offset_to_line(line_offsets, max(0, ec - 1)), - start_char=sc, - end_char=ec, - section=_nearest_heading(headings, sc), - kind="markdown_text", - context=chunk.context, - ) - ) - - for code in doc.code: - if not code.content.strip(): - continue - sc, ec = code.start_index, code.end_index - metadata = json.dumps({"language": code.language}) if code.language else None - candidates.append( - ChunkRecord( - chunk_id=_make_id(), - abs_path=abs_path, - chunk_index=0, - content=code.content, - token_count=len(code.content.split()), - start_line=_offset_to_line(line_offsets, sc), - end_line=_offset_to_line(line_offsets, max(0, ec - 1)), - start_char=sc, - end_char=ec, - section=_nearest_heading(headings, sc), - kind="markdown_code", - chunk_metadata=metadata, - ) - ) - - for table in doc.tables: - if not table.content.strip(): - continue - sc, ec = table.start_index, table.end_index - candidates.append( - ChunkRecord( - chunk_id=_make_id(), - abs_path=abs_path, - chunk_index=0, - content=table.content, - token_count=len(table.content.split()), - start_line=_offset_to_line(line_offsets, sc), - end_line=_offset_to_line(line_offsets, max(0, ec - 1)), - start_char=sc, - end_char=ec, - section=_nearest_heading(headings, sc), - kind="markdown_table", - ) - ) - - for image in doc.images: - if not image.alias: - continue - sc, ec = image.start_index, image.end_index - candidates.append( - ChunkRecord( - chunk_id=_make_id(), - abs_path=abs_path, - chunk_index=0, - content=image.alias, - token_count=len(image.alias.split()), - start_line=_offset_to_line(line_offsets, sc), - end_line=_offset_to_line(line_offsets, max(0, ec - 1)), - start_char=sc, - end_char=ec, - section=_nearest_heading(headings, sc), - kind="markdown_image", - ) - ) - - candidates.sort(key=lambda c: c.start_char) - return [_replace(c, chunk_index=i) for i, c in enumerate(candidates)] - - -def _process_code( - abs_path: str, - text: str, - pipelines: dict[PipelineKind, Pipeline], -) -> list[ChunkRecord]: - result = pipelines["code"].run(texts=text) - assert isinstance(result, Document), ( - f"code pipeline returned {type(result).__name__}" - ) - doc = result - line_offsets = _build_line_offsets(text) - records: list[ChunkRecord] = [] - for i, chunk in enumerate(doc.chunks): - sc, ec = chunk.start_index, chunk.end_index - records.append( - ChunkRecord( - chunk_id=_make_id(), - abs_path=abs_path, - chunk_index=i, - content=chunk.text, - token_count=chunk.token_count, - start_line=_offset_to_line(line_offsets, sc), - end_line=_offset_to_line(line_offsets, max(0, ec - 1)), - start_char=sc, - end_char=ec, - section=None, - kind="code", - chunk_metadata=None, - context=chunk.context, - ) - ) - return records - - -def _process_plain( - abs_path: str, - text: str, - pipelines: dict[PipelineKind, Pipeline], -) -> list[ChunkRecord]: - result = pipelines["plain"].run(texts=text) - assert isinstance(result, Document), ( - f"plain pipeline returned {type(result).__name__}" - ) - doc = result - line_offsets = _build_line_offsets(text) - records: list[ChunkRecord] = [] - for i, chunk in enumerate(doc.chunks): - sc, ec = chunk.start_index, chunk.end_index - records.append( - ChunkRecord( - chunk_id=_make_id(), - abs_path=abs_path, - chunk_index=i, - content=chunk.text, - token_count=chunk.token_count, - start_line=_offset_to_line(line_offsets, sc), - end_line=_offset_to_line(line_offsets, max(0, ec - 1)), - start_char=sc, - end_char=ec, - section=None, - kind="text", - context=chunk.context, - ) - ) - return records - - -# --------------------------------------------------------------------------- -# Heading scanning and section assignment -# --------------------------------------------------------------------------- - - -def _scan_headings(text: str) -> list[tuple[int, str]]: - return [(m.start(), m.group(0).strip()) for m in _HEADING_RE.finditer(text)] - - -def _nearest_heading(headings: list[tuple[int, str]], char_offset: int) -> str | None: - best = None - for offset, heading in headings: - if offset <= char_offset: - best = heading - else: - break - return best - - -# --------------------------------------------------------------------------- -# Utility functions -# --------------------------------------------------------------------------- - - -_NEWLINE_RE = re.compile(r"\n") - - -def _build_line_offsets(text: str) -> list[int]: - return [0] + [m.end() for m in _NEWLINE_RE.finditer(text)] - - -def _offset_to_line(offsets: list[int], char_offset: int) -> int: - lo, hi = 0, len(offsets) - 1 - while lo < hi: - mid = (lo + hi + 1) // 2 - if offsets[mid] <= char_offset: - lo = mid - else: - hi = mid - 1 - return lo + 1 - - -def _file_hash(path: Path) -> str: - h = hashlib.sha256() - with open(path, "rb") as f: - for block in iter(lambda: f.read(65536), b""): - h.update(block) - return h.hexdigest() - - -def _config_signature(config: WorkspaceConfig) -> str: - ch = config.knowledgebase.chunking - blob = json.dumps( - { - "chunk_size": ch.chunk_size, - "overlap": ch.overlap, - "overlap_mode": "token", - "overlap_method": "suffix", - "code_chunker": "production_v1", - "chunking_plan_version": CHUNKING_PLAN_VERSION, - }, - sort_keys=True, - ) - return hashlib.sha256(blob.encode()).hexdigest()[:16] - - -def _make_id() -> str: - return f"chnk_{uuid.uuid4().hex[:12]}" - - -def _prune_stale(store: SQLiteFTS5Store, disk_paths: set[str]) -> int: - indexed = store.all_indexed_paths() - stale = indexed - disk_paths - for path in stale: - store.delete_file(path) - return len(stale) diff --git a/workspace/search.py b/workspace/search.py deleted file mode 100644 index 197d14801c..0000000000 --- a/workspace/search.py +++ /dev/null @@ -1,37 +0,0 @@ -"""Workspace search API. - -Thin wrapper around SQLiteFTS5Store.search() that handles config loading -and store lifecycle. -""" - -from workspace.config import WorkspaceConfig -from workspace.constants import resolve_path_prefix -from workspace.store import SQLiteFTS5Store -from workspace.types import SearchResult - - -def search_workspace( - query: str, - config: WorkspaceConfig, - *, - limit: int | None = None, - path_prefix: str | None = None, - file_glob: str | None = None, -) -> list[SearchResult]: - if limit is None: - limit = config.knowledgebase.search.default_limit - - # Resolve symlinks + relative segments so the byte-prefix match in the store - # aligns with the indexer, which stores resolved absolute paths - # (`str(file_path.resolve())` in `indexer.py`). Mirrors what `commands.py` - # already does for CLI search — without this, callers who hand in a - # symlinked path via the Python API silently get zero hits. - resolved_prefix = resolve_path_prefix(path_prefix) - - with SQLiteFTS5Store(config.workspace_root) as store: - return store.search( - query, - limit=limit, - path_prefix=resolved_prefix, - file_glob=file_glob, - )