Function run_source_search

agent/src/opentrace_agent/cli/source_search.py:178–216

Entry point for the source-search subcommand. *db_path* must be a resolved path to an existing index. *repo* is a canonical repo id (the ``id`` field from the ``repos`` command output); pass ``None`` to search across all indexed repos. *node_types* optionally restricts results to sp

(
    query: str,
    db_path: str | None,
    *,
    repo: str | None = None,
    node_types: list[str] | None = None,
    limit: int = 20,
    output_json: bool = False,
)

def run_source_search(
    query: str,
    db_path: str | None,
    *,
    repo: str | None = None,
    node_types: list[str] | None = None,
    limit: int = 20,
    output_json: bool = False,
) -> None:
    """Entry point for the source-search subcommand.

    *db_path* must be a resolved path to an existing index. *repo* is
    a canonical repo id (the ``id`` field from the ``repos`` command
    output); pass ``None`` to search across all indexed repos.
    *node_types* optionally restricts results to specific node types
    (``Function``, ``Class``, ...). *output_json* emits a structured
    object instead of formatted text.
    """
    from opentrace_agent.store import GraphStore

    store = GraphStore(db_path, read_only=True)
    try:
        resolved_repo = _resolve_repo(store, repo)
        # Over-fetch one row beyond `limit` so we can distinguish
        # "exactly `limit` matches in the graph" from "more existed,
        # truncated here". The +1 row is then trimmed before output.
        results = _run_fts_search(store, query, resolved_repo, node_types, limit + 1)
        truncated = len(results) > limit
        if truncated:
            results = results[:limit]

        repo_ids = [resolved_repo] if resolved_repo else _load_repo_ids(store)

        if output_json:
            _emit_json(query, resolved_repo, results, truncated, limit)
        else:
            _emit_text(query, resolved_repo, results, repo_ids)
    finally:
        store.close()


def _emit_json(