fix: all tests now in sync

ingestion-pipeline/src/pipeline/config.rs

@@ -6,8 +6,8 @@ pub struct IngestionTuning {
     pub graph_store_attempts: usize,
     pub graph_initial_backoff_ms: u64,
     pub graph_max_backoff_ms: u64,
-    pub chunk_min_chars: usize,
-    pub chunk_max_chars: usize,
+    pub chunk_min_tokens: usize,
+    pub chunk_max_tokens: usize,
     pub chunk_insert_concurrency: usize,
     pub entity_embedding_concurrency: usize,
 }
@@ -21,15 +21,25 @@ impl Default for IngestionTuning {
             graph_store_attempts: 3,
             graph_initial_backoff_ms: 50,
             graph_max_backoff_ms: 800,
-            chunk_min_chars: 500,
-            chunk_max_chars: 2_000,
+            chunk_min_tokens: 500,
+            chunk_max_tokens: 2_000,
             chunk_insert_concurrency: 8,
             entity_embedding_concurrency: 4,
         }
     }
 }
 
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone)]
 pub struct IngestionConfig {
     pub tuning: IngestionTuning,
+    pub chunk_only: bool,
+}
+
+impl Default for IngestionConfig {
+    fn default() -> Self {
+        Self {
+            tuning: IngestionTuning::default(),
+            chunk_only: false,
+        }
+    }
 }

ingestion-pipeline/src/pipeline/context.rs

@@ -101,6 +101,10 @@ impl<'a> PipelineContext<'a> {
     }
 
     pub async fn build_artifacts(&mut self) -> Result<PipelineArtifacts, AppError> {
+        if self.pipeline_config.chunk_only {
+            return self.build_chunk_only_artifacts().await;
+        }
+
         let content = self.take_text_content()?;
         let analysis = self.take_analysis()?;
 
@@ -113,8 +117,7 @@ impl<'a> PipelineContext<'a> {
             )
             .await?;
 
-        let chunk_range: Range<usize> = self.pipeline_config.tuning.chunk_min_chars
-            ..self.pipeline_config.tuning.chunk_max_chars;
+        let chunk_range = self.chunk_token_range();
 
         let chunks = self.services.prepare_chunks(&content, chunk_range).await?;
 
@@ -125,4 +128,22 @@ impl<'a> PipelineContext<'a> {
             chunks,
         })
     }
+
+    pub async fn build_chunk_only_artifacts(&mut self) -> Result<PipelineArtifacts, AppError> {
+        let content = self.take_text_content()?;
+        let chunk_range = self.chunk_token_range();
+
+        let chunks = self.services.prepare_chunks(&content, chunk_range).await?;
+
+        Ok(PipelineArtifacts {
+            text_content: content,
+            entities: Vec::new(),
+            relationships: Vec::new(),
+            chunks,
+        })
+    }
+
+    fn chunk_token_range(&self) -> Range<usize> {
+        self.pipeline_config.tuning.chunk_min_tokens..self.pipeline_config.tuning.chunk_max_tokens
+    }
 }

ingestion-pipeline/src/pipeline/mod.rs

@@ -51,6 +51,27 @@ impl IngestionPipeline {
         reranker_pool: Option<Arc<RerankerPool>>,
         storage: StorageManager,
         embedding_provider: Arc<common::utils::embedding::EmbeddingProvider>,
+    ) -> Result<Self, AppError> {
+        Self::new_with_config(
+            db,
+            openai_client,
+            config,
+            reranker_pool,
+            storage,
+            embedding_provider,
+            IngestionConfig::default(),
+        )
+        .await
+    }
+
+    pub async fn new_with_config(
+        db: Arc<SurrealDbClient>,
+        openai_client: Arc<Client<async_openai::config::OpenAIConfig>>,
+        config: AppConfig,
+        reranker_pool: Option<Arc<RerankerPool>>,
+        storage: StorageManager,
+        embedding_provider: Arc<common::utils::embedding::EmbeddingProvider>,
+        pipeline_config: IngestionConfig,
     ) -> Result<Self, AppError> {
         let services = DefaultPipelineServices::new(
             db.clone(),
@@ -61,7 +82,7 @@ impl IngestionPipeline {
             embedding_provider,
         );
 
-        Self::with_services(db, IngestionConfig::default(), Arc::new(services))
+        Self::with_services(db, pipeline_config, Arc::new(services))
     }
 
     pub fn with_services(

ingestion-pipeline/src/pipeline/services.rs

@@ -21,7 +21,6 @@ use common::{
     utils::{config::AppConfig, embedding::EmbeddingProvider},
 };
 use retrieval_pipeline::{reranking::RerankerPool, retrieved_entities_to_json, RetrievedEntity};
-use text_splitter::TextSplitter;
 
 use super::{enrichment_result::LLMEnrichmentResult, preparation::to_text_content};
 use crate::pipeline::context::{EmbeddedKnowledgeEntity, EmbeddedTextChunk};
@@ -59,7 +58,7 @@ pub trait PipelineServices: Send + Sync {
     async fn prepare_chunks(
         &self,
         content: &TextContent,
-        range: Range<usize>,
+        token_range: Range<usize>,
     ) -> Result<Vec<EmbeddedTextChunk>, AppError>;
 }
 
@@ -238,23 +237,20 @@ impl PipelineServices for DefaultPipelineServices {
     async fn prepare_chunks(
         &self,
         content: &TextContent,
-        range: Range<usize>,
+        token_range: Range<usize>,
     ) -> Result<Vec<EmbeddedTextChunk>, AppError> {
-        let splitter = TextSplitter::new(range.clone());
-        let chunk_texts: Vec<String> = splitter
-            .chunks(&content.text)
-            .map(|chunk| chunk.to_string())
-            .collect();
+        let chunk_candidates =
+            split_by_token_bounds(&content.text, token_range.start, token_range.end)?;
 
-        let mut chunks = Vec::with_capacity(chunk_texts.len());
-        for chunk in chunk_texts {
+        let mut chunks = Vec::with_capacity(chunk_candidates.len());
+        for chunk_text in chunk_candidates {
             let embedding = self
                 .embedding_provider
-                .embed(&chunk)
+                .embed(&chunk_text)
                 .await
                 .context("generating FastEmbed embedding for chunk")?;
             let chunk_struct =
-                TextChunk::new(content.get_id().to_string(), chunk, content.user_id.clone());
+                TextChunk::new(content.get_id().to_string(), chunk_text, content.user_id.clone());
             chunks.push(EmbeddedTextChunk {
                 chunk: chunk_struct,
                 embedding,
@@ -264,6 +260,45 @@ impl PipelineServices for DefaultPipelineServices {
     }
 }
 
+fn split_by_token_bounds(
+    text: &str,
+    min_tokens: usize,
+    max_tokens: usize,
+) -> Result<Vec<String>, AppError> {
+    if min_tokens == 0 || max_tokens == 0 || min_tokens > max_tokens {
+        return Err(AppError::Validation(
+            "invalid chunk token bounds; ensure 0 < min <= max".into(),
+        ));
+    }
+
+    let tokens: Vec<&str> = text.split_whitespace().collect();
+    if tokens.is_empty() {
+        return Ok(vec![String::new()]);
+    }
+
+    let mut chunks = Vec::new();
+    let mut buffer: Vec<&str> = Vec::new();
+    for (idx, token) in tokens.iter().enumerate() {
+        buffer.push(token);
+        let remaining = tokens.len().saturating_sub(idx + 1);
+        let at_max = buffer.len() >= max_tokens;
+        let at_min_and_boundary =
+            buffer.len() >= min_tokens && (remaining == 0 || buffer.len() + 1 > max_tokens);
+        if at_max || at_min_and_boundary {
+            let chunk_text = buffer.join(" ");
+            chunks.push(chunk_text);
+            buffer.clear();
+        }
+    }
+
+    if !buffer.is_empty() {
+        let chunk_text = buffer.join(" ");
+        chunks.push(chunk_text);
+    }
+
+    Ok(chunks)
+}
+
 fn truncate_for_embedding(text: &str, max_chars: usize) -> String {
     if text.chars().count() <= max_chars {
         return text.to_string();

ingestion-pipeline/src/pipeline/stages/mod.rs

@@ -16,6 +16,7 @@ use tracing::{debug, instrument, warn};
 
 use super::{
     context::{EmbeddedKnowledgeEntity, EmbeddedTextChunk, PipelineArtifacts, PipelineContext},
+    enrichment_result::LLMEnrichmentResult,
     state::{ContentPrepared, Enriched, IngestionMachine, Persisted, Ready, Retrieved},
 };
 
@@ -76,6 +77,12 @@ pub async fn retrieve_related(
     machine: IngestionMachine<(), ContentPrepared>,
     ctx: &mut PipelineContext<'_>,
 ) -> Result<IngestionMachine<(), Retrieved>, AppError> {
+    if ctx.pipeline_config.chunk_only {
+        return machine
+            .retrieve()
+            .map_err(|(_, guard)| map_guard_error("retrieve", guard));
+    }
+
     let content = ctx.text_content()?;
     let similar = ctx.services.retrieve_similar_entities(content).await?;
 
@@ -102,6 +109,16 @@ pub async fn enrich(
     machine: IngestionMachine<(), Retrieved>,
     ctx: &mut PipelineContext<'_>,
 ) -> Result<IngestionMachine<(), Enriched>, AppError> {
+    if ctx.pipeline_config.chunk_only {
+        ctx.analysis = Some(LLMEnrichmentResult {
+            knowledge_entities: Vec::new(),
+            relationships: Vec::new(),
+        });
+        return machine
+            .enrich()
+            .map_err(|(_, guard)| map_guard_error("enrich", guard));
+    }
+
     let content = ctx.text_content()?;
     let analysis = ctx
         .services

ingestion-pipeline/src/pipeline/tests.rs

@@ -212,9 +212,9 @@ impl PipelineServices for FailingServices {
     async fn prepare_chunks(
         &self,
         content: &TextContent,
-        range: std::ops::Range<usize>,
+        token_range: std::ops::Range<usize>,
     ) -> Result<Vec<EmbeddedTextChunk>, AppError> {
-        self.inner.prepare_chunks(content, range).await
+        self.inner.prepare_chunks(content, token_range).await
     }
 }
 
@@ -254,7 +254,7 @@ impl PipelineServices for ValidationServices {
     async fn prepare_chunks(
         &self,
         _content: &TextContent,
-        _range: std::ops::Range<usize>,
+        _token_range: std::ops::Range<usize>,
     ) -> Result<Vec<EmbeddedTextChunk>, AppError> {
         unreachable!("prepare_chunks should not be called after validation failure")
     }
@@ -275,12 +275,13 @@ async fn setup_db() -> SurrealDbClient {
 fn pipeline_config() -> IngestionConfig {
     IngestionConfig {
         tuning: IngestionTuning {
-            chunk_min_chars: 4,
-            chunk_max_chars: 64,
+            chunk_min_tokens: 4,
+            chunk_max_tokens: 64,
             chunk_insert_concurrency: 4,
             entity_embedding_concurrency: 2,
             ..IngestionTuning::default()
         },
+        chunk_only: false,
     }
 }
 
@@ -362,6 +363,69 @@ async fn ingestion_pipeline_happy_path_persists_entities() {
     assert!(call_log[4..].iter().all(|entry| *entry == "chunk"));
 }
 
+#[tokio::test]
+async fn ingestion_pipeline_chunk_only_skips_analysis() {
+    let db = setup_db().await;
+    let worker_id = "worker-chunk-only";
+    let user_id = "user-999";
+    let services = Arc::new(MockServices::new(user_id));
+    let mut config = pipeline_config();
+    config.chunk_only = true;
+    let pipeline =
+        IngestionPipeline::with_services(Arc::new(db.clone()), config, services.clone())
+            .expect("pipeline");
+
+    let task = reserve_task(
+        &db,
+        worker_id,
+        IngestionPayload::Text {
+            text: "Chunk only payload".into(),
+            context: "Context".into(),
+            category: "notes".into(),
+            user_id: user_id.into(),
+        },
+        user_id,
+    )
+    .await;
+
+    pipeline
+        .process_task(task.clone())
+        .await
+        .expect("pipeline succeeds");
+
+    let stored_entities: Vec<KnowledgeEntity> = db
+        .get_all_stored_items::<KnowledgeEntity>()
+        .await
+        .expect("entities stored");
+    assert!(
+        stored_entities.is_empty(),
+        "chunk-only ingestion should not persist entities"
+    );
+    let relationship_count: Option<i64> = db
+        .client
+        .query("SELECT count() as count FROM relates_to;")
+        .await
+        .expect("query relationships")
+        .take::<Option<i64>>(0)
+        .unwrap_or_default();
+    assert_eq!(
+        relationship_count.unwrap_or(0),
+        0,
+        "chunk-only ingestion should not persist relationships"
+    );
+    let stored_chunks: Vec<TextChunk> = db
+        .get_all_stored_items::<TextChunk>()
+        .await
+        .expect("chunks stored");
+    assert!(
+        !stored_chunks.is_empty(),
+        "chunk-only ingestion should still persist chunks"
+    );
+
+    let call_log = services.calls.lock().await.clone();
+    assert_eq!(call_log, vec!["prepare", "chunk"]);
+}
+
 #[tokio::test]
 async fn ingestion_pipeline_failure_marks_retry() {
     let db = setup_db().await;