From 5217b99601b2b7508daaa9d1b40383e63310893f Mon Sep 17 00:00:00 2001 From: Jacob Magar Date: Sun, 12 Apr 2026 07:23:33 -0400 Subject: [PATCH] feat(noxa-68r.8): daemon binary + README MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - noxa-rag-daemon: --config, --log-level, --version args (clap) - Startup sequence: config load -> watch_dir create-if-missing -> embed probe -> vector store -> tokenizer -> Pipeline::new() -> signal handlers -> run() - Embed dims probed dynamically (dummy embed call, no hardcode) - tokenizer.json loaded from local_path (Rust tokenizers crate has no from_pretrained) - Clear error message with huggingface-cli download command - Accepts directory or direct file path - Tracing to stderr (stdout may be piped) - World-readable config warning on unix - 10s force-exit timeout after CancellationToken cancel - SIGTERM + Ctrl-C via tokio signal handlers - README.md: CRITICAL TEI --pooling last-token warning, full config reference, quickstart, architecture diagram - LEARNED: tokenizers Rust crate has no from_pretrained — local_path is required --- crates/noxa-rag/README.md | 161 +++++++++++++++++ crates/noxa-rag/src/bin/noxa-rag-daemon.rs | 200 ++++++++++++++++++++- 2 files changed, 357 insertions(+), 4 deletions(-) create mode 100644 crates/noxa-rag/README.md diff --git a/crates/noxa-rag/README.md b/crates/noxa-rag/README.md new file mode 100644 index 0000000..e4bbf0e --- /dev/null +++ b/crates/noxa-rag/README.md @@ -0,0 +1,161 @@ +# noxa-rag + +RAG pipeline for [noxa](https://github.com/jmagar/noxa) — watches noxa's output directory for `ExtractionResult` JSON files, chunks them, embeds via [HF TEI](https://github.com/huggingface/text-embeddings-inference), and upserts to [Qdrant](https://qdrant.tech/). + +## System Requirements + +- **Qdrant** running locally (gRPC port 6334) +- **HF TEI** with GPU (tested on RTX 4070) +- **CUDA** for TEI inference (CPU mode is possible but slow) +- **Rust 1.82+** +- **huggingface-cli** to download the tokenizer + +## CRITICAL: TEI Launch Command + +```bash +# CRITICAL: --pooling last-token is REQUIRED for Qwen3-0.6B +# Qwen3 is a decoder-only model. Mean pooling (TEI default) produces +# semantically incorrect embeddings. This flag is NOT optional. +docker run --gpus all -p 8080:80 \ + ghcr.io/huggingface/text-embeddings-inference:latest \ + --model-id Qwen/Qwen3-Embedding-0.6B \ + --pooling last-token \ + --max-batch-tokens 32768 \ + --max-client-batch-size 128 \ + --dtype float16 +``` + +### Verify TEI is working + +```bash +curl http://localhost:8080/health +# {"status":"ok"} + +# Check embedding dimensions (must be 1024 for Qwen3-0.6B) +curl -s http://localhost:8080/embed \ + -H "Content-Type: application/json" \ + -d '{"inputs": ["test"], "normalize": true}' | python3 -c "import sys,json; v=json.load(sys.stdin)[0]; print(f'{len(v)} dims')" +# 1024 dims +``` + +## Quickstart + +### 1. Download the tokenizer + +The Rust `tokenizers` crate cannot download from HF Hub at runtime. Download once: + +```bash +pip install huggingface_hub +huggingface-cli download Qwen/Qwen3-Embedding-0.6B tokenizer.json --local-dir ~/.cache/noxa-rag/tokenizer +``` + +### 2. Create config file + +```toml +# noxa-rag.toml + +[source] +type = "fs_watcher" +watch_dir = "/home/user/.noxa/output" +debounce_ms = 500 + +[embed_provider] +type = "tei" +url = "http://localhost:8080" +model = "Qwen/Qwen3-Embedding-0.6B" +# REQUIRED: path to directory containing tokenizer.json +local_path = "/home/user/.cache/noxa-rag/tokenizer" + +[vector_store] +type = "qdrant" +# gRPC port 6334 (NOT 6333 which is REST) +url = "http://localhost:6334" +collection = "noxa_rag" +# api_key = "..." # or set NOXA_RAG_QDRANT_API_KEY env var + +[chunker] +target_tokens = 512 +overlap_tokens = 64 +min_words = 50 +max_chunks_per_page = 100 + +[pipeline] +embed_concurrency = 4 +# Must be an absolute path (daemon may run with CWD = /) +failed_jobs_log = "/home/user/.noxa/noxa-rag-failed.jsonl" +``` + +### 3. Start Qdrant + +```bash +docker run -p 6333:6333 -p 6334:6334 \ + -v ~/.noxa/qdrant:/qdrant/storage \ + qdrant/qdrant +``` + +### 4. Run the daemon + +```bash +cargo build --release -p noxa-rag +./target/release/noxa-rag-daemon --config noxa-rag.toml +``` + +### 5. Index content with noxa + +```bash +# Extract a page — the daemon will pick up the output file automatically +noxa https://docs.example.com --output ~/.noxa/output/ +``` + +The daemon watches `watch_dir` for `.json` files. When noxa writes an `ExtractionResult` to that directory, the daemon detects it (within `debounce_ms` ms), chunks it, embeds it, and upserts to Qdrant. + +## Configuration Reference + +| Field | Default | Description | +|-------|---------|-------------| +| `source.watch_dir` | — | Directory to watch for `.json` files | +| `source.debounce_ms` | `500` | Debounce window for filesystem events (ms) | +| `embed_provider.url` | — | TEI server URL | +| `embed_provider.model` | — | Model name (used in logs) | +| `embed_provider.local_path` | **required** | Directory containing `tokenizer.json` | +| `vector_store.url` | — | Qdrant gRPC URL (port 6334) | +| `vector_store.collection` | — | Qdrant collection name | +| `vector_store.api_key` | `null` | Qdrant API key (or `NOXA_RAG_QDRANT_API_KEY` env var) | +| `chunker.target_tokens` | `512` | Target chunk size in tokens | +| `chunker.overlap_tokens` | `64` | Sliding window overlap tokens | +| `chunker.min_words` | `50` | Skip chunks shorter than this | +| `chunker.max_chunks_per_page` | `100` | Cap chunks per document | +| `pipeline.embed_concurrency` | `4` | Concurrent embed workers (must be > 0) | +| `pipeline.failed_jobs_log` | `null` | Absolute path for NDJSON error log | + +## Architecture + +``` +noxa-cli (writes .json) → watch_dir + ↓ + notify-debouncer-mini (500ms debounce) + ↓ + bounded mpsc channel (256 capacity) + ↓ + embed_concurrency worker tasks (default: 4) + ↓ + ┌─────────────────────────────────────┐ + │ process_job() │ + │ 1. Read file (TOCTOU-safe) │ + │ 2. Parse ExtractionResult JSON │ + │ 3. Validate URL scheme (http/https) │ + │ 4. chunk() → Vec │ + │ 5. embed() → Vec> │ + │ 6. UUID v5 point IDs │ + │ 7. Per-URL mutex: delete + upsert │ + └─────────────────────────────────────┘ + ↓ + Qdrant (gRPC) +``` + +## Notes + +- **Vim/Emacs compatibility**: The daemon watches all filesystem events (not just Create/Modify). Atomic saves via rename are detected correctly. +- **Idempotent indexing**: Re-indexing the same URL deletes old chunks first (delete-before-upsert), so chunk count changes are handled correctly. +- **Point IDs**: UUID v5 deterministic — same URL + chunk index always produces the same Qdrant point ID. +- **Failed jobs**: Parse failures and oversized files (>50MB) are logged to `failed_jobs_log` as NDJSON and skipped (the daemon keeps running). diff --git a/crates/noxa-rag/src/bin/noxa-rag-daemon.rs b/crates/noxa-rag/src/bin/noxa-rag-daemon.rs index 6f8c76e..4241d49 100644 --- a/crates/noxa-rag/src/bin/noxa-rag-daemon.rs +++ b/crates/noxa-rag/src/bin/noxa-rag-daemon.rs @@ -1,5 +1,197 @@ -// noxa-rag-daemon — implemented in noxa-68r.8 -fn main() { - eprintln!("[noxa-rag] daemon stub — not yet implemented (noxa-68r.8)"); - std::process::exit(1); +/// noxa-rag-daemon — watches an output directory for ExtractionResult JSON files +/// and indexes them via TEI + Qdrant. +/// +/// Usage: +/// noxa-rag-daemon [--config ] [--log-level ] [--version] +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Duration; + +use clap::Parser; +use tokio_util::sync::CancellationToken; +use tracing_subscriber::EnvFilter; + +use noxa_rag::{ + build_embed_provider, build_vector_store, load_config, + config::{EmbedProviderConfig, SourceConfig}, + pipeline::Pipeline, +}; + +#[derive(Parser)] +#[command(name = "noxa-rag-daemon", about = "noxa RAG indexing daemon")] +struct Args { + /// Config file path + #[arg(long, default_value = "noxa-rag.toml")] + config: PathBuf, + + /// Log level (overrides RUST_LOG) + #[arg(long, default_value = "info")] + log_level: String, + + /// Print version and exit + #[arg(long)] + version: bool, +} + +#[tokio::main] +async fn main() { + let args = Args::parse(); + + if args.version { + println!("noxa-rag-daemon {}", env!("CARGO_PKG_VERSION")); + std::process::exit(0); + } + + // Init tracing to stderr (stdout may be piped). + tracing_subscriber::fmt() + .with_env_filter( + EnvFilter::try_from_default_env() + .unwrap_or_else(|_| EnvFilter::new(&args.log_level)), + ) + .with_writer(std::io::stderr) + .init(); + + if let Err(e) = run(args).await { + eprintln!("[noxa-rag] fatal: {e}"); + std::process::exit(1); + } +} + +async fn run(args: Args) -> Result<(), Box> { + let config_path = &args.config; + + // Warn if config file is world-readable (may contain api_key). + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + if let Ok(meta) = std::fs::metadata(config_path) { + let mode = meta.permissions().mode(); + if mode & 0o004 != 0 { + eprintln!( + "[noxa-rag] WARNING: config file is world-readable (mode {:o}). \ + Consider: chmod 600 {}", + mode, + config_path.display() + ); + } + } + } + + // Load config — fail fast with clear error. + let config = load_config(config_path).map_err(|e| { + format!("failed to load config from {}: {e}", config_path.display()) + })?; + + // Ensure watch_dir exists (create if missing — convenience for first-run). + let watch_dir = match &config.source { + SourceConfig::FsWatcher { watch_dir, .. } => watch_dir.clone(), + }; + + if !watch_dir.exists() { + std::fs::create_dir_all(&watch_dir).map_err(|e| { + format!( + "watch_dir does not exist and could not be created ({}): {e}", + watch_dir.display() + ) + })?; + eprintln!("[noxa-rag] created watch_dir: {}", watch_dir.display()); + } + + // Build embed provider — startup probe (exits 1 if TEI unavailable). + let embed = build_embed_provider(&config) + .await + .map_err(|e| format!("embed provider startup failed: {e}"))?; + + // Probe embed dims dynamically by calling embed with a dummy text. + // This avoids hardcoding and reuses the already-built provider. + let probe_vecs = embed + .embed(&["probe".to_string()]) + .await + .map_err(|e| format!("embed dims probe failed: {e}"))?; + let embed_dims = probe_vecs + .first() + .map(|v| v.len()) + .ok_or("embed probe returned empty vector")?; + + // Build vector store — collection create/validate. + let store = build_vector_store(&config, embed_dims) + .await + .map_err(|e| format!("vector store startup failed: {e}"))?; + + // Load tokenizer. + let tokenizer_model = match &config.embed_provider { + EmbedProviderConfig::Tei { model, local_path, .. } => { + (model.clone(), local_path.clone()) + } + _ => ("Qwen/Qwen3-Embedding-0.6B".to_string(), None), + }; + + // Rust tokenizers crate has no from_pretrained — local_path is required. + // Download tokenizer.json from HF Hub before running: + // huggingface-cli download Qwen/Qwen3-Embedding-0.6B tokenizer.json --local-dir ./ + let tokenizer = { + let path = tokenizer_model.1.ok_or_else(|| { + format!( + "embed_provider.local_path is required — the Rust tokenizers crate cannot \ + download from HF Hub. Set local_path to the directory containing tokenizer.json.\n\ + Download: huggingface-cli download {} tokenizer.json --local-dir ", + tokenizer_model.0 + ) + })?; + // If given a directory, look for tokenizer.json inside it. + let tokenizer_file = if path.is_dir() { + path.join("tokenizer.json") + } else { + path.clone() + }; + tokenizers::Tokenizer::from_file(&tokenizer_file) + .map_err(|e| format!("failed to load tokenizer from {}: {e}", tokenizer_file.display()))? + }; + + eprintln!("[noxa-rag] tokenizer: {} — loaded", tokenizer_model.0); + + let shutdown = CancellationToken::new(); + let pipeline = Pipeline::new( + config, + embed, + store, + Arc::new(tokenizer), + shutdown.clone(), + ); + + // Signal handling: Ctrl-C + SIGTERM -> cancel. + let shutdown_signal = shutdown.clone(); + tokio::spawn(async move { + #[cfg(unix)] + { + use tokio::signal::unix::{signal, SignalKind}; + let mut sigterm = signal(SignalKind::terminate()) + .expect("failed to register SIGTERM handler"); + tokio::select! { + _ = tokio::signal::ctrl_c() => {} + _ = sigterm.recv() => {} + } + } + #[cfg(not(unix))] + { + let _ = tokio::signal::ctrl_c().await; + } + eprintln!("[noxa-rag] shutdown signal received"); + shutdown_signal.cancel(); + }); + + eprintln!("[noxa-rag] daemon started"); + + // Run pipeline with 10s force-exit timeout after cancellation. + match tokio::time::timeout(Duration::from_secs(10), pipeline.run()).await { + Ok(Ok(())) => {} + Ok(Err(e)) => eprintln!("[noxa-rag] pipeline error: {e}"), + Err(_elapsed) => { + eprintln!("[noxa-rag] WARNING: pipeline did not shut down within 10s, forcing exit"); + std::process::exit(0); + } + } + + eprintln!("[noxa-rag] daemon stopped"); + Ok(()) }