sqlite_graphrag/commands/ingest/
run.rs

1//! Orchestration entry point for the `ingest` command.
2//!
3//! This module decides the order of the pipeline and owns nothing else. Each
4//! stage lives in its own sibling module:
5//!
6//! 1. [`super::validate`] — mode-conditional flag checks, before any I/O
7//! 2. [`super::scan_fs`] — walk the directory and match the glob
8//! 3. [`super::plan`] — resolve every memory name, single-threaded
9//! 4. [`super::dry_run`] — preview and stop, when `--dry-run` is set
10//! 5. [`super::stage_producer`] — Phase A: read, chunk, embed, extract
11//! 6. [`super::persist_loop`] — Phase B: write and report, as results arrive
12//! 7. [`super::enrich_after`] — optional post-ingest binding pass
13
14use super::args::IngestArgs;
15use super::persist::init_storage;
16use super::persist_loop::{self, PersistContext};
17use super::plan::build_plan;
18use super::scan_fs::collect_files;
19use super::validate::validate_mode_conditional_flags_ingest;
20use super::{dry_run, enrich_after, stage_producer};
21use crate::errors::AppError;
22use crate::paths::AppPaths;
23use std::path::PathBuf;
24
25/// Run the `ingest` command (filesystem scan + stage + persist, or mode adapters).
26pub fn run(args: IngestArgs, backends: crate::cli::BackendChoice) -> Result<(), AppError> {
27    // G20: mode-conditional flag validation BEFORE any DB access.
28    // Surfaces flags that the wrong mode would silently discard.
29    validate_mode_conditional_flags_ingest(&args)?;
30    // GAP-SG-215: `ingest` emits one record per file and a summary, so it runs
31    // under the stream contract — records unannotated, the `agent_surface` block
32    // once on the summary. The sample is empty on purpose: `ingest` mutates, so
33    // the gate's write fence returns before any question about field names is
34    // asked. Refusing here would report failure for files already persisted.
35    crate::agent_surface::stream::open(crate::agent_surface::get(), &[], 0)?;
36    tracing::debug!(target: "ingest", dir = %args.dir.display(), mode = ?args.mode, "starting ingest");
37    let started = std::time::Instant::now();
38    let files = scan(&args)?;
39    let total = files.len();
40
41    let namespace = crate::namespace::resolve_namespace(args.namespace.as_deref())?;
42    let memory_type_str = args.r#type.as_str().to_string();
43
44    let paths = AppPaths::resolve(args.db.as_deref())?;
45    // Storage failure is carried rather than raised: every file must still be
46    // reported as failed, with the same cause, instead of the run dying with
47    // no per-file record of what was lost.
48    let mut conn_or_err = init_storage(&paths).map_err(|e| format!("{e}"));
49
50    let plan = build_plan(&args, &files)?;
51
52    // --dry-run: preview and exit before loading any model or touching the DB.
53    if args.dry_run {
54        return dry_run::emit_preview(&args, &plan.slots_meta, total, started);
55    }
56
57    let parallelism = stage_producer::resolve_worker_count(&args)?;
58    stage_producer::validate_extraction_flags(&args)?;
59
60    let total_to_process = plan.process_items.len();
61    tracing::info!(
62        target: "ingest",
63        phase = "pipeline_start",
64        files = total_to_process,
65        ingest_parallelism = parallelism,
66        "incremental pipeline starting: Phase A (rayon) → channel → Phase B (main thread)",
67    );
68
69    let producer = stage_producer::spawn(&args, plan.process_items, &paths, parallelism, backends)?;
70
71    let ctx = PersistContext {
72        args: &args,
73        namespace: &namespace,
74        memory_type: &memory_type_str,
75        total,
76        started,
77    };
78    let tally = persist_loop::drain_and_persist(
79        &ctx,
80        &plan.slots_meta,
81        producer.results,
82        &mut conn_or_err,
83    )?;
84
85    producer
86        .handle
87        .join()
88        .map_err(|_| AppError::Internal(anyhow::anyhow!("ingest producer thread panicked")))?;
89
90    if let Ok(ref conn) = conn_or_err {
91        if tally.succeeded > 0 {
92            let _ = conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);");
93        }
94    }
95
96    persist_loop::emit_summary(&ctx, tally)?;
97
98    if args.enrich_after && tally.succeeded > 0 {
99        enrich_after::run(&args, backends)?;
100    }
101
102    Ok(())
103}
104
105/// Validates the target directory and returns the matched files, sorted.
106///
107/// # Errors
108/// Returns [`AppError::Validation`] when the directory is missing, is not a
109/// directory, or the match count exceeds `--max-files`.
110fn scan(args: &IngestArgs) -> Result<Vec<PathBuf>, AppError> {
111    if !args.dir.exists() {
112        return Err(AppError::Validation(
113            crate::i18n::validation::directory_not_found(&args.dir.display().to_string()),
114        ));
115    }
116    if !args.dir.is_dir() {
117        return Err(AppError::Validation(
118            crate::i18n::validation::not_a_directory(&args.dir.display().to_string()),
119        ));
120    }
121
122    let mut files: Vec<PathBuf> = Vec::with_capacity(128);
123    collect_files(&args.dir, &args.pattern, args.recursive, &mut files)?;
124    files.sort_unstable();
125
126    if files.len() > args.max_files {
127        return Err(AppError::Validation(
128            crate::i18n::validation::max_files_exceeded_matching(files.len(), args.max_files),
129        ));
130    }
131    Ok(files)
132}
sqlite_graphrag/commands/ingest/run.rs

sqlite_graphrag/commands/ingest/
run.rs
