codenib

Public CodeNib API.

Exports are resolved lazily so importing :mod:codenib or starting the CLI does not initialize optional graph, embedding, and agent runtimes.

Modules:

Name	Description
agent	Agent package consolidating agent utilities and implementations.
artifacts	Portable repository-context artifacts.
cli	User-facing CodeNib command line interface.
clients	Runnable agent policies and external-agent execution adapters.
code_chunker	Code chunking module with both file-level and repository-level functionality.
code_chunking	Code chunking module for splitting source code files into semantic chunks.
codegraph_onboarding	Managed MCP client registration for the CodeGraph product path.
compat_pickle	Read pickles written before the package rename.
compiler	Compiler infrastructure for CodeNib.
context_delivery	Bounded, query-focused delivery of repository evidence.
dataset
dataset_ids	Canonical identifiers for externally hosted CodeNib datasets.
eval	Evaluation harnesses: agent runners, retrieval metrics, artifact bundles.
facts	Provider-neutral, per-file semantic fact contracts.
git_snapshot	Git-backed source boundaries for reproducible repository artifacts.
graph	Graph-related modules for CodeNib.
index	Index-related components: sparse, dense, and regex.
integrations	Compatibility providers for external coding-agent runtimes.
languages	Central language metadata registry.
llm	LLM module for CodeNib.
log_utils
ls_index	Language-server indexers and decoders.
ls_router	Unified language-server router for indexing and decoding across all languages.
mcp	CodeNib MCP server - exposes backbone capabilities over stdio.
model	Model pipelines and planners exposed without eager optional imports.
native_index_authorization	Out-of-band authorization for native index parsers.
ops	Composable retrieval operators: retrieve, expand, filter, rerank, transform.
paths	Canonical filesystem locations owned by CodeNib.
profiler
provider_routes	Secret-free provider identities and process-local inference routes.
repository_filters	Shared repository traversal policy for user-facing index builders.
repository_source_selection	Canonical, backend-neutral repository source-selection policy.
repository_summary	Extract a concise project purpose from repository README files.
sandbox	Provider-neutral isolation for repository execution.
scip_interface	Language-specific SCIP indexing and decoding.
search
serving	Retrieval-augmented speculative serving for code agents.
source_fingerprint	Deterministic content identity for repository files visible to CodeNib.
storage	Wiki-only database boundary.
toolchains	Managed external language tools for graph and LSP providers.
types
utils
web	DeepWiki-style demo backend.
wiki	Index-derived wiki generation for the DeepWiki-style demo.

Classes:

Name	Description
PublishedWorkspaceReceiptOwner	Caller-precreated one-shot owner for a published workspace generation.
LocalWorkspaceProvider	Provision strict workspaces below one private Linux authority root.
KeywordExtractor	Agent for extracting keywords from problem statements.
RerankAgent	Agent for reranking code nodes based on query relevance using LLM APIs.
CodeChunker	Unified code chunker that supports both file-level and repository-level processing.
RepoChunkingConfig	Configuration for repository chunking.
CodeGraph	A class to represent and manipulate a code graph using igraph.
CodeVectorStore	Vector store for code embeddings using FAISS and sentence-transformers.
RegexNodeIndex	In-memory regex-based index for CodeGraph nodes.
BM25CodeIndexer	A class that builds a BM25 index from CodeGraph nodes and provides
LSIndexer	Unified indexer that delegates to language-specific implementations.
CodeSearchEngine	Integrated search engine that combines SCIP indexing, code graph representation,

Functions:

Name	Description
create_code_vector_store	Factory function to create a CodeVectorStore.

PublishedWorkspaceReceiptOwner

PublishedWorkspaceReceiptOwner()

Caller-precreated one-shot owner for a published workspace generation.

Methods:

Name	Description
consume	Synchronously borrow one exact authenticated publication reader.

Attributes:

Name	Type	Description
destination_binding	PublishedWorkspaceDestinationBinding	Project the exact destination binding from this active generation.

Source code in codenib/_captured_directory.py

def __init__(self) -> None:
    self._slot: object = _WORKSPACE_RECEIPT_EMPTY
    self._lock = _CancellationSafeRLock()
    self._owner_pid = os.getpid()
    self._process_locks = {self._owner_pid: self._lock}

destination_binding property

destination_binding: PublishedWorkspaceDestinationBinding

Project the exact destination binding from this active generation.

consume

consume(
    callback: Callable[
        [PublishedWorkspaceReceipt, PublicationDirectoryReader], _WorkspaceResult
    ],
    *,
    check_cancelled: Callable[[], None] | None = None
) -> _WorkspaceResult

Synchronously borrow one exact authenticated publication reader.

Source code in codenib/_captured_directory.py

def consume(
    self,
    callback: Callable[
        [PublishedWorkspaceReceipt, PublicationDirectoryReader],
        _WorkspaceResult,
    ],
    *,
    check_cancelled: Callable[[], None] | None = None,
) -> _WorkspaceResult:
    """Synchronously borrow one exact authenticated publication reader."""

    if not callable(callback):
        raise TypeError("published workspace receipt consumer must be callable")
    if check_cancelled is not None and not callable(check_cancelled):
        raise TypeError(
            "published workspace receipt cancellation check must be callable"
        )
    self._require_owner_pid()
    if self._lock.held_by_current_thread():
        raise RuntimeError("published workspace receipt consume is reentrant")

    def consume_locked() -> _WorkspaceResult:
        self._normalize_closed_locked()
        if not isinstance(self._slot, _PUBLISHED_WORKSPACE_RECEIPT_TYPE):
            raise RuntimeError(
                "published workspace receipt owner is "
                f"{self._state_locked()}, expected active"
            )
        return self._slot._consume_from_owner(
            _WORKSPACE_RECEIPT_CLOSE,
            callback,
            check_cancelled=check_cancelled,
        )

    return self._lock.run(consume_locked)

LocalWorkspaceProvider dataclass

LocalWorkspaceProvider(
    allowed_root: Path, provision_timeout_ns: int = _DEFAULT_PROVISION_TIMEOUT_NS
)

Provision strict workspaces below one private Linux authority root.

KeywordExtractor

KeywordExtractor(llm: LiteLLMChat)

Agent for extracting keywords from problem statements.

Parameters:

Name	Type	Description	Default
llm	LiteLLMChat	A configured LiteLLMChat instance.	required

Methods:

Name	Description
extract_keywords	Extract keywords from a problem statement.

Source code in codenib/agent/extract_agent.py

def __init__(
    self,
    llm: LiteLLMChat,
):
    """Initialize the keyword extractor.

    Args:
        llm: A configured LiteLLMChat instance.
    """
    self.llm = llm
    self.structured_llm = self.llm.with_structured_output(KeywordExtraction)

extract_keywords

extract_keywords(problem_statement: str) -> KeywordExtraction

Extract keywords from a problem statement.

Parameters:

Name	Type	Description	Default
problem_statement	str	The problem statement to extract keywords from	required

Returns:

Name	Type	Description
KeywordExtraction	KeywordExtraction	Structured output with extracted keywords

Source code in codenib/agent/extract_agent.py

def extract_keywords(self, problem_statement: str) -> KeywordExtraction:
    """
    Extract keywords from a problem statement.

    Args:
        problem_statement (str): The problem statement to extract keywords from

    Returns:
        KeywordExtraction: Structured output with extracted keywords
    """
    # Create prompt with detailed instructions
    prompt = (
        "You are a keyword extraction specialist. "
        "Your task is to extract important keywords "
        "from problem statements. Focus on identifying "
        "technical terms, function names, class names, "
        "modules, file paths, and concepts that would be "
        "useful for searching in a codebase. "
        "\n\n"
        "Guidelines for extraction:\n"
        "1. Extract file paths and file names "
        "(e.g., 'django/db/models/expressions.py'"
        "-> and 'expressions.py')\n"
        "2. Extract function and method names "
        "(e.g., 'separability_matrix', 'run_validators')\n"
        "3. Extract class names and module names\n"
        "4. Prefer precise terms over general ones\n"
        "5. Remove common stopwords and general "
        "programming terms\n"
        "\n\n"
        "Please extract the key technical terms and "
        "concepts from the following problem statement:"
        f"\n\n{problem_statement}\n\n"
        "Return only the essential terms that would be "
        "most useful for searching in a codebase."
    )

    # Use structured LLM to get output directly as a KeywordExtraction object
    input_msg = human_message(prompt)
    result = self.structured_llm.invoke([input_msg])
    logger.debug(f"Extracted keywords: {result}")
    return result

RerankAgent

RerankAgent(
    llm: LiteLLMChat, listwise_format: Literal["structured", "rankgpt"] = "structured"
)

Agent for reranking code nodes based on query relevance using LLM APIs.

Parameters:

Name	Type	Description	Default
llm	LiteLLMChat	A configured LiteLLMChat instance.	required
listwise_format	Literal['structured', 'rankgpt']	How to ask the LLM to format the ranking. "structured" (default) uses a JSON schema enforced via with_structured_output; suitable for general-purpose LLMs that follow JSON-mode prompts well. "rankgpt" uses SweRank/RankGPT-style text output ([3] > [5] > [1] > ...) and a regex parser. Required for listwise rerankers fine-tuned on this format (Salesforce/SweRankLLM-*, RankZephyr, etc.) — forcing JSON on them collapses the output to the first index only.	'structured'

Methods:

Name	Description
rerank_nodes	Rerank nodes based on their relevance to the query.

Source code in codenib/agent/rerank_agent.py

def __init__(
    self,
    llm: LiteLLMChat,
    listwise_format: Literal["structured", "rankgpt"] = "structured",
):
    """Initialize the rerank agent.

    Args:
        llm: A configured LiteLLMChat instance.
        listwise_format: How to ask the LLM to format the ranking.
            ``"structured"`` (default) uses a JSON schema enforced via
            ``with_structured_output``; suitable for general-purpose LLMs
            that follow JSON-mode prompts well.
            ``"rankgpt"`` uses SweRank/RankGPT-style text output
            (``[3] > [5] > [1] > ...``) and a regex parser. Required for
            listwise rerankers fine-tuned on this format
            (Salesforce/SweRankLLM-*, RankZephyr, etc.) — forcing JSON on
            them collapses the output to the first index only.
    """
    if listwise_format not in _LISTWISE_FORMATS:
        supported = ", ".join(sorted(_LISTWISE_FORMATS))
        raise ValueError(f"listwise_format must be one of: {supported}")

    self.llm = llm
    self.listwise_format = listwise_format
    self.structured_llm = (
        self.llm.with_structured_output(RerankResult)
        if listwise_format == "structured"
        else None
    )
    logger.info(
        "Initialized rerank agent with model=%s listwise_format=%s",
        self.llm.model,
        listwise_format,
    )

rerank_nodes

rerank_nodes(
    query: str,
    nodes: list[NodeInfo],
    top_k: int | None = None,
    window_size: int | None = None,
    window_step: int | None = None,
    include_content: bool = False,
) -> list[QueriedNode]

Rerank nodes based on their relevance to the query.

Parameters:

Name	Type	Description	Default
query	str	The query to rank nodes against	required
nodes	list[NodeInfo]	List of nodes with content to rank	required
top_k	int | None	Maximum number of results to return (None for all)	None
window_size	int | None	Number of nodes per rerank window. None -> all nodes.	None
window_step	int | None	Step size between sliding windows. Defaults to window_size.	None
include_content	bool	Whether to include node content in the result objects.	False

Returns:

Type	Description
list[QueriedNode]	List[QueriedNode]: Ranked nodes with relevance scores (optionally with content)

Notes

When a sliding window is configured, each window is reranked independently and the averaged scores across all windows determine the final ordering.

Source code in codenib/agent/rerank_agent.py

def rerank_nodes(
    self,
    query: str,
    nodes: List[NodeInfo],
    top_k: Optional[int] = None,
    window_size: Optional[int] = None,
    window_step: Optional[int] = None,
    include_content: bool = False,
) -> List[QueriedNode]:
    """
    Rerank nodes based on their relevance to the query.

    Args:
        query (str): The query to rank nodes against
        nodes (List[NodeInfo]): List of nodes with content to rank
        top_k (Optional[int]): Maximum number of results to return (None for all)
        window_size (Optional[int]): Number of nodes per rerank window. None -> all nodes.
        window_step (Optional[int]): Step size between sliding
            windows. Defaults to window_size.
        include_content (bool): Whether to include node content in the result objects.

    Returns:
        List[QueriedNode]: Ranked nodes with relevance scores (optionally with content)

    Notes:
        When a sliding window is configured, each window is reranked independently and
        the averaged scores across all windows determine the final ordering.
    """
    if top_k is not None and (
        isinstance(top_k, bool) or not isinstance(top_k, int) or top_k < 0
    ):
        raise ValueError("top_k must be a non-negative integer or None")
    if top_k == 0:
        return []
    if not nodes:
        logger.warning("No nodes provided for reranking")
        return []

    if not query.strip():
        logger.warning("Empty query provided for reranking")
        return []

    try:
        # Filter nodes with content
        valid_nodes: List[Tuple[int, NodeInfo]] = []
        for i, node in enumerate(nodes):
            if node.content and node.content.strip():
                valid_nodes.append((i, node))

        if not valid_nodes:
            logger.warning(
                "No nodes with content found for reranking; preserving "
                "the first-stage candidate order."
            )
            return self._preserve_first_stage(
                nodes,
                top_k=top_k,
                include_content=include_content,
            )

        logger.info(
            f"Reranking {len(valid_nodes)} nodes with query: {query[:100]}..."
        )

        node_lookup: Dict[int, NodeInfo] = dict(enumerate(nodes))

        total_nodes = len(valid_nodes)
        window_size = window_size or total_nodes
        if window_size <= 0:
            window_size = total_nodes

        window_step = window_step or window_size
        if window_step <= 0:
            window_step = window_size

        aggregated_scores: Dict[int, float] = defaultdict(float)
        appearance_count: Counter[int] = Counter()

        window_starts = list(range(0, total_nodes, window_step))
        tail_start = max(total_nodes - window_size, 0)
        if tail_start not in window_starts:
            window_starts.append(tail_start)
        window_starts = sorted(set(window_starts))

        window_id = 0
        for window_start in window_starts:
            window_nodes = valid_nodes[window_start : window_start + window_size]
            if not window_nodes:
                continue

            window_id += 1
            logger.debug(
                "Processing rerank window %s (nodes %s-%s)",
                window_id,
                window_start,
                window_start + len(window_nodes) - 1,
            )

            window_results = self._rerank_window(query, window_nodes)
            for original_idx, score in window_results:
                aggregated_scores[original_idx] += float(score)
                appearance_count[original_idx] += 1

        if not aggregated_scores:
            logger.warning(
                "Reranker did not return any usable scores; preserving "
                "the first-stage candidate order."
            )

        averaged_scores = {
            idx: aggregated_scores[idx] / appearance_count[idx]
            for idx in aggregated_scores
            if appearance_count[idx] > 0
        }

        sorted_indices = sorted(
            averaged_scores.items(), key=lambda item: item[1], reverse=True
        )
        ranked_ids = {idx for idx, _score in sorted_indices}
        sorted_indices.extend(
            (
                original_idx,
                float(node.score) if node.score is not None else 0.0,
            )
            for original_idx, node in enumerate(nodes)
            if original_idx not in ranked_ids
        )

        ranked_nodes: List[QueriedNode] = []
        for original_idx, score in sorted_indices:
            node = node_lookup.get(original_idx)
            if not node:
                continue
            ranked_nodes.append(
                self._to_queried_node(
                    node,
                    score=float(score),
                    include_content=include_content,
                )
            )
            if top_k and len(ranked_nodes) >= top_k:
                break

        logger.info(
            "Successfully reranked %s nodes across %s window(s)",
            len(ranked_nodes),
            window_id or 1,
        )
        return ranked_nodes

    except Exception as e:
        logger.error(f"Error during reranking: {e}")
        return self._preserve_first_stage(
            nodes,
            top_k=top_k,
            include_content=include_content,
        )

CodeChunker

CodeChunker(
    language: str = "python",
    repo_config: RepoChunkingConfig | None = None,
    max_lines_per_chunk: int | None = None,
    chunk_depth: int = 2,
    include_header_epilogue: bool = False,
    l2_level_exclusive: bool = True,
    skeleton_mode: bool = False,
    include_l2_in_file_skeleton: bool = True,
)

Unified code chunker that supports both file-level and repository-level processing. Maintains backward compatibility with the old API while adding repository functionality.

Parameters:

Name	Type	Description	Default
language	str	Programming language to parse ('python', 'cpp', 'java', etc.)	'python'
repo_config	RepoChunkingConfig | None	Configuration for repository-level chunking. Uses defaults if None.	None
max_lines_per_chunk	int | None	Maximum number of lines per emitted chunk. Default: None (no splitting)	None
chunk_depth	int	Depth of AST traversal (1=top-level only, 2=include methods)	2
include_header_epilogue	bool	Whether to include file headers and epilogues. Default: False	False
l2_level_exclusive	bool	When chunk_depth is 2, whether to omit L1 container nodes (classes/structs/impls) and emit only L2 members. Default: True.	True
skeleton_mode	bool	Emit signature-only skeletons instead of full bodies when True.	False
include_l2_in_file_skeleton	bool	When chunk_depth is 0, include member signatures in file-level skeletons. Default: True.	True

Methods:

Name	Description
chunk_file	Chunk a code file into function/class level pieces.
save_chunks_to_json	Save chunks to a JSON file.
print_chunk_summary	Print a summary of the generated chunks.
chunk_repository	Chunk all relevant files in a repository.
chunk_repository_source	Chunk one retained repository generation without path-based reads.
get_repository_stats	Get statistics about the repository without processing files.
print_repository_summary	Print a summary of all chunks from the repository.
chunk_swebench_instance	Chunk a SWE-bench instance repository.
save_swebench_chunks	Save SWE-bench chunking result to JSON file.
clear_nodes	Clear the collected nodes list.

Source code in codenib/code_chunker.py

def __init__(
    self,
    language: str = "python",
    repo_config: Optional[RepoChunkingConfig] = None,
    max_lines_per_chunk: Optional[int] = None,
    chunk_depth: int = 2,
    include_header_epilogue: bool = False,
    l2_level_exclusive: bool = True,
    skeleton_mode: bool = False,
    include_l2_in_file_skeleton: bool = True,
):
    """
    Initialize the code chunker for a specific language.

    Args:
        language: Programming language to parse ('python', 'cpp', 'java', etc.)
        repo_config: Configuration for repository-level chunking. Uses defaults if
            None.
        max_lines_per_chunk: Maximum number of lines per emitted chunk.
            Default: None (no splitting)
        chunk_depth: Depth of AST traversal (1=top-level only, 2=include methods)
        include_header_epilogue: Whether to include file headers and epilogues.
            Default: False
        l2_level_exclusive: When chunk_depth is 2, whether to omit L1 container
            nodes (classes/structs/impls) and emit only L2 members. Default: True.
        skeleton_mode: Emit signature-only skeletons instead of full bodies when True.
        include_l2_in_file_skeleton: When chunk_depth is 0, include member
            signatures in file-level skeletons. Default: True.
    """
    self.language = normalize_chunker_language(language) or language
    self.max_lines_per_chunk = max_lines_per_chunk
    self.chunk_depth = chunk_depth
    self.include_header_epilogue = include_header_epilogue
    self.l2_level_exclusive = l2_level_exclusive
    self.skeleton_mode = skeleton_mode
    self.include_l2_in_file_skeleton = include_l2_in_file_skeleton
    self._chunker = create_chunker(
        language,
        max_lines_per_chunk=self.max_lines_per_chunk,
        chunk_depth=self.chunk_depth,
        include_header_epilogue=self.include_header_epilogue,
        l2_level_exclusive=self.l2_level_exclusive,
        skeleton_mode=self.skeleton_mode,
        include_l2_in_file_skeleton=self.include_l2_in_file_skeleton,
    )
    if repo_config is None:
        repo_config = RepoChunkingConfig()
    if not repo_config.languages:
        repo_config.languages = [self.language]
    self.repo_config = repo_config
    self._chunkers = {self.language: self._chunker}  # Cache chunkers by language
    self.nodes = (
        []
    )  # List of node IDs in code graph format (dir/file.py:A.b(), dir/file.py:A)

chunk_file

chunk_file(
    file_path: str, relative_path: str | None = None, skeleton_mode: bool | None = None
)

Chunk a code file into function/class level pieces.

Parameters:

Name	Type	Description	Default
file_path	str	Path to the code file to chunk	required
relative_path	str | None	Relative path for node_id generation. If None, uses file_path.	None
skeleton_mode	bool | None	Override instance-level skeleton setting. When True, chunk content contains signatures only.	None

Returns:

Type	Description
	List of CodeChunk objects

Source code in codenib/code_chunker.py

def chunk_file(
    self,
    file_path: str,
    relative_path: Optional[str] = None,
    skeleton_mode: Optional[bool] = None,
):
    """
    Chunk a code file into function/class level pieces.

    Args:
        file_path: Path to the code file to chunk
        relative_path: Relative path for node_id generation. If None, uses file_path.
        skeleton_mode: Override instance-level skeleton setting. When True, chunk content
            contains signatures only.

    Returns:
        List of CodeChunk objects
    """
    chunks = self._chunker.chunk_file(
        file_path, relative_path=relative_path, skeleton_mode=skeleton_mode
    )
    # Collect unique node IDs from chunks
    self._update_nodes_from_chunks(chunks)
    return chunks

save_chunks_to_json

save_chunks_to_json(chunks, output_path: str)

Save chunks to a JSON file.

Source code in codenib/code_chunker.py

def save_chunks_to_json(self, chunks, output_path: str):
    """Save chunks to a JSON file."""
    return self._chunker.save_chunks_to_json(chunks, output_path)

print_chunk_summary

print_chunk_summary(chunks)

Print a summary of the generated chunks.

Source code in codenib/code_chunker.py

def print_chunk_summary(self, chunks):
    """Print a summary of the generated chunks."""
    return self._chunker.print_chunk_summary(chunks)

chunk_repository

chunk_repository(repo_path: str, *, strict: bool = False) -> list[CodeChunk]

Chunk all relevant files in a repository.

Parameters:

Name	Type	Description	Default
repo_path	str	Path to the repository root. Languages come from RepoChunkingConfig; when empty they are inferred from file extensions present in the repo.	required
strict	bool	Raise when any discovered source file cannot be chunked instead of returning a partial repository result.	False

Returns:

Type	Description
list[CodeChunk]	List of CodeChunk objects from all processed files

Raises:

Type	Description
ValueError	If repo_path doesn't exist or languages are unsupported

Source code in codenib/code_chunker.py

def chunk_repository(
    self, repo_path: str, *, strict: bool = False
) -> List[CodeChunk]:
    """
    Chunk all relevant files in a repository.

    Args:
        repo_path: Path to the repository root.
            Languages come from RepoChunkingConfig; when empty they are inferred
            from file extensions present in the repo.
        strict: Raise when any discovered source file cannot be chunked instead
            of returning a partial repository result.

    Returns:
        List of CodeChunk objects from all processed files

    Raises:
        ValueError: If repo_path doesn't exist or languages are unsupported
    """
    repo_path = Path(repo_path).resolve()
    if not repo_path.exists():
        raise ValueError(f"Repository path does not exist: {repo_path}")

    if not repo_path.is_dir():
        raise ValueError(f"Repository path is not a directory: {repo_path}")

    source_selection = self._source_selection_snapshot()
    languages = self.repo_config.languages or self._detect_repo_language(
        repo_path,
        source_selection=source_selection,
    )

    logger.info(f"Chunking repository: {repo_path}")
    logger.info(f"Target languages: {', '.join(languages)}")

    # Discover files
    files_to_process = self._discover_files(
        repo_path,
        languages,
        source_selection=source_selection,
    )
    logger.info(f"Found {len(files_to_process)} files to process")

    # Process files
    all_chunks = []
    processed_count = 0

    for file_path, language in files_to_process:
        try:
            chunks = self._chunk_file_with_language(file_path, language, repo_path)
            all_chunks.extend(chunks)
            processed_count += 1

        except Exception as e:
            if strict:
                relative = file_path.relative_to(repo_path).as_posix()
                raise RuntimeError(
                    f"Failed to chunk repository source {relative}: {e}"
                ) from e
            logger.warning(f"Failed to process {file_path}: {e}")
            continue

    self._require_selected_chunks(
        all_chunks,
        source_selection,
        repository_root=repo_path,
    )

    logger.info(
        f"Successfully processed {processed_count}/{len(files_to_process)} files"
    )
    logger.info(f"Generated {len(all_chunks)} total chunks")

    # Collect unique node IDs from all chunks
    self._update_nodes_from_chunks(all_chunks)
    logger.info(f"Collected {len(self.nodes)} unique nodes")

    return all_chunks

chunk_repository_source

chunk_repository_source(
    repository_source: RepositorySourceBinding,
    *,
    check_cancelled: Callable[[], None] | None = None
) -> list[CodeChunk]

Chunk one retained repository generation without path-based reads.

The binding supplies the complete authenticated file inventory and all source bytes. Discovery, filtering, decoding, and parsing therefore operate on that fixed generation even when the repository is also visible through a mutable public pathname.

Source code in codenib/code_chunker.py

def chunk_repository_source(
    self,
    repository_source: RepositorySourceBinding,
    *,
    check_cancelled: Callable[[], None] | None = None,
) -> List[CodeChunk]:
    """Chunk one retained repository generation without path-based reads.

    The binding supplies the complete authenticated file inventory and all
    source bytes.  Discovery, filtering, decoding, and parsing therefore
    operate on that fixed generation even when the repository is also
    visible through a mutable public pathname.
    """

    if type(repository_source) is not RepositorySourceBinding:
        raise TypeError(
            "authenticated repository chunking requires an exact source binding"
        )
    if check_cancelled is not None and not callable(check_cancelled):
        raise TypeError("authenticated repository cancellation must be callable")

    if check_cancelled is not None:
        check_cancelled()
    source_identity = repository_source.authenticated_identity_snapshot(
        check_cancelled=check_cancelled,
    )
    if check_cancelled is not None:
        check_cancelled()
    source_selection = self._source_selection_snapshot()
    if source_identity.source_selection != source_selection:
        raise RuntimeError(
            "authenticated repository selection differs from chunker policy"
        )
    records = tuple(source_identity.file_records)
    languages = self.repo_config.languages or self._detect_source_languages(
        records,
        source_selection=source_selection,
        check_cancelled=check_cancelled,
    )
    extension_to_language = self._extension_language_map(languages)
    selected: List[tuple[RepositorySourceFileRecord, str]] = []
    for record in records:
        if check_cancelled is not None:
            check_cancelled()
        if self._source_record_is_selected(
            record,
            extension_to_language=extension_to_language,
            repository_root=source_identity.root,
            source_selection=source_selection,
        ):
            selected.append(
                (record, extension_to_language[Path(record.path).suffix])
            )

    logger.info("Chunking authenticated repository: %s", source_identity.root)
    logger.info("Target languages: %s", ", ".join(languages))
    logger.info("Found %d authenticated files to process", len(selected))

    chunks: List[CodeChunk] = []
    with repository_source.read_session(check_cancelled=check_cancelled):
        for record, language in selected:
            if check_cancelled is not None:
                check_cancelled()
            payload = repository_source.read_bytes(
                record.path,
                max_bytes=max(1, record.size),
            )
            if len(payload) != record.size:
                raise RuntimeError(
                    "authenticated repository source returned the wrong byte count"
                )
            source_text = (
                payload.decode("utf-8", errors="replace")
                .replace("\r\n", "\n")
                .replace("\r", "\n")
            )
            if self.repo_config.skip_minified and self._is_minified_source(
                source_text,
                path=record.path,
                check_cancelled=check_cancelled,
            ):
                continue
            try:
                chunks.extend(
                    self._chunk_source_with_language(
                        source_text,
                        record.path,
                        language,
                    )
                )
            except Exception as exc:
                raise RuntimeError(
                    "Failed to chunk authenticated repository source "
                    f"{record.path}: {exc}"
                ) from exc
            if check_cancelled is not None:
                check_cancelled()

    if check_cancelled is not None:
        check_cancelled()
    self._require_selected_chunks(
        chunks,
        source_selection,
        repository_root=source_identity.root,
        check_cancelled=check_cancelled,
    )
    self._update_nodes_from_chunks(
        chunks,
        check_cancelled=check_cancelled,
    )
    logger.info("Generated %d authenticated source chunks", len(chunks))
    return chunks

get_repository_stats

get_repository_stats(repo_path: str) -> dict[str, Any]

Get statistics about the repository without processing files.

Parameters:

Name	Type	Description	Default
repo_path	str	Path to the repository root. Languages come from RepoChunkingConfig; when empty they are inferred.	required

Returns:

Type	Description
dict[str, Any]	Dictionary containing repository statistics

Source code in codenib/code_chunker.py

def get_repository_stats(self, repo_path: str) -> Dict[str, Any]:
    """
    Get statistics about the repository without processing files.

    Args:
        repo_path: Path to the repository root.
            Languages come from RepoChunkingConfig; when empty they are inferred.

    Returns:
        Dictionary containing repository statistics
    """
    repo_path = Path(repo_path).resolve()
    if not repo_path.exists():
        raise ValueError(f"Repository path does not exist: {repo_path}")

    source_selection = self._source_selection_snapshot()
    languages = self.repo_config.languages or self._detect_repo_language(
        repo_path,
        source_selection=source_selection,
    )
    files_to_process = self._discover_files(
        repo_path,
        languages,
        source_selection=source_selection,
    )

    stats = {
        "repo_path": str(repo_path),
        "total_files": len(files_to_process),
        "languages": list(languages),
        "files_by_language": {},
        "total_size_mb": 0.0,
    }

    # Group by language and calculate sizes
    for file_path, language in files_to_process:
        if language not in stats["files_by_language"]:
            stats["files_by_language"][language] = []

        file_size = file_path.stat().st_size / (1024 * 1024)  # MB
        stats["files_by_language"][language].append(
            {
                "path": str(file_path.relative_to(repo_path)),
                "size_mb": round(file_size, 3),
            }
        )
        stats["total_size_mb"] += file_size

    stats["total_size_mb"] = round(stats["total_size_mb"], 3)

    # Add counts by language
    for language in stats["files_by_language"]:
        count = len(stats["files_by_language"][language])
        logger.debug(f"  {language}: {count} files")

    return stats

print_repository_summary

print_repository_summary(chunks: list[CodeChunk])

Print a summary of all chunks from the repository.

Source code in codenib/code_chunker.py

def print_repository_summary(self, chunks: List[CodeChunk]):
    """Print a summary of all chunks from the repository."""
    if not chunks:
        logger.info("No chunks generated")
        return

    logger.info(f"\n=== Repository Chunk Summary ===")
    logger.info(f"Total chunks: {len(chunks)}")

    # Group by file
    chunks_by_file = {}
    chunk_types = {}

    for chunk in chunks:
        if chunk.file not in chunks_by_file:
            chunks_by_file[chunk.file] = []
        chunks_by_file[chunk.file].append(chunk)

        # Count chunk types
        chunk_types[chunk.chunk_type] = chunk_types.get(chunk.chunk_type, 0) + 1

    logger.info(f"Files processed: {len(chunks_by_file)}")
    logger.info("Chunk types:")
    for chunk_type, count in sorted(chunk_types.items()):
        logger.info(f"  {chunk_type}: {count}")

    logger.info(f"\nChunks per file:")
    for file_path, file_chunks in chunks_by_file.items():
        rel_path = Path(file_path).name  # Just show filename for brevity
        logger.info(f"  {rel_path}: {len(file_chunks)} chunks")

chunk_swebench_instance

chunk_swebench_instance(
    instance: dict[str, Any],
    cache_dir: str | None = None,
    languages: list[str] | None = None,
) -> dict[str, Any]

Chunk a SWE-bench instance repository.

Parameters:

Name	Type	Description	Default
instance	dict[str, Any]	SWE-bench instance dictionary containing repo, base_commit, instance_id, etc.	required
cache_dir	str | None	Directory to cache repositories. Defaults to ~/.codenib	None
languages	list[str] | None	List of languages to process. If None, uses repository's primary language	None

Returns:

Type	Description
dict[str, Any]	Dictionary containing instance info, repo path, and chunks

Source code in codenib/code_chunker.py

def chunk_swebench_instance(
    self,
    instance: Dict[str, Any],
    cache_dir: Optional[str] = None,
    languages: Optional[List[str]] = None,
) -> Dict[str, Any]:
    """
    Chunk a SWE-bench instance repository.

    Args:
        instance: SWE-bench instance dictionary containing repo, base_commit,
            instance_id, etc.
        cache_dir: Directory to cache repositories. Defaults to ~/.codenib
        languages: List of languages to process. If None, uses repository's
            primary language

    Returns:
        Dictionary containing instance info, repo path, and chunks
    """
    from .dataset.swebench import SwebenchDataset

    if cache_dir is None:
        cache_dir = str(user_state_dir())

    dataset_obj = SwebenchDataset(root=cache_dir, repo_root=cache_dir)
    dataset_obj.process_instance(instance)
    repo_path = dataset_obj.get_repo_path(instance)

    # Auto-detect primary language if not specified
    if languages is None:
        languages = self._detect_repo_language(Path(repo_path))

    logger.info(f"Processing SWE-bench instance: {instance['instance_id']}")
    logger.info(f"Repository: {instance['repo']}")
    logger.info(f"Base commit: {instance['base_commit']}")
    logger.info(f"Repository path: {repo_path}")

    # Chunk the repository
    previous_languages = list(self.repo_config.languages)
    self.repo_config.languages = languages
    try:
        chunks = self.chunk_repository(repo_path)
    finally:
        self.repo_config.languages = previous_languages

    # Return comprehensive information
    result = {
        "instance_id": instance["instance_id"],
        "repo": instance["repo"],
        "base_commit": instance["base_commit"],
        "problem_statement": instance.get("problem_statement", ""),
        "repo_path": repo_path,
        "chunks": chunks,
        "languages": languages,
        "total_chunks": len(chunks),
        "chunk_summary": self._get_chunk_summary_dict(chunks),
        "nodes": self.nodes.copy(),  # Include collected nodes
    }

    return result

save_swebench_chunks

save_swebench_chunks(swebench_result: dict[str, Any], output_path: str)

Save SWE-bench chunking result to JSON file.

Parameters:

Name	Type	Description	Default
swebench_result	dict[str, Any]	Result from chunk_swebench_instance()	required
output_path	str	Path to save the JSON file	required

Source code in codenib/code_chunker.py

def save_swebench_chunks(self, swebench_result: Dict[str, Any], output_path: str):
    """
    Save SWE-bench chunking result to JSON file.

    Args:
        swebench_result: Result from chunk_swebench_instance()
        output_path: Path to save the JSON file
    """
    import json

    # Create a serializable version (excluding chunks content for size)
    save_data = {
        "instance_id": swebench_result["instance_id"],
        "repo": swebench_result["repo"],
        "base_commit": swebench_result["base_commit"],
        "problem_statement": swebench_result["problem_statement"],
        "repo_path": swebench_result["repo_path"],
        "languages": swebench_result["languages"],
        "total_chunks": swebench_result["total_chunks"],
        "chunk_summary": swebench_result["chunk_summary"],
        "chunks_preview": [
            {
                "file": chunk.file,
                "chunk_type": chunk.chunk_type,
                "name": chunk.name,
                "start_line": chunk.start_line,
                "end_line": chunk.end_line,
                "content_length": len(chunk.content),
            }
            for chunk in swebench_result["chunks"][
                :10
            ]  # Save preview of first 10 chunks
        ],
    }

    try:
        with open(output_path, "w", encoding="utf-8") as f:
            json.dump(save_data, f, indent=2, ensure_ascii=False)
        logger.info(f"SWE-bench chunking result saved to {output_path}")
    except Exception as e:
        logger.error(f"Error saving SWE-bench result: {e}")

clear_nodes

clear_nodes()

Clear the collected nodes list.

Source code in codenib/code_chunker.py

def clear_nodes(self):
    """Clear the collected nodes list."""
    self.nodes.clear()

RepoChunkingConfig dataclass

RepoChunkingConfig(
    languages: list[str] = list(),
    python_extensions: set[str] | None = None,
    cpp_extensions: set[str] | None = None,
    csharp_extensions: set[str] | None = None,
    rust_extensions: set[str] | None = None,
    go_extensions: set[str] | None = None,
    java_extensions: set[str] | None = None,
    ruby_extensions: set[str] | None = None,
    php_extensions: set[str] | None = None,
    kotlin_extensions: set[str] | None = None,
    swift_extensions: set[str] | None = None,
    scala_extensions: set[str] | None = None,
    lua_extensions: set[str] | None = None,
    javascript_extensions: set[str] | None = None,
    typescript_extensions: set[str] | None = None,
    ignore_dirs: set[str] = None,
    include_dirs: set[str] = None,
    source_selection: RepositorySourceSelection = RepositorySourceSelection(),
    ignore_patterns: set[str] = None,
    max_file_size_mb: float = 10.0,
    filter_tests: bool = True,
    skip_minified: bool = True,
    minified_line_threshold: int = 10000,
)

Configuration for repository chunking.

Methods:

Name	Description
__post_init__	Initialize default values.

__post_init__

__post_init__()

Initialize default values.

Source code in codenib/code_chunker.py

def __post_init__(self):
    """Initialize default values."""
    if type(self.source_selection) is not RepositorySourceSelection:
        raise TypeError("source_selection must be a RepositorySourceSelection")
    # Retain a detached snapshot.  RepositorySourceSelection is frozen, but
    # callers can still bypass dataclass guards with object.__setattr__.
    self.source_selection = RepositorySourceSelection(
        self.source_selection.exclude_subtrees
    )

    if not self.languages:
        self.languages = []

    if self.python_extensions is None:
        self.python_extensions = extensions_for_language("python", "chunker")

    if self.cpp_extensions is None:
        self.cpp_extensions = extensions_for_language("cpp", "chunker")

    if self.csharp_extensions is None:
        self.csharp_extensions = extensions_for_language("csharp", "chunker")

    if self.rust_extensions is None:
        self.rust_extensions = extensions_for_language("rust", "chunker")

    if self.go_extensions is None:
        self.go_extensions = extensions_for_language("go", "chunker")

    if self.java_extensions is None:
        self.java_extensions = extensions_for_language("java", "chunker")

    if self.ruby_extensions is None:
        self.ruby_extensions = extensions_for_language("ruby", "chunker")

    if self.php_extensions is None:
        self.php_extensions = extensions_for_language("php", "chunker")

    if self.kotlin_extensions is None:
        self.kotlin_extensions = extensions_for_language("kotlin", "chunker")

    if self.swift_extensions is None:
        self.swift_extensions = extensions_for_language("swift", "chunker")

    if self.scala_extensions is None:
        self.scala_extensions = extensions_for_language("scala", "chunker")

    if self.lua_extensions is None:
        self.lua_extensions = extensions_for_language("lua", "chunker")

    if self.javascript_extensions is None:
        self.javascript_extensions = extensions_for_language(
            "javascript", "chunker"
        )

    if self.typescript_extensions is None:
        self.typescript_extensions = extensions_for_language(
            "typescript", "chunker"
        )

    if self.ignore_dirs is None:
        self.ignore_dirs = set(DEFAULT_IGNORED_DIRS)

    if self.ignore_patterns is None:
        self.ignore_patterns = {
            ".*",
            "*.pyc",
            "*.pyo",
            "*.pyd",
            "*.so",
            "*.dll",
            "*.exe",
            "*.o",
            "*.obj",
            "*~",
            "*.bak",
            "*.tmp",
            "*.min.js",
            "*.min.css",
            "*.min.mjs",
            "*.bundle.js",
            "*.bundle.mjs",
        }

CodeGraph

CodeGraph(project_root=None)

A class to represent and manipulate a code graph using igraph. start_line uses 0-based index.

Methods:

Name	Description
add_file_node	Add a file node to the graph and set it as the current file and scope.
add_symbol_node	Add a symbol node to the graph.
add_symbol_reference	Add a reference to a symbol.
update_current_scope	Update the current scope to the given symbol with its range.
exit_scopes_by_line	Exit scopes that have ended based on current line number.
add_containment_edge	Add a containment edge from current scope to a symbol.
batch_edges	Defer igraph edge insertion until the block exits, flushing all
invalidate_caches	Invalidate derived indexes after direct graph mutation.
build_range_indexes	Rebuild per-file line-range indexes from current graph state.
merge_from	Merge another CodeGraph into this one.
query_range	Query the graph by source-file line range (LSP-aligned).
query_range_by_symbol	Range-query by symbol identity (name).
layers	Build a multi-graph layer index over this graph.
layer	Return one named graph layer view.
add_root_node	Add the root node to the graph
add_directory_node	Add a directory node to the graph
save_graph	Save the graph to a pickle file for fast serialization.
load_graph	Load a graph from a pickle file.
load_graph_stream	Load a graph from a caller-authenticated binary stream.
get_graph	Get the igraph Graph object.
unified_index	Readable unified_name (full + bare) -> [canonical names], cached.
display_name	Readable label for a canonical name (unified_name or itself).
resolve_symbol	Map a readable/bare symbol to (canonical_name | None, candidates).
get_node_info_by_name	Get information about a node in the graph.
get_node_info_by_id	Get information about a node in the graph.
get_node_content	Get the content of a node in the graph.
get_neighbors	Get the neighbors of a node in the graph.
get_successors	Get the successors of a node_name (outgoing edges).
get_predecessors	Get the predecessors of a node_name (incoming edges).
print_graph_basic_info	Print basic information about the graph.
visualize_graph	Visualize the code graph with different colors for different node and edge types.

Source code in codenib/graph/code_graph.py

def __init__(self, project_root=None):
    # Create a directed graph
    self.graph = ig.Graph(directed=True)
    self.current_file = None
    self.current_scope = None
    self.project_root = project_root
    self.scope_stack = []  # List of {symbol: [start_line, end_line]} dicts
    # Store line ranges for symbols
    self.symbol_ranges = {}
    # Map symbol names to vertex IDs
    self.name_to_vertex = {}

    # Range indexes — per-file, populated by build_range_indexes() after the
    # graph is fully built or after a patcher batch. Pickled with the graph.
    # _file_nodes[file] -> list of (start_line, end_line, vid), unsorted.
    # _file_edge_anchors[file] -> sorted list of (anchor_line, eid).
    self._file_nodes: Dict[str, List[Tuple[int, int, int]]] = {}
    self._file_edge_anchors: Dict[str, List[Tuple[int, int]]] = {}

    # Aux: unified_name (display, may collide) -> list of identity names.
    # Populated by build_range_indexes(). Pickled with the graph.
    # Consumers disambiguate via file context (this layer just lists candidates).
    self._unified_to_names: Dict[str, List[str]] = {}

    # Edge dedup index: (src_id, tgt_id, type, anchor_file, anchor_line) -> eid.
    # Populated lazily on first _add_edge / invalidated to None when callers
    # delete edges or vertices (igraph compacts eids on delete, so cached
    # eids would point at wrong edges). Lookup via _ensure_edge_index().
    # Not pickled — rebuilt on demand after load.
    self._edge_index: Optional[Dict[Tuple, int]] = None

    # Edge insertion buffer, active only inside a `batch_edges()` block.
    # igraph rebuilds its adjacency index on every `add_edges` call, so
    # inserting one edge at a time is O(E^2). Decoders wrap their build
    # loop in `batch_edges()` to defer insertion into a single add_edges
    # flush (O(E)). When None, `_add_edge` inserts immediately as before.
    self._edge_buffer: Optional[List[Tuple[int, int]]] = None
    self._edge_attr_buffer: Optional[List[Tuple]] = None

add_file_node

add_file_node(file_path)

Add a file node to the graph and set it as the current file and scope.

Parameters:

Name	Type	Description	Default
file_path		Path of the file to add	required

Source code in codenib/graph/code_graph.py

def add_file_node(self, file_path):
    """
    Add a file node to the graph and set it as the current file and scope.

    Args:
        file_path: Path of the file to add
    """
    self.current_file = file_path

    # Add vertex for file
    self._add_vertex(file_path, {"type": NODE_TYPE_FILE})

    self.current_scope = file_path
    # File scope has no range (special case)
    self.scope_stack = [{file_path: None}]

add_symbol_node

add_symbol_node(
    symbol,
    line,
    scope_start_line=None,
    scope_end_line=None,
    symbol_type=None,
    symbol_kind=None,
)

Add a symbol node to the graph.

Parameters:

Name	Type	Description	Default
symbol		Symbol name	required
line		Line number of the symbol	required
scope_start_line		Start line of the symbol's scope (optional)	None
scope_end_line		End line of the symbol's scope (optional)	None
symbol_type		Type of symbol (NODE_TYPE_CLASS, NODE_TYPE_METHOD, NODE_TYPE_FUNCTION) (optional)	None

Source code in codenib/graph/code_graph.py

def add_symbol_node(
    self,
    symbol,
    line,
    scope_start_line=None,
    scope_end_line=None,
    symbol_type=None,
    symbol_kind=None,
):
    """
    Add a symbol node to the graph.

    Args:
        symbol: Symbol name
        line: Line number of the symbol
        scope_start_line: Start line of the symbol's scope (optional)
        scope_end_line: End line of the symbol's scope (optional)
        symbol_type: Type of symbol
            (NODE_TYPE_CLASS, NODE_TYPE_METHOD, NODE_TYPE_FUNCTION) (optional)
    """
    # Use specific symbol type if provided, otherwise default to generic symbol
    node_type = symbol_type if symbol_type else NODE_TYPE_SYMBOL

    attributes = {
        "type": node_type,
        "file": self.current_file,
        "selection_line": line,
        "has_definition": True,
    }
    if symbol_kind is not None:
        attributes["symbol_kind"] = symbol_kind

    if scope_start_line is not None and scope_end_line is not None:
        # Store symbol range
        self.symbol_ranges[symbol] = (scope_start_line, scope_end_line)

        # Add symbol vertex with scope range
        attributes.update(
            {"start_line": scope_start_line, "end_line": scope_end_line}
        )
        self._add_vertex(symbol, attributes)
    else:
        # Add symbol vertex without scope range
        attributes.update({"start_line": line, "end_line": line})
        self._add_vertex(symbol, attributes)

add_symbol_reference

add_symbol_reference(
    symbol,
    module_path=None,
    symbol_type=None,
    anchor_file: str | None = None,
    anchor_line: int | None = None,
    symbol_kind=None,
)

Add a reference to a symbol.

Parameters:

Name	Type	Description	Default
symbol		Symbol being referenced	required
module_path		Path of the module containing the symbol (optional)	None
symbol_type		Type of symbol (NODE_TYPE_CLASS, NODE_TYPE_METHOD, NODE_TYPE_FUNCTION) (optional)	None
anchor_file	str | None	File where the reference site is located (optional)	None
anchor_line	int | None	0-based line of the reference site (optional)	None

Source code in codenib/graph/code_graph.py

def add_symbol_reference(
    self,
    symbol,
    module_path=None,
    symbol_type=None,
    anchor_file: Optional[str] = None,
    anchor_line: Optional[int] = None,
    symbol_kind=None,
):
    """
    Add a reference to a symbol.

    Args:
        symbol: Symbol being referenced
        module_path: Path of the module containing the symbol (optional)
        symbol_type: Type of symbol
            (NODE_TYPE_CLASS, NODE_TYPE_METHOD, NODE_TYPE_FUNCTION) (optional)
        anchor_file: File where the reference site is located (optional)
        anchor_line: 0-based line of the reference site (optional)
    """
    # If the symbol doesn't exist, create it without range info
    if symbol not in self.name_to_vertex:
        file_attr = module_path if module_path else None
        node_type = symbol_type if symbol_type else NODE_TYPE_SYMBOL
        attributes = {
            "type": node_type,
            "file": file_attr,
            "has_definition": False,
        }
        if symbol_kind is not None:
            attributes["symbol_kind"] = symbol_kind
        self._add_vertex(symbol, attributes)
    elif symbol_kind is not None:
        vertex = self.graph.vs[self.name_to_vertex[symbol]]
        if not vertex.attributes().get("symbol_kind"):
            vertex["symbol_kind"] = symbol_kind

    # Add reference edge with anchor info (defaults: current file)
    if anchor_file is None:
        anchor_file = self.current_file
    self._add_edge(
        self.current_scope,
        symbol,
        EDGE_TYPE_REFERENCE,
        anchor_file=anchor_file,
        anchor_line=anchor_line,
    )

update_current_scope

update_current_scope(symbol, start_line=None, end_line=None)

Update the current scope to the given symbol with its range.

Parameters:

Name	Type	Description	Default
symbol		Symbol to set as current scope	required
start_line		Start line of the symbol's scope	None
end_line		End line of the symbol's scope	None

Source code in codenib/graph/code_graph.py

def update_current_scope(self, symbol, start_line=None, end_line=None):
    """
    Update the current scope to the given symbol with its range.

    Args:
        symbol: Symbol to set as current scope
        start_line: Start line of the symbol's scope
        end_line: End line of the symbol's scope
    """
    self.current_scope = symbol
    # Add symbol with its range to scope stack
    scope_range = (
        [start_line, end_line]
        if start_line is not None and end_line is not None
        else None
    )
    self.scope_stack.append({symbol: scope_range})

exit_scopes_by_line

exit_scopes_by_line(current_line)

Exit scopes that have ended based on current line number. File scope (with None range) is never popped.

Parameters:

Name	Type	Description	Default
current_line		Current line number being processed	required

Source code in codenib/graph/code_graph.py

def exit_scopes_by_line(self, current_line):
    """
    Exit scopes that have ended based on current line number.
    File scope (with None range) is never popped.

    Args:
        current_line: Current line number being processed
    """
    # Pop scopes whose range has ended
    while len(self.scope_stack) > 1:  # Keep at least the file scope
        top_scope_dict = self.scope_stack[-1]
        scope_symbol = list(top_scope_dict.keys())[0]
        scope_range = top_scope_dict[scope_symbol]

        # If scope has no range (file scope), don't pop
        if scope_range is None:
            break

        # If current line is beyond scope's end, pop it
        start_line, end_line = scope_range
        if current_line > end_line:
            self.scope_stack.pop()
            # Update current_scope to new top of stack
            if self.scope_stack:
                new_top = self.scope_stack[-1]
                self.current_scope = list(new_top.keys())[0]
            else:
                self.current_scope = self.current_file
        else:
            # Scope is still active
            break

add_containment_edge

add_containment_edge(target_symbol)

Add a containment edge from current scope to a symbol.

CONTAIN edges deliberately carry no anchor — containment is a structural relation, not a call/reference site. (See anchor invariant ii on query_range.)

Parameters:

Name	Type	Description	Default
target_symbol		Symbol being contained.	required

Source code in codenib/graph/code_graph.py

def add_containment_edge(self, target_symbol):
    """
    Add a containment edge from current scope to a symbol.

    CONTAIN edges deliberately carry no anchor — containment is a
    structural relation, not a call/reference site. (See anchor invariant
    ii on `query_range`.)

    Args:
        target_symbol: Symbol being contained.
    """
    # Use the current scope directly (not parent scope)
    self._add_edge(self.current_scope, target_symbol, EDGE_TYPE_CONTAIN)

batch_edges

batch_edges()

Defer igraph edge insertion until the block exits, flushing all buffered edges in a single add_edges call.

igraph rebuilds its internal adjacency index on every add_edges call, so inserting one edge at a time (as bare _add_edge does) is O(E^2) in the number of edges — the dominant cost when decoding a large repo. Wrapping a build loop in this context collapses that to O(E):

with graph.batch_edges():
    # ... many add_symbol_reference / _add_edge calls ...

Edge dedup is unaffected (the dedup index is updated as edges are buffered). Edge ids returned by _add_edge while buffering are provisional and only become real after flush; no caller relies on them mid-build. Outside this block _add_edge inserts immediately, so the incremental patcher and all other callers are unchanged. Nesting is a no-op (the outermost block owns the flush).

Source code in codenib/graph/code_graph.py

@contextmanager
def batch_edges(self):
    """Defer igraph edge insertion until the block exits, flushing all
    buffered edges in a single ``add_edges`` call.

    igraph rebuilds its internal adjacency index on every ``add_edges``
    call, so inserting one edge at a time (as bare ``_add_edge`` does) is
    O(E^2) in the number of edges — the dominant cost when decoding a large
    repo. Wrapping a build loop in this context collapses that to O(E):

        with graph.batch_edges():
            # ... many add_symbol_reference / _add_edge calls ...

    Edge dedup is unaffected (the dedup index is updated as edges are
    buffered). Edge *ids* returned by ``_add_edge`` while buffering are
    provisional and only become real after flush; no caller relies on them
    mid-build. Outside this block ``_add_edge`` inserts immediately, so the
    incremental patcher and all other callers are unchanged. Nesting is a
    no-op (the outermost block owns the flush)."""
    if self._edge_buffer is not None:
        # Already batching — let the outer block own the buffer/flush.
        yield
        return
    if self._edge_index is None:
        self._rebuild_edge_index()
    self._edge_buffer = []
    self._edge_attr_buffer = []
    try:
        yield
    finally:
        self._flush_edge_buffer()

invalidate_caches

invalidate_caches() -> None

Invalidate derived indexes after direct graph mutation.

Source code in codenib/graph/code_graph.py

def invalidate_caches(self) -> None:
    """Invalidate derived indexes after direct graph mutation."""
    self._edge_index = None

build_range_indexes

build_range_indexes() -> None

Rebuild per-file line-range indexes from current graph state.

Call this after the graph is fully constructed by a decoder, and again after any incremental patcher batch. The two indexes are pickled with the graph by save_graph, so subsequent load_graph calls do not need to rebuild.

Cost: O(V + E) — one linear pass over vertices and edges.

Source code in codenib/graph/code_graph.py

def build_range_indexes(self) -> None:
    """Rebuild per-file line-range indexes from current graph state.

    Call this after the graph is fully constructed by a decoder, and
    again after any incremental patcher batch. The two indexes are
    pickled with the graph by `save_graph`, so subsequent `load_graph`
    calls do not need to rebuild.

    Cost: O(V + E) — one linear pass over vertices and edges.
    """
    nodes: Dict[str, List[Tuple[int, int, int]]] = defaultdict(list)
    for vid in range(self.graph.vcount()):
        v = self.graph.vs[vid]
        attrs = v.attributes()
        f = attrs.get("file")
        s = attrs.get("start_line")
        e = attrs.get("end_line")
        if f and s is not None and e is not None:
            nodes[f].append((s, e, vid))
    self._file_nodes = dict(nodes)

    edges: Dict[str, List[Tuple[int, int]]] = defaultdict(list)
    for eid in range(self.graph.ecount()):
        edge = self.graph.es[eid]
        attrs = edge.attributes()
        f = attrs.get("anchor_file")
        line = attrs.get("anchor_line")
        if f and line is not None:
            edges[f].append((line, eid))
    for f in edges:
        edges[f].sort()
    self._file_edge_anchors = dict(edges)

    unified: Dict[str, List[str]] = defaultdict(list)
    for vid in range(self.graph.vcount()):
        v = self.graph.vs[vid]
        uname = v.attributes().get("unified_name")
        if uname:
            unified[uname].append(v["name"])
    self._unified_to_names = dict(unified)
    self._unified_index_cache = None

merge_from

merge_from(other: CodeGraph) -> None

Merge another CodeGraph into this one.

Vertices are keyed by their canonical name attribute. Existing vertices receive any attributes from other; missing vertices are inserted. Edges are copied through _add_edge so structural edge deduplication and anchored reference multi-edges keep the same semantics as decoder-built graphs.

Source code in codenib/graph/code_graph.py

def merge_from(self, other: "CodeGraph") -> None:
    """Merge another ``CodeGraph`` into this one.

    Vertices are keyed by their canonical ``name`` attribute. Existing
    vertices receive any attributes from ``other``; missing vertices are
    inserted. Edges are copied through ``_add_edge`` so structural edge
    deduplication and anchored reference multi-edges keep the same
    semantics as decoder-built graphs.
    """
    for vertex in other.graph.vs:
        attrs = vertex.attributes()
        name = attrs.get("name")
        if not name:
            continue
        copied = {key: value for key, value in attrs.items() if key != "name"}
        existing_id = self.name_to_vertex.get(name)
        if existing_id is None:
            self._add_vertex(name, copied)
            continue

        existing = self.graph.vs[existing_id].attributes()
        if is_symbol_node(existing.get("type")) and not node_has_definition(copied):
            # A later reference-only shard must not replace first-reference
            # provenance or erase an observed declaration. It may fill
            # additive semantic metadata.
            symbol_kind = copied.get("symbol_kind")
            if symbol_kind and not existing.get("symbol_kind"):
                self.graph.vs[existing_id]["symbol_kind"] = symbol_kind
            continue
        self._add_vertex(name, copied)

    with self.batch_edges():
        for edge in other.graph.es:
            attrs = edge.attributes()
            edge_type = attrs.get("type")
            if edge_type is None:
                continue
            source_name = other.graph.vs[edge.source]["name"]
            target_name = other.graph.vs[edge.target]["name"]
            self._add_edge(
                source_name,
                target_name,
                edge_type,
                anchor_file=attrs.get("anchor_file"),
                anchor_line=attrs.get("anchor_line"),
            )

    self.symbol_ranges.update(other.symbol_ranges)
    self.build_range_indexes()

query_range

query_range(
    file: str,
    start_line: int,
    end_line: int,
    kinds: set | None = None,
    depth: int = 1,
    *,
    layer: str | None = None
) -> RangeQueryResult

Query the graph by source-file line range (LSP-aligned).

Parameters:

Name	Type	Description	Default
file	str	source file (graph-relative path).	required
start_line	int	first line of the inclusive 0-based span.	required
end_line	int	last line of the inclusive 0-based span.	required
kinds	set | None	edge kinds to include in outgoing AND incoming. Defaults to {EDGE_TYPE_REFERENCE} — CONTAIN edges have no meaningful anchor and are excluded by default. Pass an explicit set (e.g. {EDGE_TYPE_REFERENCE, EDGE_TYPE_CONTAIN}) to opt in.	None
depth	int	reserved for future multi-hop expansion. Only depth=1 is supported in v1; raises NotImplementedError otherwise.	1
layer	str | None	optional named graph layer (reference, dependency, containment, all, ...). Mutually exclusive with kinds.	None

Returns: RangeQueryResult with typed NodeRef/EdgeRef records.

Anchor invariants

(i) anchor_file == source.file (anchor lives at the call site, never the target). (ii) anchor_* is meaningful only on EDGE_TYPE_REFERENCE; CONTAIN edges leave both fields None. (iii) outgoing ∩ incoming = ∅ for any single query (call site and target are in distinct scopes by definition).

Source code in codenib/graph/code_graph.py

def query_range(
    self,
    file: str,
    start_line: int,
    end_line: int,
    kinds: Optional[set] = None,
    depth: int = 1,
    *,
    layer: Optional[str] = None,
) -> RangeQueryResult:
    """Query the graph by source-file line range (LSP-aligned).

    Args:
        file: source file (graph-relative path).
        start_line: first line of the inclusive 0-based span.
        end_line: last line of the inclusive 0-based span.
        kinds: edge kinds to include in `outgoing` AND `incoming`.
            Defaults to `{EDGE_TYPE_REFERENCE}` — CONTAIN edges have no
            meaningful anchor and are excluded by default. Pass an
            explicit set (e.g. `{EDGE_TYPE_REFERENCE, EDGE_TYPE_CONTAIN}`)
            to opt in.
        depth: reserved for future multi-hop expansion. Only `depth=1`
            is supported in v1; raises NotImplementedError otherwise.
        layer: optional named graph layer (`reference`, `dependency`,
            `containment`, `all`, ...). Mutually exclusive with ``kinds``.

    Returns: RangeQueryResult with typed `NodeRef`/`EdgeRef` records.

    Anchor invariants:
        (i)   anchor_file == source.file (anchor lives at the call site,
              never the target).
        (ii)  anchor_* is meaningful only on EDGE_TYPE_REFERENCE; CONTAIN
              edges leave both fields None.
        (iii) outgoing ∩ incoming = ∅ for any single query (call site
              and target are in distinct scopes by definition).
    """
    if depth != 1:
        raise NotImplementedError(
            f"query_range supports only depth=1 in v1; got depth={depth}"
        )
    if layer is not None:
        if kinds is not None:
            raise ValueError("query_range accepts either kinds or layer, not both")
        layer_kinds = edge_types_for_graph_layer(layer)
        if layer_kinds is None:
            kinds = {edge.attributes().get("type") for edge in self.graph.es}
        else:
            kinds = set(layer_kinds)
    if kinds is None:
        kinds = {EDGE_TYPE_REFERENCE}

    # defined: brute-force overlap on per-file node list.
    nodes = self._file_nodes.get(file, [])
    defined_vids = [vid for s, e, vid in nodes if s <= end_line and e >= start_line]
    defined = [self._build_node_ref(vid) for vid in defined_vids]

    # outgoing: bisect on sorted (anchor_line, eid) list, then filter by kind.
    # Use bisect_left for the upper bound: bisect_right would include
    # `(end_line+1, 0)` (eid==0 entries at the line just past the query),
    # leaking an off-anchor edge into the slice.
    arr = self._file_edge_anchors.get(file, [])
    lo = bisect.bisect_left(arr, (start_line, -1))
    hi = bisect.bisect_left(arr, (end_line + 1, 0))
    outgoing: List[EdgeRef] = []
    for _, eid in arr[lo:hi]:
        edge = self.graph.es[eid]
        if edge["type"] in kinds:
            outgoing.append(self._build_edge_ref(eid))

    # incoming: walk igraph adjacency directly; carry target_vid so the
    # consumer can map the inbound edge back to which defined node it hit.
    # Self-edges (recursion) are emitted by `outgoing` already, so skip
    # them here to keep outgoing ∩ incoming = ∅ (invariant iii).
    incoming: List[EdgeRef] = []
    for vid in defined_vids:
        for eid in self.graph.incident(vid, mode="in"):
            edge = self.graph.es[eid]
            if edge.source == edge.target:
                continue
            if edge["type"] in kinds:
                incoming.append(self._build_edge_ref(eid))

    return RangeQueryResult(defined=defined, outgoing=outgoing, incoming=incoming)

query_range_by_symbol

query_range_by_symbol(
    name: str, kinds: set | None = None, depth: int = 1, *, layer: str | None = None
) -> RangeQueryResult

Range-query by symbol identity (name).

Resolves name -> vid -> (file, start_line, end_line) and forwards to query_range. name is the globally-unique identity attribute (semi-raw SCIP symbol), not the unified_name display.

Returns an empty RangeQueryResult if the symbol is unknown or has no associated file/range (e.g. a SCIP reference-only vertex).

Source code in codenib/graph/code_graph.py

def query_range_by_symbol(
    self,
    name: str,
    kinds: Optional[set] = None,
    depth: int = 1,
    *,
    layer: Optional[str] = None,
) -> RangeQueryResult:
    """Range-query by symbol identity (`name`).

    Resolves `name` -> vid -> `(file, start_line, end_line)` and forwards
    to `query_range`. `name` is the globally-unique identity attribute
    (semi-raw SCIP symbol), not the `unified_name` display.

    Returns an empty RangeQueryResult if the symbol is unknown or has
    no associated file/range (e.g. a SCIP reference-only vertex).
    """
    vid = self.name_to_vertex.get(name)
    if vid is None:
        return RangeQueryResult()
    attrs = self.graph.vs[vid].attributes()
    f = attrs.get("file")
    s = attrs.get("start_line")
    e = attrs.get("end_line")
    if f is None or s is None or e is None:
        return RangeQueryResult()
    return self.query_range(f, s, e, kinds=kinds, depth=depth, layer=layer)

layers

layers(*, use_core: bool = True)

Build a multi-graph layer index over this graph.

Source code in codenib/graph/code_graph.py

def layers(self, *, use_core: bool = True):
    """Build a multi-graph layer index over this graph."""
    from .layers import build_graph_layers

    return build_graph_layers(self, use_core=use_core)

layer

layer(name: str = GRAPH_LAYER_REFERENCE, *, use_core: bool = True)

Return one named graph layer view.

Source code in codenib/graph/code_graph.py

def layer(self, name: str = GRAPH_LAYER_REFERENCE, *, use_core: bool = True):
    """Return one named graph layer view."""
    return self.layers(use_core=use_core).get(name)

add_root_node

add_root_node(project_root)

Add the root node to the graph

Source code in codenib/graph/code_graph.py

def add_root_node(self, project_root):
    """Add the root node to the graph"""
    self._add_vertex(project_root, {"type": "root"})

add_directory_node

add_directory_node(dir_path)

Add a directory node to the graph

Source code in codenib/graph/code_graph.py

def add_directory_node(self, dir_path):
    """Add a directory node to the graph"""
    self._add_vertex(dir_path, {"type": NODE_TYPE_DIRECTORY})

save_graph

save_graph(output_path)

Save the graph to a pickle file for fast serialization.

Parameters:

Name	Type	Description	Default
output_path		Path to save the pickle file	required

Source code in codenib/graph/code_graph.py

def save_graph(self, output_path):
    """
    Save the graph to a pickle file for fast serialization.

    Args:
        output_path: Path to save the pickle file
    """
    # Prepare data for pickling
    data = {
        "schema_version": _SCHEMA_VERSION,
        "project_root": str(self.project_root) if self.project_root else None,
        "graph": self.graph,  # igraph objects are picklable
        "symbol_ranges": self.symbol_ranges,
        "name_to_vertex": self.name_to_vertex,
        "file_nodes": self._file_nodes,
        "file_edge_anchors": self._file_edge_anchors,
        "unified_to_names": self._unified_to_names,
    }

    with open(output_path, "wb") as f:
        pickle.dump(data, f, protocol=pickle.HIGHEST_PROTOCOL)

load_graph classmethod

load_graph(input_path)

Load a graph from a pickle file.

Parameters:

Name	Type	Description	Default
input_path		Path to the pickle file	required

Returns:

Name	Type	Description
CodeGraph		Loaded graph instance

Source code in codenib/graph/code_graph.py

@classmethod
def load_graph(cls, input_path):
    """
    Load a graph from a pickle file.

    Args:
        input_path: Path to the pickle file

    Returns:
        CodeGraph: Loaded graph instance
    """
    with open(input_path, "rb") as f:
        return cls.load_graph_stream(f, input_label=str(input_path))

load_graph_stream classmethod

load_graph_stream(handle, *, input_label='<authenticated stream>')

Load a graph from a caller-authenticated binary stream.

The serialized schema is unchanged. Runtime loaders use this entry point after binding the bytes to the manifest receipt, so they never need to reopen an already-verified pathname before unpickling it.

Source code in codenib/graph/code_graph.py

@classmethod
def load_graph_stream(cls, handle, *, input_label="<authenticated stream>"):
    """Load a graph from a caller-authenticated binary stream.

    The serialized schema is unchanged.  Runtime loaders use this entry
    point after binding the bytes to the manifest receipt, so they never
    need to reopen an already-verified pathname before unpickling it.
    """

    data = compat_pickle.load(handle)

    on_disk = data.get("schema_version")
    if on_disk != _SCHEMA_VERSION:
        raise ValueError(
            f"graph.pkl at {input_label} has schema_version={on_disk!r}, "
            f"expected {_SCHEMA_VERSION}. Rebuild the repository graph "
            "with `codenib index <repo> --preset graph --rebuild`."
        )

    # Create new CodeGraph instance
    graph_instance = cls(project_root=data.get("project_root"))

    # Restore the igraph object and internal state
    graph_instance.graph = data["graph"]
    graph_instance.symbol_ranges = data.get("symbol_ranges", {})
    graph_instance.name_to_vertex = data.get("name_to_vertex", {})
    graph_instance._file_nodes = data.get("file_nodes", {})
    graph_instance._file_edge_anchors = data.get("file_edge_anchors", {})
    graph_instance._unified_to_names = data.get("unified_to_names", {})

    return graph_instance

get_graph

get_graph()

Get the igraph Graph object.

Returns:

Type	Description
	The igraph Graph instance

Source code in codenib/graph/code_graph.py

def get_graph(self):
    """
    Get the igraph Graph object.

    Returns:
        The igraph Graph instance
    """
    return self.graph

unified_index

unified_index()

Readable unified_name (full + bare) -> [canonical names], cached.

Rebuilds from vertex attributes when the persisted _unified_to_names is empty (older prebuilt graphs ship it empty even though every vertex carries a unified_name).

Source code in codenib/graph/code_graph.py

def unified_index(self):
    """Readable ``unified_name`` (full + bare) -> [canonical names], cached.

    Rebuilds from vertex attributes when the persisted ``_unified_to_names``
    is empty (older prebuilt graphs ship it empty even though every vertex
    carries a ``unified_name``).
    """
    cache = getattr(self, "_unified_index_cache", None)
    if cache is not None:
        return cache
    index = {}

    def _add(key, name):
        index.setdefault(key, []).append(name)

    pre = self._unified_to_names or {}
    if pre:
        for disp, names in pre.items():
            for nm in names:
                _add(disp, nm)
                _add(self._bare_symbol(disp), nm)
    else:
        for nm in self.name_to_vertex:
            info = self.get_node_info_by_name(nm) or {}
            u = info.get("unified_name")
            if u:
                _add(u, nm)
                _add(self._bare_symbol(u), nm)
    self._unified_index_cache = index
    return index

display_name

display_name(name)

Readable label for a canonical name (unified_name or itself).

Source code in codenib/graph/code_graph.py

def display_name(self, name):
    """Readable label for a canonical *name* (``unified_name`` or itself)."""
    info = self.get_node_info_by_name(name) or {}
    return info.get("unified_name") or name

resolve_symbol

resolve_symbol(symbol)

Map a readable/bare symbol to (canonical_name | None, candidates).

Tries the canonical name first, then the readable unified_name (full display, then bare token), then a substring match. None when ambiguous or missing; candidates are readable display names.

Source code in codenib/graph/code_graph.py

def resolve_symbol(self, symbol):
    """Map a readable/bare *symbol* to (canonical_name | None, candidates).

    Tries the canonical name first, then the readable ``unified_name``
    (full display, then bare token), then a substring match. ``None`` when
    ambiguous or missing; candidates are readable display names.
    """
    n2v = self.name_to_vertex or {}
    s = (symbol or "").strip().strip("`'\"")
    if not s:
        return None, []
    if s in n2v:
        return s, [self.display_name(s)]

    uni = self.unified_index()
    cands = []
    if uni:
        sbase = self._bare_symbol(s)
        for key in (s, s + "()", sbase):
            if key in uni:
                cands = list(dict.fromkeys(uni[key]))
                break
        if not cands:
            sub = []
            for disp, names in uni.items():
                if sbase and sbase.lower() in disp.lower():
                    sub += names
            cands = list(dict.fromkeys(sub))
    if not cands:  # languages whose canonical name is already readable
        base = s.split(".")[-1].split(":")[-1]
        cands = [k for k in n2v if k.endswith("." + s) or k.split(".")[-1] == base]
    canonical = cands[0] if len(cands) == 1 else None
    return canonical, [self.display_name(c) for c in cands[:8]]

get_node_info_by_name

get_node_info_by_name(node_name)

Get information about a node in the graph.

Parameters:

Name	Type	Description	Default
node_name		Name of the node	required

Returns:

Type	Description
	Dictionary with vertex attributes or None if not found

Source code in codenib/graph/code_graph.py

def get_node_info_by_name(self, node_name):
    """
    Get information about a node in the graph.

    Args:
        node_name: Name of the node

    Returns:
        Dictionary with vertex attributes or None if not found
    """
    if isinstance(node_name, str):
        vertex = self.name_to_vertex.get(node_name)
        if vertex is None:
            return None

    return self.graph.vs[vertex].attributes()

get_node_info_by_id

get_node_info_by_id(node_id)

Get information about a node in the graph.

Parameters:

Name	Type	Description	Default
node_id		ID of the node	required

Returns:

Type	Description
	Dictionary with vertex attributes or None if not found

Source code in codenib/graph/code_graph.py

def get_node_info_by_id(self, node_id):
    """
    Get information about a node in the graph.

    Args:
        node_id: ID of the node

    Returns:
        Dictionary with vertex attributes or None if not found
    """
    if isinstance(node_id, int):
        vertex = self.graph.vs[node_id]
        return vertex.attributes()

    return None

get_node_content

get_node_content(node_id)

Get the content of a node in the graph. Read the file from start_line to end_line. If the node is a file, return the file content.

Parameters:

Name	Type	Description	Default
node_id		ID of the node	required

Returns:

Type	Description
	Content of the node or None if not found

Source code in codenib/graph/code_graph.py

def get_node_content(self, node_id):
    """
    Get the content of a node in the graph.
    Read the file from start_line to end_line.
    If the node is a file, return the file content.

    Args:
        node_id: ID of the node

    Returns:
        Content of the node or None if not found
    """
    node = self.graph.vs[node_id]
    node_type = node["type"] if "type" in node.attributes() else "unknown"
    if node_type == NODE_TYPE_FILE:
        # file path need to add self.project_root
        file_path = node["name"]
        if self.project_root:
            file_path = f"{self.project_root}/{file_path}"
        try:
            with open(file_path, "r") as f:
                content = f.read()
            return content
        except FileNotFoundError:
            print(f"File not found: {file_path}")
            return None
    elif is_symbol_node(node_type) and node_has_definition(node.attributes()):
        # Graph source ranges are 0-based and inclusive.
        start_line = node["start_line"]
        end_line = node["end_line"]
        if (
            not isinstance(start_line, int)
            or isinstance(start_line, bool)
            or not isinstance(end_line, int)
            or isinstance(end_line, bool)
            or start_line < 0
            or end_line < start_line
        ):
            return None

        # Get the file path
        file_path = node["file"] if "file" in node.attributes() else None
        if self.project_root and file_path:
            file_path = f"{self.project_root}/{file_path}"

        # Read the file content
        if file_path:
            try:
                with open(file_path, "r") as f:
                    lines = f.readlines()
                return "".join(lines[start_line : end_line + 1])
            except FileNotFoundError:
                print(f"File not found: {file_path}")
                return None
    return None

get_neighbors

get_neighbors(node_name)

Get the neighbors of a node in the graph.

Parameters:

Name	Type	Description	Default
node_name		Name of the node	required

Returns:

Type	Description
	List of neighbor vertex IDs

Source code in codenib/graph/code_graph.py

def get_neighbors(self, node_name):
    """
    Get the neighbors of a node in the graph.

    Args:
        node_name: Name of the node

    Returns:
        List of neighbor vertex IDs
    """
    if isinstance(node_name, str):
        vertex = self.name_to_vertex.get(node_name)
        if vertex is None:
            return []

    return self.graph.neighbors(vertex)

get_successors

get_successors(node_name, edge_types=None)

Get the successors of a node_name (outgoing edges).

Parameters:

Name	Type	Description	Default
node_name		Name of the node	required
edge_types		Optional edge types to retain. Typed traversal returns unique neighbors even when the graph contains multi-edges.	None

Returns:

Type	Description
	List of successor vertex IDs

Source code in codenib/graph/code_graph.py

def get_successors(self, node_name, edge_types=None):
    """
    Get the successors of a node_name (outgoing edges).

    Args:
        node_name: Name of the node
        edge_types: Optional edge types to retain. Typed traversal returns
            unique neighbors even when the graph contains multi-edges.

    Returns:
        List of successor vertex IDs
    """
    if isinstance(node_name, str):
        vertex = self.name_to_vertex.get(node_name)
        if vertex is None:
            return []

    if edge_types is None:
        return self.graph.successors(vertex)
    return self._get_typed_adjacent(vertex, edge_types, mode="out")

get_predecessors

get_predecessors(node_name, edge_types=None)

Get the predecessors of a node_name (incoming edges).

Parameters:

Name	Type	Description	Default
node_name		Name of the node	required
edge_types		Optional edge types to retain. Typed traversal returns unique neighbors even when the graph contains multi-edges.	None

Returns:

Type	Description
	List of predecessor vertex IDs

Source code in codenib/graph/code_graph.py

def get_predecessors(self, node_name, edge_types=None):
    """
    Get the predecessors of a node_name (incoming edges).

    Args:
        node_name: Name of the node
        edge_types: Optional edge types to retain. Typed traversal returns
            unique neighbors even when the graph contains multi-edges.

    Returns:
        List of predecessor vertex IDs
    """
    if isinstance(node_name, str):
        vertex = self.name_to_vertex.get(node_name)
        if vertex is None:
            return []

    if edge_types is None:
        return self.graph.predecessors(vertex)
    return self._get_typed_adjacent(vertex, edge_types, mode="in")

print_graph_basic_info

print_graph_basic_info()

Print basic information about the graph.

Source code in codenib/graph/code_graph.py

def print_graph_basic_info(self):
    """
    Print basic information about the graph.
    """
    print("Graph Summary:")
    print(f"  Number of vertices: {self.graph.vcount()}")
    print(f"  Number of edges: {self.graph.ecount()}")
    print(f"  Directed: {self.graph.is_directed()}")

    # Print vertex types
    vertex_types = self.graph.vs["type"]
    unique_vertex_types = set(vertex_types)
    print(f"  Unique vertex types: {unique_vertex_types}")

    # Print edge types
    edge_types = self.graph.es["type"]
    unique_edge_types = set(edge_types)
    print(f"  Unique edge types: {unique_edge_types}")

visualize_graph

visualize_graph(output_path=None, width=800, height=600, layout='fruchterman_reingold')

Visualize the code graph with different colors for different node and edge types.

Parameters:

Name	Type	Description	Default
output_path		Path to save the visualization (optional)	None
width		Width of the plot in pixels	800
height		Height of the plot in pixels	600
layout		Layout algorithm to use ('fruchterman_reingold', 'kk', 'grid', etc.)	'fruchterman_reingold'

Returns:

Type	Description
	A matplotlib figure object if output_path is not provided

Source code in codenib/graph/code_graph.py

def visualize_graph(
    self, output_path=None, width=800, height=600, layout="fruchterman_reingold"
):
    """
    Visualize the code graph with different colors for different node and edge types.

    Args:
        output_path: Path to save the visualization (optional)
        width: Width of the plot in pixels
        height: Height of the plot in pixels
        layout: Layout algorithm to use ('fruchterman_reingold', 'kk', 'grid', etc.)

    Returns:
        A matplotlib figure object if output_path is not provided
    """
    if self.graph.vcount() == 0:
        print("Graph is empty. Nothing to visualize.")
        return None

    try:
        import matplotlib.pyplot as plt
    except ImportError as exc:
        raise RuntimeError(
            "static graph rendering requires matplotlib; install it separately"
        ) from exc

    # Define color schemes
    node_type_colors = {
        NODE_TYPE_FILE: "skyblue",
        NODE_TYPE_SYMBOL: "lightgreen",
        NODE_TYPE_CLASS: "gold",
        NODE_TYPE_FUNCTION: "lightgreen",
        NODE_TYPE_METHOD: "lightcoral",
        NODE_TYPE_FIELD: "plum",
        NODE_TYPE_DIRECTORY: "orange",  # Add directory node color
        "root": "lightgrey",  # Root node color
        # Add more node types and colors as needed
    }

    edge_type_colors = {
        EDGE_TYPE_REFERENCE: "red",
        EDGE_TYPE_CONTAIN: "blue",
        # Add more edge types and colors as needed
    }

    # Define visual style
    visual_style = {}

    # Set vertex colors based on type
    vertex_colors = []
    for vertex in self.graph.vs:
        node_type = vertex["type"] if "type" in vertex.attributes() else "unknown"
        vertex_colors.append(node_type_colors.get(node_type, "grey"))
    visual_style["vertex_color"] = vertex_colors

    # Set vertex labels
    visual_style["vertex_label"] = [
        v["name"].split("/")[-1] if "/" in v["name"] else v["name"]
        for v in self.graph.vs
    ]
    visual_style["vertex_label_size"] = 8

    # Set vertex sizes (files can be larger than symbols)
    vertex_sizes = []
    for vertex in self.graph.vs:
        if "type" in vertex.attributes() and vertex["type"] == NODE_TYPE_FILE:
            vertex_sizes.append(20)
        else:
            vertex_sizes.append(10)
    visual_style["vertex_size"] = vertex_sizes

    # Set edge colors based on type
    edge_colors = []
    for edge in self.graph.es:
        edge_type = edge["type"] if "type" in edge.attributes() else "unknown"
        edge_colors.append(edge_type_colors.get(edge_type, "grey"))
    visual_style["edge_color"] = edge_colors

    # Set edge width
    visual_style["edge_width"] = 1.0

    # Calculate layout
    if layout == "fruchterman_reingold":
        visual_style["layout"] = self.graph.layout_fruchterman_reingold()
    elif layout == "kk":
        visual_style["layout"] = self.graph.layout_kamada_kawai()
    elif layout == "grid":
        visual_style["layout"] = self.graph.layout_grid()
    else:
        visual_style["layout"] = self.graph.layout(layout)

    # Adjust dimensions
    visual_style["bbox"] = (width, height)
    visual_style["margin"] = 40

    # Create legend data
    legend_data = {
        "Node Types": [
            (color, node_type) for node_type, color in node_type_colors.items()
        ],
        "Edge Types": [
            (color, edge_type) for edge_type, color in edge_type_colors.items()
        ],
    }

    # Create the plot
    fig, ax = plt.subplots(figsize=(width / 100, height / 100))

    # Plot the graph
    ig.plot(self.graph, target=ax, **visual_style)

    # Add legend for node types
    node_legend_patches = []
    for color, label in legend_data["Node Types"]:
        node_legend_patches.append(
            plt.Line2D(
                [0],
                [0],
                marker="o",
                color="w",
                markerfacecolor=color,
                markersize=10,
                label=label,
            )
        )

    # Add legend for edge types
    edge_legend_patches = []
    for color, label in legend_data["Edge Types"]:
        edge_legend_patches.append(
            plt.Line2D([0], [0], color=color, lw=2, label=label)
        )

    # Add legends to plot
    ax.legend(
        handles=node_legend_patches + edge_legend_patches,
        loc="upper right",
        title="Legend",
        frameon=True,
    )

    # Save or show the plot
    if output_path:
        plt.savefig(output_path, dpi=100, bbox_inches="tight")
        print(f"Graph visualization saved to {output_path}")
        plt.close(fig)
        return None
    else:
        plt.tight_layout()
        return fig

CodeVectorStore

CodeVectorStore(
    embedding_model: str = "text-embedding-ada-002",
    embedding_provider: str = "openai",
    dimension: int = 1536,
    index_type: str = "flat",
    index_metric: str = "ip",
    ivf_nlist: int = 100,
    ivf_nprobe: int = 8,
    store_path: str | None = None,
    profiler: Profiler | None = None,
    embedding: Any | None = None,
    artifact_metadata: dict[str, Any] | None = None,
    **embedding_kwargs
)

Vector store for code embeddings using FAISS and sentence-transformers. Provides semantic search capabilities over code chunks.

Supports hierarchical indexing: - L0: File-level skeletons - L2: Function/method-level chunks for fine-grained retrieval (default)

Parameters:

Name	Type	Description	Default
embedding_model	str	Name of the embedding model to use	'text-embedding-ada-002'
embedding_provider	str	Provider for embeddings ("openai" or "huggingface")	'openai'
dimension	int	Dimension of the embedding vectors	1536
index_type	str	FAISS index type — "flat" (exact brute force, default) or "ivf" (IVF inverted-file; approximate, faster at scale). IVF indices are trained lazily on the first batch of vectors.	'flat'
index_metric	str	Distance metric ("ip" for inner product, "l2" for L2 distance)	'ip'
ivf_nlist	int	IVF only — number of Voronoi cells (coarse centroids). On small corpora it is clamped down to the training-set size, since FAISS k-means needs at least nlist training points.	100
ivf_nprobe	int	IVF only — cells probed per query; the recall/latency knob. Clamped to the effective nlist.	8
store_path	str | None	Path to store/load the vector store	None
profiler	Profiler | None	Optional profiler instance to capture detailed timings	None
embedding	Any | None	A pre-built embedding wrapper to reuse. When several stores share one model (e.g. one per repo), pass the same instance so the model is loaded onto the GPU only once.	None
artifact_metadata	dict[str, Any] | None	Optional immutable source/build identity persisted with the top-level configuration.	None
**embedding_kwargs		Additional arguments for embedding model	{}

Methods:

Name	Description
reuse_query_embedding	Reuse one query vector within a composed request, then discard it.
clear_query_cache	Clear the single-query embedding reused by consecutive search stages.
swap_index	Hot-swap the FAISS index without reloading the embedding model.
close	Release embeddings and FAISS resources to free memory.
add_code_chunks	Add code chunks to the vector store.
add_nodes_with_content	Add NodeInfo objects (with content) to the vector store.
search	Search for similar code chunks using semantic similarity.
search_with_content	Search and return results with content included.
search_within_ids	Search only within a restricted set of node IDs.
hierarchical_search	Note: This method is implemented by Claude and is just for future reference.
save	Save the vector store to disk.
load	Load the vector store from disk.
get_stats	Get statistics about the vector store.
get_embeddings_by_content_hash	Extract raw embedding vectors from the FAISS index, keyed by content hash.
rebuild_from_embeddings	Clear level and rebuild its FAISS index from pre-computed embeddings.
delta_update	Patch a flat FAISS index in place when the change set is small.
clear	Clear data from the vector store.

Attributes:

Name	Type	Description
closed	bool	Whether every resource owned by this store has been released.

Source code in codenib/index/embedding/vector_store.py

def __init__(
    self,
    embedding_model: str = "text-embedding-ada-002",
    embedding_provider: str = "openai",
    dimension: int = 1536,
    index_type: str = "flat",
    index_metric: str = "ip",
    ivf_nlist: int = 100,
    ivf_nprobe: int = 8,
    store_path: Optional[str] = None,
    profiler: Optional[Profiler] = None,
    embedding: Optional[Any] = None,
    artifact_metadata: Optional[Dict[str, Any]] = None,
    **embedding_kwargs,
):
    """
    Initialize the CodeVectorStore.

    Args:
        embedding_model: Name of the embedding model to use
        embedding_provider: Provider for embeddings ("openai" or "huggingface")
        dimension: Dimension of the embedding vectors
        index_type: FAISS index type — "flat" (exact brute force, default)
            or "ivf" (IVF inverted-file; approximate, faster at scale).
            IVF indices are trained lazily on the first batch of vectors.
        index_metric: Distance metric ("ip" for inner product, "l2" for L2 distance)
        ivf_nlist: IVF only — number of Voronoi cells (coarse centroids). On
            small corpora it is clamped down to the training-set size, since
            FAISS k-means needs at least ``nlist`` training points.
        ivf_nprobe: IVF only — cells probed per query; the recall/latency
            knob. Clamped to the effective ``nlist``.
        store_path: Path to store/load the vector store
        profiler: Optional profiler instance to capture detailed timings
        embedding: A pre-built embedding wrapper to reuse. When several
            stores share one model (e.g. one per repo), pass the same
            instance so the model is loaded onto the GPU only once.
        artifact_metadata: Optional immutable source/build identity persisted
            with the top-level configuration.
        **embedding_kwargs: Additional arguments for embedding model
    """
    self.embedding_model = embedding_model
    self.embedding_provider = _validate_provider_options(
        embedding_provider,
        embedding_kwargs,
    )
    self.embedding_load_policy = (
        resolve_embedding_load_policy_from_options(
            embedding_model,
            embedding_kwargs,
        )
        if self.embedding_provider == "huggingface"
        else None
    )
    self.embedding_revision = (
        self.embedding_load_policy.revision if self.embedding_load_policy else None
    )
    self.embedding_trust_remote_code = bool(
        self.embedding_load_policy and self.embedding_load_policy.trust_remote_code
    )
    self.dimension = dimension
    self.index_type = index_type.lower()
    if self.index_type not in ("flat", "ivf"):
        raise ValueError(
            f"Unsupported index_type: {index_type}. Must be 'flat' or 'ivf'."
        )
    self.index_metric = index_metric.lower()
    if self.index_metric not in ["ip", "l2"]:
        raise ValueError(
            f"Unsupported index_metric: {index_metric}. Must be 'ip' or 'l2'."
        )
    self.ivf_nlist = max(1, int(ivf_nlist))
    self.ivf_nprobe = max(1, int(ivf_nprobe))
    initial_store_path = Path(store_path) if store_path else None
    self.profiler = profiler
    initial_artifact_metadata = dict(artifact_metadata or {})

    # Initialize the embedding model — or reuse a shared one so the same
    # model isn't loaded onto the GPU once per store.
    self.embedding = (
        embedding
        if embedding is not None
        else self._initialize_embedding_model(**embedding_kwargs)
    )
    self._cached_query_text: Optional[str] = None
    self._cached_query_vector: Optional[np.ndarray] = None
    self._query_cache_depth = 0
    self.dimension = self._infer_embedding_dimension(dimension)

    # Initialize L0 (file-level skeletons)
    l0_index = self._build_faiss_index()

    # Initialize L2 (function/method-level) - default
    l2_index = self._build_faiss_index()
    self._loaded_state = _LoadedVectorState(
        l0_index=l0_index,
        l0_documents=[],
        l2_index=l2_index,
        l2_documents=[],
        artifact_metadata=initial_artifact_metadata,
        store_path=initial_store_path,
    )

    logger.info(
        f"Initialized CodeVectorStore with {embedding_provider}:{embedding_model}"
    )

closed property

closed: bool

Whether every resource owned by this store has been released.

reuse_query_embedding

reuse_query_embedding()

Reuse one query vector within a composed request, then discard it.

Source code in codenib/index/embedding/vector_store.py

@contextmanager
def reuse_query_embedding(self):
    """Reuse one query vector within a composed request, then discard it."""

    depth = getattr(self, "_query_cache_depth", 0)
    if depth == 0:
        self.clear_query_cache()
    self._query_cache_depth = depth + 1
    try:
        yield
    finally:
        self._query_cache_depth -= 1
        if self._query_cache_depth == 0:
            self.clear_query_cache()

clear_query_cache

clear_query_cache() -> None

Clear the single-query embedding reused by consecutive search stages.

Source code in codenib/index/embedding/vector_store.py

def clear_query_cache(self) -> None:
    """Clear the single-query embedding reused by consecutive search stages."""

    self._cached_query_text = None
    self._cached_query_vector = None

swap_index

swap_index(
    path: str, *, native_index_authorization: NativeIndexAuthorization | None = None
) -> None

Hot-swap the FAISS index without reloading the embedding model.

The replacement is fully loaded and validated before the current L0/L2 state is released. The embedding model is left intact so the caller can reuse the same model across many instances.

Source code in codenib/index/embedding/vector_store.py

def swap_index(
    self,
    path: str,
    *,
    native_index_authorization: NativeIndexAuthorization | None = None,
) -> None:
    """Hot-swap the FAISS index without reloading the embedding model.

    The replacement is fully loaded and validated before the current
    L0/L2 state is released. The embedding model is left intact so the
    caller can reuse the same model across many instances.
    """
    self.load(
        path,
        native_index_authorization=native_index_authorization,
    )

close

close() -> None

Release embeddings and FAISS resources to free memory.

Source code in codenib/index/embedding/vector_store.py

def close(self) -> None:
    """Release embeddings and FAISS resources to free memory."""

    state = self._loaded_state
    released: dict[int, bool] = {}
    deferred: BaseException | None = None
    for index in (state.l0_index, state.l2_index):
        if index is None or id(index) in released:
            continue
        reset = getattr(index, "reset", None)
        if not callable(reset):
            released[id(index)] = True
            continue
        try:
            reset()
            released[id(index)] = True
        except BaseException as exc:  # noqa: B036 - visit both native indices
            released[id(index)] = False
            if deferred is None:
                deferred = exc
            else:
                _annotate_secondary_error(
                    deferred,
                    "additional FAISS index cleanup also failed",
                    exc,
                )

    l0_released = state.l0_index is None or released.get(id(state.l0_index), False)
    l2_released = state.l2_index is None or released.get(id(state.l2_index), False)
    self._loaded_state = _LoadedVectorState(
        l0_index=None if l0_released else state.l0_index,
        l0_documents=[] if l0_released else state.l0_documents,
        l2_index=None if l2_released else state.l2_index,
        l2_documents=[] if l2_released else state.l2_documents,
        artifact_metadata=state.artifact_metadata,
        store_path=state.store_path,
    )
    if deferred is not None:
        raise deferred

    self.embedding = None
    self._query_cache_depth = 0
    self._cached_query_text = None
    self._cached_query_vector = None

    try:
        import torch

        if torch.cuda.is_available():
            torch.cuda.empty_cache()
    except Exception:
        pass

add_code_chunks

add_code_chunks(code_chunks: list[dict[str, Any]], level: Level = 'l2') -> None

Add code chunks to the vector store.

Parameters:

Name	Type	Description	Default
code_chunks	list[dict[str, Any]]	List of code chunk dictionaries with content and metadata	required
level	Level	Index level to add chunks to ("l0" for file skeletons, "l2" for functions/methods)	'l2'

Source code in codenib/index/embedding/vector_store.py

def add_code_chunks(
    self, code_chunks: List[Dict[str, Any]], level: Level = "l2"
) -> None:
    """
    Add code chunks to the vector store.

    Args:
        code_chunks: List of code chunk dictionaries with content and metadata
        level: Index level to add chunks to
            ("l0" for file skeletons, "l2" for functions/methods)
    """
    if not code_chunks:
        logger.warning("No code chunks provided")
        return

    index, documents_list = self._get_index_and_docs(level)
    logger.info(f"Adding {len(code_chunks)} code chunks to {level} vector store")

    # Convert chunks to _Document objects
    documents: List[_Document] = []
    for i, chunk in enumerate(code_chunks):
        content = chunk.get("content", "")
        content_hash = hashlib.md5(
            content.encode("utf-8", errors="replace")
        ).hexdigest()
        metadata = {
            "chunk_id": len(documents_list) + i,
            "chunk_type": chunk.get("chunk_type", "unknown"),
            "name": chunk.get("name", f"chunk_{i}"),
            "file": chunk.get("file", ""),
            "start_line": chunk.get("start_line", 0),
            "end_line": chunk.get("end_line", 0),
            "node_id": chunk.get("node_id", ""),
            "level": level,
            "content_hash": content_hash,
        }
        for key, value in chunk.items():
            if key not in ["content"] and key not in metadata:
                metadata[key] = value

        documents.append(_Document(page_content=content, metadata=metadata))

    # Store documents
    documents_list.extend(documents)

    texts = [doc.page_content for doc in documents]

    # Phase 1: Embed texts (typically the bottleneck)
    with self._profile_section(
        f"embedding_encode_{level}",
        {"num_documents": len(documents), "level": level},
    ):
        embeddings = self.embedding.embed_documents(texts)

    # Phase 2: Add pre-computed vectors to FAISS index
    with self._profile_section(
        f"faiss_index_add_{level}",
        {"num_vectors": len(embeddings), "level": level},
    ):
        vectors = np.array(embeddings, dtype=np.float32)
        self._add_to_index(level, vectors)

    logger.info(
        f"Successfully added {len(documents)} documents to {level} vector store"
    )

add_nodes_with_content

add_nodes_with_content(nodes: list[NodeInfo], level: Level = 'l2') -> None

Add NodeInfo objects (with content) to the vector store.

Parameters:

Name	Type	Description	Default
nodes	list[NodeInfo]	List of NodeInfo objects	required
level	Level	Index level to add nodes to ("l0" or "l2")	'l2'

Source code in codenib/index/embedding/vector_store.py

def add_nodes_with_content(
    self, nodes: List[NodeInfo], level: Level = "l2"
) -> None:
    """
    Add NodeInfo objects (with content) to the vector store.

    Args:
        nodes: List of NodeInfo objects
        level: Index level to add nodes to ("l0" or "l2")
    """
    chunks = []
    for node in nodes:
        chunk = {
            "content": node.content,
            "chunk_type": node.type,
            "name": node.node_name,
            "file": node.file,
            "start_line": node.start_line,
            "end_line": node.end_line,
        }
        chunks.append(chunk)

    self.add_code_chunks(chunks, level=level)

search

search(
    query: str,
    top_k: int = 10,
    score_threshold: float | None = None,
    level: Level = "l2",
    mask_node_ids: set[str] | None = None,
) -> list[NodeInfo]

Search for similar code chunks using semantic similarity.

Parameters:

Name	Type	Description	Default
query	str	Search query text	required
top_k	int	Number of top results to return	10
score_threshold	float | None	Minimum similarity score threshold	None
level	Level	Index level to search ("l0" for file skeletons, "l2" for functions/methods)	'l2'
mask_node_ids	set[str] | None	Optional set of CodeChunk.node_id values to filter results.	None

Returns:

Type	Description
list[NodeInfo]	List of NodeInfo objects with scores populated

Source code in codenib/index/embedding/vector_store.py

def search(
    self,
    query: str,
    top_k: int = 10,
    score_threshold: Optional[float] = None,
    level: Level = "l2",
    mask_node_ids: Optional[Set[str]] = None,
) -> List[NodeInfo]:
    """
    Search for similar code chunks using semantic similarity.

    Args:
        query: Search query text
        top_k: Number of top results to return
        score_threshold: Minimum similarity score threshold
        level: Index level to search ("l0" for file skeletons, "l2" for
            functions/methods)
        mask_node_ids: Optional set of CodeChunk.node_id values to filter results.

    Returns:
        List of NodeInfo objects with scores populated
    """
    index, documents = self._get_index_and_docs(level)

    if index is None or index.ntotal == 0:
        logger.warning(f"No {level} vector store available. Add code chunks first.")
        return []

    logger.debug(f"Searching {level} for: {query[:100]}...")

    docs_with_scores = self._search_index(query, index, documents, top_k)

    results = []
    for doc, score in docs_with_scores:
        metadata = doc.metadata

        if score_threshold is not None and self._should_filter_by_threshold(
            score, score_threshold
        ):
            continue

        node_with_score = NodeInfo(
            node_name=metadata.get("name", "unknown"),
            type=metadata.get("chunk_type", "unknown"),
            file=_result_source_file(metadata),
            node_id=metadata.get("node_id", ""),
            start_line=metadata.get("start_line", 0),
            end_line=metadata.get("end_line", 0),
            score=float(score),
        )
        results.append(node_with_score)

    if mask_node_ids:
        results = [r for r in results if r.node_id in mask_node_ids]
    if top_k:
        results = results[:top_k]

    logger.debug(
        f"Found {len(results)} results in {level} (masked={bool(mask_node_ids)})"
    )
    return results

search_with_content

search_with_content(
    query: str,
    top_k: int = 10,
    score_threshold: float | None = None,
    level: Level = "l2",
    mask_node_ids: set[str] | None = None,
) -> list[NodeInfo]

Search and return results with content included.

Parameters:

Name	Type	Description	Default
query	str	Search query text	required
top_k	int	Number of top results to return	10
score_threshold	float | None	Minimum similarity score threshold	None
level	Level	Index level to search ("l0" for file skeletons, "l2" for functions/methods)	'l2'
mask_node_ids	set[str] | None	Optional set of CodeChunk.node_id values to filter results.	None

Returns:

Type	Description
list[NodeInfo]	List of NodeInfo objects with content populated

Source code in codenib/index/embedding/vector_store.py

def search_with_content(
    self,
    query: str,
    top_k: int = 10,
    score_threshold: Optional[float] = None,
    level: Level = "l2",
    mask_node_ids: Optional[Set[str]] = None,
) -> List[NodeInfo]:
    """
    Search and return results with content included.

    Args:
        query: Search query text
        top_k: Number of top results to return
        score_threshold: Minimum similarity score threshold
        level: Index level to search ("l0" for file skeletons, "l2" for
            functions/methods)
        mask_node_ids: Optional set of CodeChunk.node_id values to filter results.

    Returns:
        List of NodeInfo objects with content populated
    """
    index, documents = self._get_index_and_docs(level)

    if index is None or index.ntotal == 0:
        logger.warning(f"No {level} vector store available. Add code chunks first.")
        return []

    docs_with_scores = self._search_index(query, index, documents, top_k)

    results = []
    for doc, score in docs_with_scores:
        metadata = doc.metadata

        if score_threshold is not None and self._should_filter_by_threshold(
            score, score_threshold
        ):
            continue

        node_with_content = NodeInfo(
            node_name=metadata.get("name", "unknown"),
            type=metadata.get("chunk_type", "unknown"),
            file=_result_source_file(metadata),
            node_id=metadata.get("node_id", ""),
            start_line=metadata.get("start_line", 0),
            end_line=metadata.get("end_line", 0),
            score=float(score),
            content=doc.page_content,
        )
        results.append(node_with_content)

    if mask_node_ids:
        results = [r for r in results if r.node_id in mask_node_ids]
    if top_k:
        results = results[:top_k]

    logger.debug(
        f"Found {len(results)} results with content in {level} "
        f"(masked={bool(mask_node_ids)})"
    )
    return results

search_within_ids

search_within_ids(
    query: str, mask_node_ids: set[str], top_k: int = 10, level: Level = "l2"
) -> list[NodeInfo]

Search only within a restricted set of node IDs.

Instead of searching the full FAISS index globally and filtering afterwards, this method restricts the search space before computing similarity. It reconstructs stored vectors for matching documents and computes similarity against the query embedding directly.

Parameters:

Name	Type	Description	Default
query	str	Search query text.	required
mask_node_ids	set[str]	Set of node_id / node_name values to restrict search to.	required
top_k	int	Number of top results to return.	10
level	Level	Index level to search.	'l2'

Returns:

Type	Description
list[NodeInfo]	List of NodeInfo objects sorted by similarity score.

Source code in codenib/index/embedding/vector_store.py

def search_within_ids(
    self,
    query: str,
    mask_node_ids: Set[str],
    top_k: int = 10,
    level: Level = "l2",
) -> List[NodeInfo]:
    """Search only within a restricted set of node IDs.

    Instead of searching the full FAISS index globally and filtering
    afterwards, this method restricts the search space *before* computing
    similarity.  It reconstructs stored vectors for matching documents
    and computes similarity against the query embedding directly.

    Args:
        query: Search query text.
        mask_node_ids: Set of node_id / node_name values to restrict
            search to.
        top_k: Number of top results to return.
        level: Index level to search.

    Returns:
        List of NodeInfo objects sorted by similarity score.
    """
    index, documents = self._get_index_and_docs(level)
    if index is None or index.ntotal == 0:
        logger.warning(f"No {level} vector store available.")
        return []

    # Find documents whose node_id or name is in mask set
    matched: list[tuple[int, _Document]] = []
    for i, doc in enumerate(documents):
        meta = doc.metadata
        if (
            meta.get("node_id", "") in mask_node_ids
            or meta.get("name", "") in mask_node_ids
        ):
            matched.append((i, doc))

    if not matched:
        logger.debug("search_within_ids: no matching documents found")
        return []

    # Encode query
    query_vec = self._embed_query(query)

    # Reconstruct stored vectors and compute similarity
    results: list[NodeInfo] = []
    for faiss_idx, doc in matched:
        vec = index.reconstruct(faiss_idx)
        if self.index_metric == "ip":
            score = float(np.dot(query_vec, vec))
        else:  # l2 — lower is better
            score = float(np.sum((query_vec - vec) ** 2))

        metadata = doc.metadata
        results.append(
            NodeInfo(
                node_name=metadata.get("name", "unknown"),
                type=metadata.get("chunk_type", "unknown"),
                file=_result_source_file(metadata),
                node_id=metadata.get("node_id", ""),
                start_line=metadata.get("start_line", 0),
                end_line=metadata.get("end_line", 0),
                score=score,
                content=doc.page_content,
            )
        )

    best_by_node = {}
    for result in results:
        identity = result.node_id or (
            result.file,
            result.node_name,
            result.start_line,
            result.end_line,
        )
        existing = best_by_node.get(identity)
        if existing is None:
            best_by_node[identity] = result
            continue
        if self.index_metric == "ip":
            is_better = result.score > existing.score
        else:
            is_better = result.score < existing.score
        if is_better:
            best_by_node[identity] = result
    results = list(best_by_node.values())

    # Sort: ip → higher is better; l2 → lower is better
    results.sort(
        key=lambda r: r.score,
        reverse=(self.index_metric == "ip"),
    )

    logger.debug(
        "search_within_ids: %d matched, %d unique, returning top %d",
        len(matched),
        len(results),
        min(top_k, len(results)),
    )
    return results[:top_k]

hierarchical_search

hierarchical_search(
    query: str,
    l0_top_k: int = 5,
    l2_top_k: int = 10,
    l0_score_threshold: float | None = None,
    l2_score_threshold: float | None = None,
    filter_l2_by_l0: bool = True,
) -> dict[str, list[NodeInfo]]

Note: This method is implemented by Claude and is just for future reference. Perform hierarchical search: first L0 (files), then L2 (functions).

This implements a coarse-to-fine retrieval strategy: 1. Search L0 to find relevant files based on their skeletons 2. Search L2 for specific functions/methods 3. Optionally filter L2 results to only include those from L0 files

Parameters:

Name	Type	Description	Default
query	str	Search query text	required
l0_top_k	int	Number of top L0 results (files)	5
l2_top_k	int	Number of top L2 results (functions/methods)	10
l0_score_threshold	float | None	Score threshold for L0 results	None
l2_score_threshold	float | None	Score threshold for L2 results	None
filter_l2_by_l0	bool	If True, only return L2 results from files found in L0	True

Returns:

Type	Description
dict[str, list[NodeInfo]]	Dict with 'l0' and 'l2' keys containing search results

Source code in codenib/index/embedding/vector_store.py

def hierarchical_search(
    self,
    query: str,
    l0_top_k: int = 5,
    l2_top_k: int = 10,
    l0_score_threshold: Optional[float] = None,
    l2_score_threshold: Optional[float] = None,
    filter_l2_by_l0: bool = True,
) -> Dict[str, List[NodeInfo]]:
    """
    Note: This method is implemented by Claude and is just for future reference.
    Perform hierarchical search: first L0 (files), then L2 (functions).

    This implements a coarse-to-fine retrieval strategy:
    1. Search L0 to find relevant files based on their skeletons
    2. Search L2 for specific functions/methods
    3. Optionally filter L2 results to only include those from L0 files

    Args:
        query: Search query text
        l0_top_k: Number of top L0 results (files)
        l2_top_k: Number of top L2 results (functions/methods)
        l0_score_threshold: Score threshold for L0 results
        l2_score_threshold: Score threshold for L2 results
        filter_l2_by_l0: If True, only return L2 results from files found in L0

    Returns:
        Dict with 'l0' and 'l2' keys containing search results
    """
    # Step 1: Search L0 to find relevant files
    l0_results = self.search(
        query, top_k=l0_top_k, score_threshold=l0_score_threshold, level="l0"
    )

    # Step 2: Search L2 for functions/methods (fetch more if filtering)
    l2_fetch_k = l2_top_k * 3 if filter_l2_by_l0 else l2_top_k
    l2_results = self.search(
        query, top_k=l2_fetch_k, score_threshold=l2_score_threshold, level="l2"
    )

    # Step 3: Optionally filter L2 results by L0 files
    if filter_l2_by_l0 and l0_results:
        l0_files = {result.file for result in l0_results}
        l2_results = [r for r in l2_results if r.file in l0_files]

    # Limit L2 results to top_k
    l2_results = l2_results[:l2_top_k]

    return {
        "l0": l0_results,
        "l2": l2_results,
    }

save

save(path: str | None = None) -> None

Save the vector store to disk.

Parameters:

Name	Type	Description	Default
path	str | None	Path to save the store (uses self.store_path if not provided)	None

Source code in codenib/index/embedding/vector_store.py

def save(self, path: Optional[str] = None) -> None:
    """
    Save the vector store to disk.

    Args:
        path: Path to save the store (uses self.store_path if not provided)
    """
    save_path = Path(path) if path else self.store_path
    if save_path is None:
        raise ValueError("No save path provided")

    save_path = Path(save_path)
    save_path.mkdir(parents=True, exist_ok=True)

    logger.info(f"Saving vector store to {save_path}")

    model_suffix = self.embedding_model.replace("/", "__")

    level_state = {}
    for level in ("l0", "l2"):
        index, documents = self._get_index_and_docs(level)
        if documents and (index is None or int(index.ntotal) != len(documents)):
            vector_count = 0 if index is None else int(index.ntotal)
            raise ValueError(
                f"Cannot save misaligned {level} vector store: "
                f"{vector_count} vectors for {len(documents)} documents"
            )
        if index is not None:
            self._validate_loaded_faiss_index(
                index,
                relative=f"{level}/index_{model_suffix}.faiss",
                document_count=len(documents),
            )
        level_state[level] = (index, documents)

    config_path = save_path / f"config_{model_suffix}.json"
    save_marker = save_path / f".{config_path.name}.save-in-progress"
    _atomic_json_dump(
        save_marker,
        {"persistence_schema": VECTOR_PERSISTENCE_SCHEMA},
    )
    try:
        level_artifacts: dict[str, dict[str, dict[str, Any]]] = {}
        for level, (_index, documents) in level_state.items():
            if documents:
                level_artifacts[level] = self._save_level(
                    save_path, level, model_suffix
                )
            else:
                self._remove_level_files(save_path, level, model_suffix)

        config = {
            "embedding_model": self.embedding_model,
            "embedding_provider": self.embedding_provider,
            "embedding_revision": self.embedding_revision,
            "dimension": self.dimension,
            "index_type": self.index_type,
            "index_metric": self.index_metric,
            "l0_documents": len(self.l0_documents),
            "l2_documents": len(self.l2_documents),
            "persistence_schema": VECTOR_PERSISTENCE_SCHEMA,
            "level_artifacts": level_artifacts,
        }
        if self.artifact_metadata:
            config["artifact"] = self.artifact_metadata
        if (
            type(self.artifact_metadata.get("builder_schema")) is int
            and self.artifact_metadata.get("builder_schema") == 8
        ):
            config["row_mapping"] = VECTOR_ROW_MAPPING_CONTRACT
        # This config is the commit record for both levels. Publishing it
        # last makes an interrupted multi-file save detectable on load.
        _atomic_json_dump(config_path, config)
    except Exception:
        # Keep the marker so a later load cannot accept a partially
        # replaced legacy artifact. A successful retry removes it.
        raise
    else:
        save_marker.unlink()

    logger.info("Vector store saved successfully")

load

load(
    path: str | None = None,
    *,
    native_index_authorization: NativeIndexAuthorization | None = None
) -> None

Load the vector store from disk.

Parameters:

Name	Type	Description	Default
path	str | None	Path to load the store from (uses self.store_path if not provided)	None
native_index_authorization	NativeIndexAuthorization | None	Process-local authorization bound to the exact captured tree and semantic view contract. Artifact fields cannot provide this capability.	None

Source code in codenib/index/embedding/vector_store.py

def load(
    self,
    path: Optional[str] = None,
    *,
    native_index_authorization: NativeIndexAuthorization | None = None,
) -> None:
    """
    Load the vector store from disk.

    Args:
        path: Path to load the store from (uses self.store_path if not provided)
        native_index_authorization: Process-local authorization bound to the
            exact captured tree and semantic view contract. Artifact fields
            cannot provide this capability.
    """
    ambient_error = sys.exc_info()[1]
    # Exception subclasses can shadow ``__traceback__``. Read the base
    # descriptor so the ambient snapshot cannot run user code.
    ambient_traceback = (
        None
        if ambient_error is None
        else BaseException.__traceback__.__get__(
            ambient_error,
            type(ambient_error),
        )
    )
    load_path = Path(path) if path else self.store_path
    if load_path is None:
        raise ValueError("No load path provided")
    require_native_index_authorization_preflight(
        native_index_authorization,
        view_type="vector",
    )
    view = capture_authenticated_vector_view(load_path)
    loaded_state: _LoadedVectorState | None = None
    previous_state: _LoadedVectorState | None = None
    view_closed = False
    body_completed = False
    try:
        require_native_index_authorization(
            native_index_authorization,
            view.ownership,
            view_type="vector",
            semantic_contract=self.artifact_metadata,
        )
        loaded_state = self._load_captured(view)
        view.verify_final()
        # A close failure is still a failed replacement. Finish the
        # captured-view lifecycle before exposing any newly parsed state.
        view.close()
        view_closed = True
        previous_state = self._loaded_state
        self._replace_loaded_state(loaded_state)
        body_completed = True
    finally:
        primary_error = sys.exc_info()[1]
        if (
            body_completed
            and primary_error is ambient_error
            and ambient_error is not None
            and BaseException.__traceback__.__get__(
                primary_error,
                type(primary_error),
            )
            is ambient_traceback
        ):
            primary_error = None
        cleanup_error: BaseException | None = None
        if not view_closed:
            try:
                view.close()
                view_closed = True
            except BaseException as exc:  # noqa: B036 - preserve primary
                cleanup_error = exc
        if loaded_state is not None:
            release_state = (
                previous_state
                if self._loaded_state is loaded_state
                else loaded_state
            )
            if release_state is not None:
                try:
                    self._release_vector_indices(
                        tuple(
                            index
                            for index in (
                                release_state.l0_index,
                                release_state.l2_index,
                            )
                            if index is not self.l0_index
                            and index is not self.l2_index
                        )
                    )
                except BaseException as exc:  # noqa: B036 - visit all cleanup
                    if primary_error is not None:
                        _attach_vector_index_cleanup_owner(primary_error, exc)
                    if cleanup_error is None:
                        cleanup_error = exc
                    else:
                        _attach_vector_index_cleanup_owner(cleanup_error, exc)
                        _annotate_secondary_error(
                            cleanup_error,
                            "vector index cleanup also failed",
                            exc,
                        )
        if cleanup_error is not None:
            if primary_error is not None:
                if not view_closed:
                    _attach_vector_view_cleanup_owner(
                        primary_error,
                        view,
                        cleanup_error,
                    )
                _annotate_secondary_error(
                    primary_error,
                    "vector load cleanup also failed",
                    cleanup_error,
                )
            else:
                raise cleanup_error

get_stats

get_stats() -> dict[str, Any]

Get statistics about the vector store.

Returns:

Type	Description
dict[str, Any]	Dictionary with store statistics

Source code in codenib/index/embedding/vector_store.py

def get_stats(self) -> Dict[str, Any]:
    """
    Get statistics about the vector store.

    Returns:
        Dictionary with store statistics
    """
    stats = {
        "embedding_model": self.embedding_model,
        "embedding_provider": self.embedding_provider,
        "embedding_revision": self.embedding_revision,
        "dimension": self.dimension,
        "index_type": self.index_type,
        "index_metric": self.index_metric,
        "l0_documents": len(self.l0_documents),
        "l2_documents": len(self.l2_documents),
        "total_documents": len(self.l0_documents) + len(self.l2_documents),
    }

    # Analyze L0 chunk types
    if self.l0_documents:
        l0_chunk_types = {}
        for doc in self.l0_documents:
            chunk_type = doc.metadata.get("chunk_type", "unknown")
            l0_chunk_types[chunk_type] = l0_chunk_types.get(chunk_type, 0) + 1
        stats["l0_chunk_types"] = l0_chunk_types

    # Analyze L2 chunk types
    if self.l2_documents:
        l2_chunk_types = {}
        for doc in self.l2_documents:
            chunk_type = doc.metadata.get("chunk_type", "unknown")
            l2_chunk_types[chunk_type] = l2_chunk_types.get(chunk_type, 0) + 1
        stats["l2_chunk_types"] = l2_chunk_types

    return stats

get_embeddings_by_content_hash

get_embeddings_by_content_hash(level: Level = 'l2') -> dict[str, ndarray]

Extract raw embedding vectors from the FAISS index, keyed by content hash.

This is used to seed the EmbeddingsCache after a full build so that the first incremental update achieves ~100% cache hit rate for unchanged chunks.

Each document's content is MD5-hashed to produce the key. If the document metadata already contains a content_hash field it is used directly; otherwise the hash is computed on the fly.

Returns:

Type	Description
dict[str, ndarray]	Dict mapping content_hash → np.ndarray (float32 vectors).

Source code in codenib/index/embedding/vector_store.py

def get_embeddings_by_content_hash(
    self, level: Level = "l2"
) -> Dict[str, np.ndarray]:
    """
    Extract raw embedding vectors from the FAISS index, keyed by content hash.

    This is used to seed the ``EmbeddingsCache`` after a full build so that
    the first incremental update achieves ~100% cache hit rate for unchanged
    chunks.

    Each document's content is MD5-hashed to produce the key.  If the
    document metadata already contains a ``content_hash`` field it is used
    directly; otherwise the hash is computed on the fly.

    Returns:
        Dict mapping content_hash → np.ndarray (float32 vectors).
    """
    index, documents = self._get_index_and_docs(level)
    if not documents or index is None or index.ntotal == 0:
        return {}

    result: Dict[str, np.ndarray] = {}
    for i, doc in enumerate(documents):
        content_hash = doc.metadata.get("content_hash")
        if content_hash is None:
            content_hash = hashlib.md5(
                doc.page_content.encode("utf-8", errors="replace")
            ).hexdigest()

        vec = index.reconstruct(i)
        result[content_hash] = np.asarray(vec, dtype=np.float32)

    logger.info(
        "Extracted %d embedding vectors from %s FAISS index for cache seeding.",
        len(result),
        level,
    )
    return result

rebuild_from_embeddings

rebuild_from_embeddings(
    documents: list, embeddings: list[ndarray], level: Level = "l2"
) -> None

Clear level and rebuild its FAISS index from pre-computed embeddings.

Used by the incremental update path: unchanged chunks contribute their cached vectors, so only genuinely new/modified chunks require model inference. No embedding model calls are made by this method.

Parameters:

Name	Type	Description	Default
documents	list	Document-like objects with page_content and metadata attributes (_Document or compatible).	required
embeddings	list[ndarray]	Corresponding embedding vectors as np.ndarray (shape [dim], dtype float32).	required
level	Level	Which index level to rebuild ("l0" or "l2").	'l2'

Raises:

Type	Description
ValueError	If documents and embeddings have different lengths.

Source code in codenib/index/embedding/vector_store.py

def rebuild_from_embeddings(
    self,
    documents: list,
    embeddings: List[np.ndarray],
    level: Level = "l2",
) -> None:
    """
    Clear *level* and rebuild its FAISS index from pre-computed embeddings.

    Used by the incremental update path: unchanged chunks contribute their
    cached vectors, so only genuinely new/modified chunks require model
    inference.  No embedding model calls are made by this method.

    Args:
        documents: Document-like objects with ``page_content`` and
            ``metadata`` attributes (``_Document`` or compatible).
        embeddings: Corresponding embedding vectors as ``np.ndarray``
            (shape ``[dim]``, dtype ``float32``).
        level: Which index level to rebuild (``"l0"`` or ``"l2"``).

    Raises:
        ValueError: If *documents* and *embeddings* have different lengths.
    """
    if len(documents) != len(embeddings):
        raise ValueError(
            f"documents ({len(documents)}) and embeddings ({len(embeddings)}) "
            "must have the same length."
        )

    # Wipe the existing index for this level
    self.clear(level)

    if not documents:
        logger.debug(
            "rebuild_from_embeddings: no documents; level %s cleared.", level
        )
        return

    # Convert to _Document if needed and add vectors to the raw FAISS index
    native_docs = [_to_document(d) for d in documents]
    vectors = np.array(
        [
            emb if isinstance(emb, np.ndarray) else np.asarray(emb)
            for emb in embeddings
        ],
        dtype=np.float32,
    )

    self._add_to_index(level, vectors)
    if level == "l0":
        self.l0_documents = native_docs
    else:
        self.l2_documents = native_docs

    logger.info(
        "rebuild_from_embeddings: %s index rebuilt with %d documents.",
        level,
        len(documents),
    )

delta_update

delta_update(
    all_documents: list,
    all_embeddings: list[ndarray],
    changed_content_hashes: set[str],
    level: Level = "l2",
    threshold: float = 0.1,
) -> None

Patch a flat FAISS index in place when the change set is small.

When the fraction of changed chunks is below threshold, this uses IndexFlat.remove_ids + add to modify only the affected rows, keeping unchanged vectors and their aligned documents untouched. IVF indexes are rebuilt because removing their implicit IDs does not compact the remaining labels to match the document array. If the change ratio exceeds the threshold (or the index is empty), this also falls back to :meth:rebuild_from_embeddings.

Parameters:

Name	Type	Description	Default
all_documents	list	The complete desired set of documents for level after the update. Must carry content_hash in metadata.	required
all_embeddings	list[ndarray]	Corresponding embedding vectors, aligned with all_documents.	required
changed_content_hashes	set[str]	Content hashes of chunks that were added, removed, or modified in this update cycle. Used both to decide between delta/rebuild and to identify stale rows.	required
level	Level	Which index level to update.	'l2'
threshold	float	Maximum change ratio (changed/total) for the delta path; above this a full rebuild is performed.	0.1

Source code in codenib/index/embedding/vector_store.py

def delta_update(
    self,
    all_documents: list,
    all_embeddings: List[np.ndarray],
    changed_content_hashes: Set[str],
    level: Level = "l2",
    threshold: float = 0.1,
) -> None:
    """
    Patch a flat FAISS index in place when the change set is small.

    When the fraction of changed chunks is below *threshold*, this uses
    ``IndexFlat.remove_ids`` + ``add`` to modify only the affected rows,
    keeping unchanged vectors and their aligned documents untouched. IVF
    indexes are rebuilt because removing their implicit IDs does not
    compact the remaining labels to match the document array. If the
    change ratio exceeds the threshold (or the index is empty), this also
    falls back to :meth:`rebuild_from_embeddings`.

    Args:
        all_documents: The complete desired set of documents for *level*
            after the update.  Must carry ``content_hash`` in metadata.
        all_embeddings: Corresponding embedding vectors, aligned with
            *all_documents*.
        changed_content_hashes: Content hashes of chunks that were
            added, removed, or modified in this update cycle.  Used both
            to decide between delta/rebuild and to identify stale rows.
        level: Which index level to update.
        threshold: Maximum change ratio (changed/total) for the delta
            path; above this a full rebuild is performed.
    """
    total = len(all_documents)

    if total == 0:
        self.clear(level)
        return

    index, current_docs = self._get_index_and_docs(level)
    change_ratio = len(changed_content_hashes) / total

    # Fall back to full rebuild when the delta path can't help.
    if (
        index is None
        or index.ntotal == 0
        or not current_docs
        or self.index_type != "flat"
        or change_ratio > threshold
    ):
        logger.info(
            "delta_update: %d/%d changed (%.0f%%) → full rebuild of %s.",
            len(changed_content_hashes),
            total,
            change_ratio * 100,
            level,
        )
        self.rebuild_from_embeddings(all_documents, all_embeddings, level=level)
        return

    # --- Delta path: in-place patch -------------------------------
    # Use a list per hash so duplicate-content docs (same code in
    # different files) are all preserved.
    from collections import defaultdict

    target_by_hash: Dict[str, List[Tuple[object, np.ndarray]]] = defaultdict(list)
    for doc, emb in zip(all_documents, all_embeddings, strict=True):
        ch = doc.metadata.get("content_hash")
        if ch is None:
            # Can't align by hash → safest to rebuild.
            logger.warning(
                "delta_update: target doc missing content_hash → full rebuild."
            )
            self.rebuild_from_embeddings(all_documents, all_embeddings, level=level)
            return
        target_by_hash[ch].append((doc, emb))

    current_hashes = [d.metadata.get("content_hash") for d in current_docs]

    # For each hash, allow at most target-count survivors (handles
    # both duplicate-content additions and removals correctly).
    target_avail: Dict[str, int] = {h: len(v) for h, v in target_by_hash.items()}
    rows_to_remove: List[int] = []
    for i, h in enumerate(current_hashes):
        if (
            h is None
            or h not in target_avail
            or h in changed_content_hashes
            or target_avail[h] <= 0
        ):
            rows_to_remove.append(i)
        else:
            target_avail[h] -= 1

    # Unclaimed target entries become additions.
    docs_to_add: List[Tuple[object, np.ndarray]] = []
    for h, entries in target_by_hash.items():
        claimed = len(entries) - target_avail.get(h, 0)
        docs_to_add.extend(entries[claimed:])

    if rows_to_remove:
        selector = faiss.IDSelectorBatch(np.array(rows_to_remove, dtype=np.int64))
        index.remove_ids(selector)

    # Survivors: prefer the fresh target doc (same content_hash) so that
    # pure metadata changes — file rename, start_line shift, name edit —
    # are reflected without requiring a full rebuild.  The vector is
    # identical because content_hash is identical, so no FAISS op needed.
    remove_set = set(rows_to_remove)
    survivor_idx: Dict[str, int] = {}
    new_docs_list: List[_Document] = []
    for i, d in enumerate(current_docs):
        if i in remove_set:
            continue
        h = current_hashes[i]
        idx = survivor_idx.get(h, 0)
        survivor_idx[h] = idx + 1
        entries = target_by_hash.get(h)
        if entries and idx < len(entries):
            new_docs_list.append(_to_document(entries[idx][0]))
        else:
            new_docs_list.append(d)

    if docs_to_add:
        add_vectors = np.array(
            [np.asarray(e, dtype=np.float32) for _, e in docs_to_add],
            dtype=np.float32,
        )
        self._add_to_index(level, add_vectors)
        new_docs_list.extend(_to_document(d) for d, _ in docs_to_add)

    if level == "l0":
        self.l0_documents = new_docs_list
    else:
        self.l2_documents = new_docs_list

    logger.info(
        "delta_update: %s patched in place — removed %d, added %d "
        "(ntotal=%d, %.0f%% changed).",
        level,
        len(rows_to_remove),
        len(docs_to_add),
        index.ntotal,
        change_ratio * 100,
    )

clear

clear(level: Level | None = None) -> None

Clear data from the vector store.

Parameters:

Name	Type	Description	Default
level	Level | None	If specified, only clear that level ("l0" or "l2"). If None, clear both levels.	None

Source code in codenib/index/embedding/vector_store.py

def clear(self, level: Optional[Level] = None) -> None:
    """
    Clear data from the vector store.

    Args:
        level: If specified, only clear that level ("l0" or "l2").
               If None, clear both levels.
    """
    if level is None or level == "l0":
        logger.info("Clearing L0 vector store")
        self.l0_index = self._build_faiss_index()
        self.l0_documents = []

    if level is None or level == "l2":
        logger.info("Clearing L2 vector store")
        self.l2_index = self._build_faiss_index()
        self.l2_documents = []

    logger.info("Vector store cleared")

RegexNodeIndex

RegexNodeIndex(code_graph: CodeGraph)

In-memory regex-based index for CodeGraph nodes. Supports regex pattern matching on node content with glob filtering.

Parameters:

Name	Type	Description	Default
code_graph	CodeGraph	CodeGraph instance containing nodes to index	required

Methods:

Name	Description
search	Search for pattern in node content (grep-like functionality).

Source code in codenib/index/regex_idx/regex_idx.py

def __init__(self, code_graph: CodeGraph):
    """
    Initialize RegexNodeIndex and build index from CodeGraph.

    Args:
        code_graph: CodeGraph instance containing nodes to index
    """
    self.code_graph = code_graph
    self.nodes: List[NodeInfo] = []
    self._build_index()

    logger.info(f"RegexNodeIndex initialized with {len(self.nodes)} nodes")

search

search(
    pattern: str,
    file_glob: str | None = None,
    node_type: str | None = None,
    case_sensitive: bool = False,
    use_regex: bool = True,
    top_k: int | None = None,
) -> list[NodeInfo]

Search for pattern in node content (grep-like functionality).

Parameters:

Name	Type	Description	Default
pattern	str	Search pattern (regex or plain string)	required
file_glob	str | None	Optional glob to filter by file path (e.g., '.py', '*/calc.py')	None
node_type	str | None	Optional node type to filter (e.g., 'function', 'class', 'file')	None
case_sensitive	bool	Whether search is case-sensitive (default: False)	False
use_regex	bool	Whether to use regex (default: True) or plain string matching	True
top_k	int | None	Optional result limit. Regex searches stop as soon as it is met.	None

Returns:

Type	Description
list[NodeInfo]	List of NodeInfo objects matching the pattern

Examples:

>>> idx.search(r'def\s+\w+', file_glob='*.py')  # Find function defs
>>> idx.search('calculator', use_regex=False)  # Plain string search
>>> idx.search('class', node_type='file')  # Search in file nodes only

Source code in codenib/index/regex_idx/regex_idx.py

def search(
    self,
    pattern: str,
    file_glob: Optional[str] = None,
    node_type: Optional[str] = None,
    case_sensitive: bool = False,
    use_regex: bool = True,
    top_k: Optional[int] = None,
) -> List[NodeInfo]:
    r"""
    Search for pattern in node content (grep-like functionality).

    Args:
        pattern: Search pattern (regex or plain string)
        file_glob: Optional glob to filter by file path (e.g., '*.py', '**/calc.py')
        node_type: Optional node type to filter (e.g., 'function', 'class', 'file')
        case_sensitive: Whether search is case-sensitive (default: False)
        use_regex: Whether to use regex (default: True) or plain string matching
        top_k: Optional result limit. Regex searches stop as soon as it is met.

    Returns:
        List of NodeInfo objects matching the pattern

    Examples:
        >>> idx.search(r'def\s+\w+', file_glob='*.py')  # Find function defs
        >>> idx.search('calculator', use_regex=False)  # Plain string search
        >>> idx.search('class', node_type='file')  # Search in file nodes only
    """
    if top_k is not None and (
        isinstance(top_k, bool) or not isinstance(top_k, int) or top_k < 1
    ):
        raise ValueError("top_k must be a positive integer")

    if use_regex:
        if len(pattern) > MAX_REGEX_PATTERN_CHARS:
            raise ValueError(
                "Regex pattern exceeds "
                f"the {MAX_REGEX_PATTERN_CHARS}-character limit"
            )

        # Start the request-wide clock before compiling or evaluating either
        # user-controlled structural filter. A single filtered-out scan is
        # still work charged to this request.
        deadline = time.monotonic() + REGEX_SEARCH_TIMEOUT_SECONDS
        flags = 0 if case_sensitive else regex.IGNORECASE
        try:
            compiled = regex.compile(pattern, flags)
        except regex.error as exc:
            logger.error("Invalid regex pattern %r: %s", pattern, exc)
            raise ValueError(f"Invalid regex pattern: {exc}") from exc
        _remaining_time(deadline)

        matches: List[NodeInfo] = []
        scanned_nodes = 0
        candidate_nodes = 0
        for node in self.nodes:
            _remaining_time(deadline)
            scanned_nodes += 1
            if scanned_nodes > MAX_REGEX_SCANNED_NODES:
                raise RegexSearchBudgetError(
                    "Regex search exceeded the "
                    f"{MAX_REGEX_SCANNED_NODES}-node scan budget; "
                    "use file_glob or node_type to narrow the search"
                )

            if file_glob:
                path_matches = bool(node.file) and fnmatch(node.file, file_glob)
                _remaining_time(deadline)
                if not path_matches:
                    continue
            if node_type and node.type != node_type:
                continue
            if not node.content:
                continue

            candidate_nodes += 1
            if candidate_nodes > MAX_REGEX_CANDIDATES:
                raise RegexSearchBudgetError(
                    "Regex search exceeded the "
                    f"{MAX_REGEX_CANDIDATES}-candidate match budget; "
                    "use file_glob or node_type to narrow the search"
                )

            remaining = _remaining_time(deadline)
            try:
                matched = compiled.search(node.content, timeout=remaining)
            except TimeoutError as exc:
                raise _timeout_error() from exc
            _remaining_time(deadline)
            if matched:
                matches.append(node)
                if top_k is not None and len(matches) >= top_k:
                    break

        logger.debug(
            "Regex search pattern=%r file_glob=%r node_type=%r: "
            "found %d matches after scanning %d nodes and %d candidates",
            pattern,
            file_glob,
            node_type,
            len(matches),
            scanned_nodes,
            candidate_nodes,
        )
        return matches

    # Preserve the unrestricted plain-string path for direct index users.
    candidates = self.nodes
    if file_glob:
        candidates = [
            node
            for node in candidates
            if node.file and fnmatch(node.file, file_glob)
        ]
    if node_type:
        candidates = [node for node in candidates if node.type == node_type]

    if case_sensitive:
        matches = [
            node for node in candidates if node.content and pattern in node.content
        ]
    else:
        pattern_lower = pattern.lower()
        matches = [
            node
            for node in candidates
            if node.content and pattern_lower in node.content.lower()
        ]
    if top_k is not None:
        matches = matches[:top_k]

    logger.debug(
        "Plain search pattern=%r file_glob=%r node_type=%r: "
        "found %d matches from %d candidates",
        pattern,
        file_glob,
        node_type,
        len(matches),
        len(candidates),
    )

    return matches

BM25CodeIndexer

BM25CodeIndexer(
    code_graph=None,
    chunks=None,
    max_k: int = 15,
    language: str = "english",
    project_root: str | None = None,
    *,
    prepare_only: bool = False,
    check_cancelled: Callable[[], None] | None = None
)

A class that builds a BM25 index from CodeGraph nodes and provides search functionality with stemming support.

Parameters:

Name	Type	Description	Default
code_graph		CodeGraph instance containing nodes to index. If provided, the index will be built immediately.	None
chunks		List of CodeChunk objects to index. If provided, the index will be built immediately.	None
max_k	int	Maximum number of results to return in searches	15
language	str	Language for stopword removal Default is "english" which works well for processing code tokens as it treats special characters as separators	'english'
project_root	str | None	Repository root used to resolve relative source paths	None
prepare_only	bool	Prepare canonical persisted documents without building the serving-time in-memory rank index.	False
check_cancelled	Callable[[], None] | None	Cooperative stop check for prepare-only chunk builds.	None

Methods:

Name	Description
build_index_from_graph	Build a BM25 index from a CodeGraph.
build_index_from_chunks	Build a BM25 index from a list of CodeChunk objects.
bind_repository_source	Attach a retained source authority chosen by the runtime caller.
save_index	Save the index to a directory.
load_index	Load the index from a directory.
load_index_values	Load already-decoded BM25 values from an authenticated reader.

Source code in codenib/index/sparse_idx/bm25_index.py

def __init__(
    self,
    code_graph=None,
    chunks=None,
    max_k: int = 15,
    language: str = "english",
    project_root: Optional[str] = None,
    *,
    prepare_only: bool = False,
    check_cancelled: Callable[[], None] | None = None,
):
    """
    Initialize the BM25CodeIndexer and optionally build the index immediately.

    Args:
        code_graph: CodeGraph instance containing nodes to index. If provided,
                   the index will be built immediately.
        chunks: List of CodeChunk objects to index. If provided, the index will
               be built immediately.
        max_k: Maximum number of results to return in searches
        language: Language for stopword removal
                  Default is "english" which works well for processing code tokens
                  as it treats special characters as separators
        project_root: Repository root used to resolve relative source paths
        prepare_only: Prepare canonical persisted documents without building
                      the serving-time in-memory rank index.
        check_cancelled: Cooperative stop check for prepare-only chunk builds.
    """
    self.max_k = max_k
    self.language = language
    self.documents = []
    if type(prepare_only) is not bool:
        raise TypeError("BM25 prepare-only policy must be an exact boolean")
    if check_cancelled is not None and not callable(check_cancelled):
        raise TypeError("BM25 build cancellation must be callable")
    if code_graph is not None and (prepare_only or check_cancelled is not None):
        raise ValueError(
            "BM25 prepare-only mode and cancellation require source chunks "
            "without a code graph"
        )
    if prepare_only and chunks is None:
        raise ValueError("BM25 prepare-only mode requires source chunks")
    if check_cancelled is not None and chunks is None:
        raise ValueError("BM25 build cancellation requires source chunks")
    if check_cancelled is not None and not prepare_only:
        raise ValueError("BM25 build cancellation requires prepare-only mode")

    self.retriever = None
    self._documents_prepared = False
    self.code_graph: CodeGraph = None
    self.project_root = project_root
    self.source_mode = SOURCE_MODE_LEGACY_DIRECT
    self._source_binding: RepositorySourceBinding | None = None
    self.nodes: List[str] = []

    # Build the index immediately if a code_graph is provided
    if code_graph is not None:
        self.build_index_from_graph(code_graph)
    elif chunks is not None:
        if prepare_only:
            self._prepare_documents_from_chunks(
                chunks,
                project_root=project_root,
                check_cancelled=check_cancelled,
            )
            self._documents_prepared = True
        else:
            # Preserve the established virtual-call shape for subclasses
            # that override this public method.
            self.build_index_from_chunks(chunks, project_root=project_root)

build_index_from_graph

build_index_from_graph(code_graph: CodeGraph) -> BM25Retriever

Build a BM25 index from a CodeGraph.

Parameters:

Name	Type	Description	Default
code_graph	CodeGraph	CodeGraph instance containing nodes to index	required

Source code in codenib/index/sparse_idx/bm25_index.py

def build_index_from_graph(self, code_graph: CodeGraph) -> BM25Retriever:
    """
    Build a BM25 index from a CodeGraph.

    Args:
        code_graph: CodeGraph instance containing nodes to index
    """
    # Reset the index
    self.documents = []
    self.nodes = []
    self.code_graph = code_graph
    self.project_root = code_graph.project_root
    self.source_mode = SOURCE_MODE_LEGACY_DIRECT
    self._source_binding = None
    self._documents_prepared = False
    self.retriever = None

    # Convert graph nodes to documents
    for vertex in code_graph.graph.vs:
        doc = self._convert_vertex_to_document(vertex)
        if doc is not None:
            self.documents.append(doc)
            node_name = doc.metadata.get("node_id") or doc.metadata.get("name")
            if node_name:
                self.nodes.append(node_name)

    # Create BM25Retriever with LangChain format
    self.retriever = BM25Retriever.from_documents(self.documents, k=self.max_k)
    self._documents_prepared = True

    return self.retriever

build_index_from_chunks

build_index_from_chunks(
    chunks: list[CodeChunk], *, project_root: str | None = None
) -> BM25Retriever

Build a BM25 index from a list of CodeChunk objects.

Parameters:

Name	Type	Description	Default
chunks	list[CodeChunk]	List of CodeChunk objects (with node_id, chunk_type, name, file, etc.)	required
project_root	str | None	Repository root used to resolve relative source paths	None

Returns:

Type	Description
BM25Retriever	BM25Retriever instance

Source code in codenib/index/sparse_idx/bm25_index.py

def build_index_from_chunks(
    self,
    chunks: List[CodeChunk],
    *,
    project_root: Optional[str] = None,
) -> BM25Retriever:
    """
    Build a BM25 index from a list of CodeChunk objects.

    Args:
        chunks: List of CodeChunk objects (with node_id, chunk_type, name, file, etc.)
        project_root: Repository root used to resolve relative source paths

    Returns:
        BM25Retriever instance
    """
    self._prepare_documents_from_chunks(
        chunks,
        project_root=project_root,
        check_cancelled=None,
    )
    self.retriever = BM25Retriever.from_documents(self.documents, k=self.max_k)
    self._documents_prepared = True
    return self.retriever

bind_repository_source

bind_repository_source(binding: RepositorySourceBinding) -> None

Attach a retained source authority chosen by the runtime caller.

Source code in codenib/index/sparse_idx/bm25_index.py

def bind_repository_source(self, binding: RepositorySourceBinding) -> None:
    """Attach a retained source authority chosen by the runtime caller."""

    source_identity = binding.authenticated_identity_snapshot()
    self._source_binding = binding
    self.project_root = str(source_identity.root)
    self.source_mode = SOURCE_MODE_BOUND_REPOSITORY

save_index

save_index(directory_path: str, *, check_cancelled: Callable[[], None] | None = None)

Save the index to a directory.

Parameters:

Name	Type	Description	Default
directory_path	str	Path to save the index to	required

Source code in codenib/index/sparse_idx/bm25_index.py

def save_index(
    self,
    directory_path: str,
    *,
    check_cancelled: Callable[[], None] | None = None,
):
    """
    Save the index to a directory.

    Args:
        directory_path: Path to save the index to
    """
    if check_cancelled is not None and not callable(check_cancelled):
        raise TypeError("BM25 persistence cancellation must be callable")
    if self.retriever is None and not self._documents_prepared:
        raise ValueError(
            "Index has not been built. Call build_index_from_graph first."
        )
    if check_cancelled is not None:
        check_cancelled()

    # Create directory if it doesn't exist
    os.makedirs(directory_path, exist_ok=True)

    # Save documents as JSON since LangChain BM25Retriever doesn't have persist method
    documents_data = []
    for doc in self.documents:
        if check_cancelled is not None:
            check_cancelled()
        documents_data.append(
            {"page_content": doc.page_content, "metadata": doc.metadata}
        )

    documents_file = os.path.join(directory_path, "documents.json")
    _write_json_interruptibly(
        documents_file,
        documents_data,
        check_cancelled,
    )

    # Save additional metadata including project_root
    metadata = {
        "project_root": (
            str(self.project_root) if self.project_root is not None else None
        ),
        "max_k": self.max_k,
        "language": self.language,
    }
    metadata_file = os.path.join(directory_path, "bm25_metadata.json")
    _write_json_interruptibly(
        metadata_file,
        metadata,
        check_cancelled,
    )

load_index

load_index(directory_path: str)

Load the index from a directory.

Parameters:

Name	Type	Description	Default
directory_path	str	Path to load the index from	required

Source code in codenib/index/sparse_idx/bm25_index.py

def load_index(self, directory_path: str):
    """
    Load the index from a directory.

    Args:
        directory_path: Path to load the index from
    """
    if not os.path.exists(directory_path):
        raise ValueError(f"Directory {directory_path} does not exist.")

    # Load documents from JSON
    documents_file = os.path.join(directory_path, "documents.json")
    if not os.path.exists(documents_file):
        raise ValueError(f"Documents file not found: {documents_file}")

    with open(documents_file, "r", encoding="utf-8") as f:
        documents_data = json.load(f)

    metadata_file = os.path.join(directory_path, "bm25_metadata.json")
    if os.path.exists(metadata_file):
        with open(metadata_file, "r", encoding="utf-8") as f:
            metadata = json.load(f)
    else:
        metadata = {
            "project_root": None,
            "max_k": self.max_k,
            "language": self.language,
        }
    self.load_index_values(documents_data, metadata)

load_index_values

load_index_values(
    documents_data: Iterable[object],
    metadata: Mapping[str, Any] | None,
    *,
    source_mode: str = SOURCE_MODE_LEGACY_DIRECT
) -> None

Load already-decoded BM25 values from an authenticated reader.

Artifact runtimes use this entry point so persisted bytes can be read through a pinned directory authority instead of being reopened by path.

Source code in codenib/index/sparse_idx/bm25_index.py

def load_index_values(
    self,
    documents_data: Iterable[object],
    metadata: Mapping[str, Any] | None,
    *,
    source_mode: str = SOURCE_MODE_LEGACY_DIRECT,
) -> None:
    """Load already-decoded BM25 values from an authenticated reader.

    Artifact runtimes use this entry point so persisted bytes can be read
    through a pinned directory authority instead of being reopened by path.
    """

    if metadata is not None and not isinstance(metadata, Mapping):
        raise ValueError("BM25 metadata must be an object")
    metadata = {} if metadata is None else metadata
    max_k = metadata.get("max_k", 10)
    language = metadata.get("language", "english")
    project_root = metadata.get("project_root")
    if isinstance(max_k, bool) or not isinstance(max_k, int) or max_k <= 0:
        raise ValueError("BM25 max_k must be a positive integer")
    if not isinstance(language, str) or not language:
        raise ValueError("BM25 language must be a non-empty string")
    if project_root is not None and not isinstance(project_root, str):
        raise ValueError("BM25 project_root must be a string or null")
    if source_mode not in _SOURCE_MODES:
        raise ValueError(f"unsupported BM25 source mode: {source_mode!r}")

    # Restore serving configuration before creating the retriever.  Its k
    # value is fixed at construction time.
    self.project_root = project_root
    self.source_mode = source_mode
    self._source_binding = None
    self._documents_prepared = False
    self.max_k = max_k
    self.language = language

    # Reconstruct Document objects one decoded element at a time.
    self.documents = []
    self.nodes = []
    seen_nodes = set()
    for doc_data in documents_data:
        if not isinstance(doc_data, Mapping):
            raise ValueError("BM25 document must be an object")
        page_content = doc_data.get("page_content")
        document_metadata = doc_data.get("metadata")
        if not isinstance(page_content, str) or not isinstance(
            document_metadata, Mapping
        ):
            raise ValueError("BM25 document content or metadata is invalid")
        doc = Document(
            page_content=page_content,
            metadata=dict(document_metadata),
        )
        self.documents.append(doc)
        node_name = doc.metadata.get("node_id") or doc.metadata.get("name")
        if node_name and node_name not in seen_nodes:
            self.nodes.append(node_name)
            seen_nodes.add(node_name)

    self.retriever = BM25Retriever.from_documents(self.documents, k=self.max_k)
    self._documents_prepared = True

LSIndexer

LSIndexer(
    project_root: str | Path,
    output_dir: str | Path | None = None,
    exclude_patterns: list | None = None,
    profiler: Profiler | None = None,
    language: str | None = None,
    decoder_backend: str | None = None,
    graph_route: str = ACTIVE_GRAPH_ROUTE,
)

Unified indexer that delegates to language-specific implementations.

Methods:

Name	Description
process_query_index	Build a backend-specific capability-scoped query index.
process_query_provider	Build a backend-specific provider with compatible fallbacks.
graph_patch	Incrementally update graph using LSP graph-patching.

Attributes:

Name	Type	Description
index_quality_report		Latest backend quality report, when the indexer provides one.
index_generation_report		Latest backend generation report, when the indexer provides one.
decoded_input_receipt		Decoded input generation consumed by graph construction, if any.
graph_output_receipt		Graph generation written by the pipeline invocation, if any.
supports_partial_index	bool	Whether this backend can identify partial output from this run.

Source code in codenib/ls_router.py

def __init__(
    self,
    project_root: Union[str, Path],
    output_dir: Optional[Union[str, Path]] = None,
    exclude_patterns: Optional[List] = None,
    profiler: Optional[Profiler] = None,
    language: Optional[str] = None,
    decoder_backend: Optional[str] = None,
    graph_route: str = ACTIVE_GRAPH_ROUTE,
):
    activate_managed_toolchain()
    self.project_root = Path(project_root).absolute()
    self.graph_route = graph_route
    self.language = _normalize_language(language, graph_route=graph_route)
    self.decoder_backend = decoder_backend

    self._delegate = self._create_indexer(
        project_root=project_root,
        output_dir=output_dir,
        exclude_patterns=exclude_patterns,
        profiler=profiler,
    )

    # Expose delegate attributes for backward compatibility
    self.output_dir = self._delegate.output_dir
    self.index_file = self._delegate.index_file
    self.decoded_file = self._delegate.decoded_file
    self.graph_file = self._delegate.graph_file
    self.profiler = self._delegate.profiler
    self._graph_output_receipt: Optional[Dict[str, Any]] = None

    logger.info(
        "Initialized LSIndexer for %s at %s (route=%s)",
        self.language,
        self.project_root,
        self.graph_route,
    )

index_quality_report property

index_quality_report

Latest backend quality report, when the indexer provides one.

index_generation_report property

index_generation_report

Latest backend generation report, when the indexer provides one.

decoded_input_receipt property

decoded_input_receipt

Decoded input generation consumed by graph construction, if any.

graph_output_receipt property

graph_output_receipt

Graph generation written by the pipeline invocation, if any.

supports_partial_index property

supports_partial_index: bool

Whether this backend can identify partial output from this run.

process_query_index

process_query_index() -> Any

Build a backend-specific capability-scoped query index.

Source code in codenib/ls_router.py

def process_query_index(self) -> Any:
    """Build a backend-specific capability-scoped query index."""

    method = getattr(self._delegate, "process_query_index", None)
    if not callable(method):
        raise RuntimeError(
            f"{self.language} indexer does not expose a query-index path"
        )
    return method()

process_query_provider

process_query_provider(*, require_native: bool = False) -> Any

Build a backend-specific provider with compatible fallbacks.

Source code in codenib/ls_router.py

def process_query_provider(self, *, require_native: bool = False) -> Any:
    """Build a backend-specific provider with compatible fallbacks."""

    method = getattr(self._delegate, "process_query_provider", None)
    if not callable(method):
        raise RuntimeError(
            f"{self.language} indexer does not expose a query-provider path"
        )
    if require_native:
        return method(require_native=True)
    return method()

graph_patch

graph_patch(graph: CodeGraph, base_commit: str, target_commit: str = 'HEAD') -> dict

Incrementally update graph using LSP graph-patching.

Parameters:

Name	Type	Description	Default
graph	CodeGraph	Existing CodeGraph to update in place.	required
base_commit	str	Git commit hash the graph was built from.	required
target_commit	str	Git commit hash to patch to (default HEAD).	'HEAD'

Returns:

Type	Description
dict	Statistics dict from the patcher.

Source code in codenib/ls_router.py

def graph_patch(
    self,
    graph: "CodeGraph",
    base_commit: str,
    target_commit: str = "HEAD",
) -> dict:
    """Incrementally update graph using LSP graph-patching.

    Args:
        graph: Existing CodeGraph to update in place.
        base_commit: Git commit hash the graph was built from.
        target_commit: Git commit hash to patch to (default HEAD).

    Returns:
        Statistics dict from the patcher.
    """
    from .graph.incremental.graph_patcher import LANGUAGE_EXTENSIONS, GraphPatcher

    patcher = GraphPatcher(
        project_root=str(self.project_root),
        code_graph=graph,
        language=self.language,
    )

    changed = patcher.detect_changed_files(
        str(self.project_root),
        base_commit,
        target_commit,
        extensions=LANGUAGE_EXTENSIONS.get(self.language),
    )
    return patcher.patch_files(
        changed,
        earlier_commit=base_commit,
        later_commit=target_commit,
    )

CodeSearchEngine

CodeSearchEngine(
    repo_path: str,
    llm_model: str = "gpt-4o",
    llm_temperature: float = 0.0,
    llm_max_tokens: int = 8192,
    top_k: int = 10,
    language: str = "english",
    instance_id: str | None = None,
)

Integrated search engine that combines SCIP indexing, code graph representation, BM25 indexing, and keyword extraction to provide context-aware code search.

Parameters:

Name	Type	Description	Default
repo_path	str	Path to the repository to index and search.	required
llm_model	str	Full litellm model identifier for keyword extraction (e.g., "gpt-4o", "vertex_ai/gemini-2.5-flash").	'gpt-4o'
llm_temperature	float	Temperature for LLM sampling.	0.0
llm_max_tokens	int	Maximum tokens for LLM responses.	8192
top_k	int	Number of top results to return.	10
language	str	Language for BM25 indexer (default is "english").	'english'
instance_id	str | None	Optional instance ID for the dataset.	None

Methods:

Name	Description
index_repository	Index the repository using SCIP and build BM25 index.
extract_keywords	Extract keywords from a problem statement.
search	Search the repository for code relevant to the problem statement.
get_code_context	Get source code context for a search result.
get_code_context_by_node_name	Get source code context for a node by its name.
get_predecessor_nodes	Get predecessor nodes for a given node name.
get_rich_result_context	Enhance a search result with source code context.
search_with_context	Search and include source code context in the results.

Source code in codenib/search.py

def __init__(
    self,
    repo_path: str,
    llm_model: str = "gpt-4o",
    llm_temperature: float = 0.0,
    llm_max_tokens: int = 8192,
    top_k: int = 10,
    language: str = "english",
    instance_id: Optional[str] = None,
):
    """
    Initialize the CodeSearchEngine.

    Args:
        repo_path: Path to the repository to index and search.
        llm_model: Full litellm model identifier for keyword extraction
            (e.g., ``"gpt-4o"``, ``"vertex_ai/gemini-2.5-flash"``).
        llm_temperature: Temperature for LLM sampling.
        llm_max_tokens: Maximum tokens for LLM responses.
        top_k: Number of top results to return.
        language: Language for BM25 indexer (default is "english").
        instance_id: Optional instance ID for the dataset.
    """
    self.repo_path = os.path.abspath(repo_path)
    self.repo_name = os.path.basename(self.repo_path)

    self.llm = LiteLLMChat(
        model=llm_model,
        temperature=llm_temperature,
        max_tokens=llm_max_tokens,
    )
    self.top_k = top_k
    self.language = language

    # Components
    self.code_graph: CodeGraph = None
    self.bm25_indexer = None

    logger.info(f"Initialized CodeSearchEngine for repository: {self.repo_name}")
    self.index_repository(repo_name=self.repo_name, instance_id=instance_id)

index_repository

index_repository(repo_name: str | None = None, instance_id: str | None = None) -> bool

Index the repository using SCIP and build BM25 index.

Parameters:

Name	Type	Description	Default
repo_name	str | None	Optional name for the repository	None
instance_id	str | None	Optional instance ID for the dataset	None

Returns:

Name	Type	Description
bool	bool	True if indexing was successful, False otherwise

Source code in codenib/search.py

def index_repository(
    self,
    repo_name: Optional[str] = None,
    instance_id: Optional[str] = None,
) -> bool:
    """
    Index the repository using SCIP and build BM25 index.

    Args:
        repo_name: Optional name for the repository
        instance_id: Optional instance ID for the dataset

    Returns:
        bool: True if indexing was successful, False otherwise
    """
    # Create SCIP indexer
    logger.info(f"Indexing {repo_name} repository using SCIP...")

    output_path = str(user_state_dir() / (instance_id or self.repo_name))
    os.makedirs(output_path, exist_ok=True)
    scip_indexer = LSIndexer(self.repo_path, output_dir=output_path)

    logger.info(f"Saved decoded scip index to {output_path}")

    # Build code graph
    # Run the indexing pipeline
    self.code_graph = scip_indexer.run_pipeline(project_name=instance_id)

    # Build BM25 index
    logger.info("Building BM25 index from code graph...")
    try:
        self.bm25_indexer = BM25CodeIndexer(
            code_graph=self.code_graph, max_k=self.top_k, language=self.language
        )
    except Exception as e:
        logger.error(f"Error building BM25 index: {e}")
        return False

    return True

extract_keywords

extract_keywords(problem_statement: str) -> KeywordExtraction

Extract keywords from a problem statement.

Parameters:

Name	Type	Description	Default
problem_statement	str	The problem statement to analyze	required

Returns:

Type	Description
KeywordExtraction	KeywordExtraction object containing extracted keywords

Source code in codenib/search.py

def extract_keywords(self, problem_statement: str) -> KeywordExtraction:
    """
    Extract keywords from a problem statement.

    Args:
        problem_statement: The problem statement to analyze

    Returns:
        KeywordExtraction object containing extracted keywords
    """
    logger.info("Extracting keywords from problem statement...")
    try:
        keywords = extract_keywords_from_statement(
            problem_statement,
            llm=self.llm,
        )
        logger.info(
            f"Extracted {len(keywords.keywords)} keywords: {', '.join(keywords.keywords)}"
        )
        return keywords
    except Exception as e:
        logger.error(f"Error extracting keywords: {e}")
        # Return empty keywords if extraction fails
        return KeywordExtraction(keywords=[])

search

search(problem_statement: str, use_keyword_extraction: bool = True) -> list[NodeInfo]

Search the repository for code relevant to the problem statement.

Parameters:

Name	Type	Description	Default
problem_statement	str	The problem statement or query	required
use_keyword_extraction	bool	Whether to use keyword extraction.	True

Returns:

Type	Description
list[NodeInfo]	List of search results with scores and locations

Source code in codenib/search.py

def search(
    self,
    problem_statement: str,
    use_keyword_extraction: bool = True,
) -> List[NodeInfo]:
    """
    Search the repository for code relevant to the problem statement.

    Args:
        problem_statement: The problem statement or query
        use_keyword_extraction: Whether to use keyword extraction.

    Returns:
        List of search results with scores and locations
    """
    if self.bm25_indexer is None:
        logger.error("Repository is not indexed. Call index_repository() first.")
        return []

    search_query = problem_statement

    # Extract keywords if requested
    if use_keyword_extraction:
        extracted = self.extract_keywords(problem_statement)
        if extracted and extracted.keywords:
            # Combine extracted keywords into a search query
            search_query = " ".join(extracted.keywords)

    logger.info(f"Searching with query: {search_query}")

    # Perform search
    try:
        results = self.bm25_indexer.search(search_query)
        logger.info(f"Found {len(results)} results")
        return results
    except Exception as e:
        logger.error(f"Error during search: {e}")
        return []

get_code_context

get_code_context(result: Mapping[str, Any] | NodeInfo) -> str | None

Get source code context for a search result.

Parameters:

Name	Type	Description	Default
result	Mapping[str, Any] | NodeInfo	A single search result from the search method	required

Returns:

Type	Description
str | None	Source code context as a string, or None if not available

Source code in codenib/search.py

def get_code_context(
    self,
    result: Mapping[str, Any] | NodeInfo,
) -> Optional[str]:
    """
    Get source code context for a search result.

    Args:
        result: A single search result from the search method

    Returns:
        Source code context as a string, or None if not available
    """
    normalized = self._result_mapping(result)
    path = self._resolve_source_file(normalized.get("file"))
    if path is None:
        return None
    content = self._read_source(path)
    if content is None:
        return None

    type_of_result = normalized.get("type")
    if type_of_result == NODE_TYPE_FILE:
        return content
    if not is_symbol_node(type_of_result):
        return None
    return self._source_range(
        content,
        normalized.get("start_line"),
        normalized.get("end_line"),
    )

get_code_context_by_node_name

get_code_context_by_node_name(node_name: str) -> str | None

Get source code context for a node by its name.

Parameters:

Name	Type	Description	Default
node_name	str	The name of the graph node	required

Returns:

Type	Description
str | None	Source code context as a string, or None if not available

Source code in codenib/search.py

def get_code_context_by_node_name(
    self,
    node_name: str,
) -> Optional[str]:
    """
    Get source code context for a node by its name.

    Args:
        node_name: The name of the graph node

    Returns:
        Source code context as a string, or None if not available
    """
    # Search for the node in the code graph
    attributes = self.code_graph.get_node_info_by_name(node_name)
    if not attributes:
        logger.warning(f"Node {node_name!r} not found in the code graph.")
        return None
    if attributes.get("type") == NODE_TYPE_FILE:
        attributes = dict(attributes)
        attributes["file"] = attributes.get("name")
    return self.get_code_context(attributes)

get_predecessor_nodes

get_predecessor_nodes(node_name: str) -> list[str]

Get predecessor nodes for a given node name.

Parameters:

Name	Type	Description	Default
node_name	str	The name of the graph node	required

Returns:

Type	Description
list[str]	List of predecessor node names

Source code in codenib/search.py

def get_predecessor_nodes(
    self,
    node_name: str,
) -> List[str]:
    """
    Get predecessor nodes for a given node name.

    Args:
        node_name: The name of the graph node

    Returns:
        List of predecessor node names
    """
    if self.code_graph is None:
        logger.error("Code graph is not initialized.")
        return []

    predecessors = self.code_graph.get_predecessors(node_name)
    if not predecessors:
        logger.warning(f"No predecessors found for node {node_name!r}.")

    return predecessors

get_rich_result_context

get_rich_result_context(
    result: Mapping[str, Any] | NodeInfo, context_lines: int = 5
) -> dict[str, Any]

Enhance a search result with source code context.

Parameters:

Name	Type	Description	Default
result	Mapping[str, Any] | NodeInfo	A single search result from the search method	required
context_lines	int	Number of context lines to include before and after	5

Returns:

Type	Description
dict[str, Any]	Enhanced result with source code context

Source code in codenib/search.py

def get_rich_result_context(
    self,
    result: Mapping[str, Any] | NodeInfo,
    context_lines: int = 5,
) -> Dict[str, Any]:
    """
    Enhance a search result with source code context.

    Args:
        result: A single search result from the search method
        context_lines: Number of context lines to include before and after

    Returns:
        Enhanced result with source code context
    """
    if (
        not isinstance(context_lines, int)
        or isinstance(context_lines, bool)
        or not 0 <= context_lines <= _MAX_CONTEXT_LINES
    ):
        raise ValueError(
            f"context_lines must be between 0 and {_MAX_CONTEXT_LINES}"
        )
    enhanced_result = self._result_mapping(result)
    path = self._resolve_source_file(enhanced_result.get("file"))
    if path is None:
        return enhanced_result
    content = self._read_source(path)
    if content is None:
        return enhanced_result

    start_line = self._line_number(enhanced_result.get("start_line"))
    end_line = self._line_number(enhanced_result.get("end_line"))
    if start_line is None or end_line is None or end_line < start_line:
        return enhanced_result
    lines = content.splitlines(keepends=True)
    context_start = max(0, start_line - context_lines)
    context_end = min(len(lines) - 1, end_line + context_lines)
    if context_start > context_end:
        return enhanced_result
    enhanced_result["context_code"] = "".join(
        lines[context_start : context_end + 1]
    )
    enhanced_result["context_range"] = {
        "start": context_start,
        "end": context_end,
    }
    return enhanced_result

search_with_context

search_with_context(
    problem_statement: str, context_lines: int = 5, use_keyword_extraction: bool = True
) -> list[dict[str, Any]]

Search and include source code context in the results.

Parameters:

Name	Type	Description	Default
problem_statement	str	The problem statement or query	required
context_lines	int	Number of context lines to include	5
use_keyword_extraction	bool	Whether to use keyword extraction	True

Returns:

Type	Description
list[dict[str, Any]]	List of search results with code context

Source code in codenib/search.py

def search_with_context(
    self,
    problem_statement: str,
    context_lines: int = 5,
    use_keyword_extraction: bool = True,
) -> List[Dict[str, Any]]:
    """
    Search and include source code context in the results.

    Args:
        problem_statement: The problem statement or query
        context_lines: Number of context lines to include
        use_keyword_extraction: Whether to use keyword extraction

    Returns:
        List of search results with code context
    """
    # First get the basic search results
    results = self.search(
        problem_statement,
        use_keyword_extraction=use_keyword_extraction,
    )

    # Enhance each result with context
    enhanced_results = []
    for result in results:
        enhanced_result = self.get_rich_result_context(result, context_lines)
        enhanced_results.append(enhanced_result)

    return enhanced_results

create_code_vector_store

create_code_vector_store(
    embedding_model: str = "text-embedding-ada-002",
    embedding_provider: str = "openai",
    store_path: str | None = None,
    **kwargs
) -> CodeVectorStore

Factory function to create a CodeVectorStore.

Parameters:

Name	Type	Description	Default
embedding_model	str	Name of the embedding model	'text-embedding-ada-002'
embedding_provider	str	Provider for embeddings	'openai'
store_path	str | None	Path to store/load the vector store	None
**kwargs		Additional arguments for CodeVectorStore	{}

Returns:

Type	Description
CodeVectorStore	CodeVectorStore instance

Source code in codenib/index/embedding/vector_store.py

def create_code_vector_store(
    embedding_model: str = "text-embedding-ada-002",
    embedding_provider: str = "openai",
    store_path: Optional[str] = None,
    **kwargs,
) -> CodeVectorStore:
    """
    Factory function to create a CodeVectorStore.

    Args:
        embedding_model: Name of the embedding model
        embedding_provider: Provider for embeddings
        store_path: Path to store/load the vector store
        **kwargs: Additional arguments for CodeVectorStore

    Returns:
        CodeVectorStore instance
    """
    return CodeVectorStore(
        embedding_model=embedding_model,
        embedding_provider=embedding_provider,
        store_path=store_path,
        **kwargs,
    )