Function _run_indexing_pipeline

agent/src/opentrace_agent/cli/main.py:356–438

Run the four-stage pipeline with atomic-write staging. Writes to ``<db_path>.staging`` (seeded from the live DB so prior repos are preserved) and atomically renames over ``<db_path>`` on success. Holds an exclusive flock on ``<db>.indexlock`` so two concurrent indexes can't race the

(
    *,
    source_path: Path,
    repo_id: str,
    db_path: str,
    batch_size: int,
    verbose: bool,
    extra_metadata: dict[str, object] | None = None,
    on_event: "Callable[[object], None] | None" = None,
)

def _run_indexing_pipeline(
    *,
    source_path: Path,
    repo_id: str,
    db_path: str,
    batch_size: int,
    verbose: bool,
    extra_metadata: dict[str, object] | None = None,
    on_event: "Callable[[object], None] | None" = None,
) -> float:
    """Run the four-stage pipeline with atomic-write staging.

    Writes to ``<db_path>.staging`` (seeded from the live DB so prior repos are
    preserved) and atomically renames over ``<db_path>`` on success. Holds an
    exclusive flock on ``<db>.indexlock`` so two concurrent indexes can&#x27;t race
    the swap. *extra_metadata* is merged on top of the auto-collected metadata
    before persistence. Returns elapsed seconds.

    ``on_event`` (optional): invoked for every PipelineEvent yielded by the
    inner pipeline. Used by the serve.py /api/index-url worker to publish
    live progress (phase / current / total / nodes / edges) so the UI&#x27;s
    polling endpoint can report meaningful numbers instead of zeros.
    Exceptions raised in the callback are swallowed — observability must
    never crash the pipeline.
    """
    from opentrace_agent.pipeline import PipelineInput, run_pipeline
    from opentrace_agent.pipeline.adapters import GraphStoreAdapter
    from opentrace_agent.store import GraphStore

    db_dir = Path(db_path).parent
    db_dir.mkdir(parents=True, exist_ok=True)
    _ensure_gitignore(db_dir)

    # Staging file avoids contending with readers (MCP) holding the live DB lock.
    staging_db = db_path + ".staging"

    lock_fh = _acquire_index_lock(db_path)
    try:
        _clean_stale_staging(staging_db)

        click.echo(f"Opening staging database at {staging_db} ...")
        try:
            _seed_staging_from_live(db_path, staging_db)

            with GraphStore(staging_db) as graph_store:
                store = GraphStoreAdapter(graph_store, batch_size=batch_size)

                click.echo(f"Indexing {source_path} ...")
                t0 = time.monotonic()

                inp = PipelineInput(path=str(source_path), repo_id=repo_id)

                last_result = None
                for event in run_pipeline(inp, store=store):
                    _print_event(event, verbose)
                    if getattr(event, "result", None) is not None:
                        last_result = event.result
                    if on_event is not None: