sqlite_graphrag/commands/remember/
run.rs

1//! Orchestration entry point for the `remember` command.
2
3use super::args::RememberArgs;
4use super::graph_input::normalize_and_validate_graph_input;
5use crate::chunking;
6use crate::entity_type::DEFAULT_ENTITY_TYPE;
7use crate::errors::AppError;
8use crate::i18n::errors_msg;
9use crate::output;
10use crate::paths::AppPaths;
11use crate::storage::chunks as storage_chunks;
12use crate::storage::connection::{ensure_schema, open_rw};
13use crate::storage::entities::{self as entities, NewEntity};
14use crate::storage::memories::{self as memories, NewMemory};
15use crate::storage::versions;
16
17/// Run the `remember` command: validate, embed, persist memory + graph.
18pub fn run(args: RememberArgs, backends: crate::cli::BackendChoice) -> Result<(), AppError> {
19    let crate::cli::BackendChoice {
20        llm: llm_backend,
21        embedding: embedding_backend,
22    } = backends;
23    use crate::constants::*;
24
25    let started = std::time::Instant::now();
26    let _ = args.format;
27    tracing::debug!(
28        target: "remember",
29        name = ?args.name_positional.as_deref().or(args.name.as_deref()),
30        "persisting memory"
31    );
32    let namespace = crate::namespace::resolve_namespace(args.namespace.as_deref())?;
33
34    let resolved_name = super::name::resolve(&args)?;
35    let original_name = resolved_name.original;
36    let normalized_name = resolved_name.normalized;
37    let name_was_normalized = resolved_name.was_normalized;
38
39    let resolved_input = super::input::resolve(&args)?;
40    let mut raw_body = resolved_input.raw_body;
41    let mut graph = resolved_input.graph;
42    let entities_provided_externally = resolved_input.entities_provided_externally;
43    let mut relationships_updated = resolved_input.relationships_updated;
44
45    // GAP-SG-216: refuse a label outside the canonical set HERE, before
46    // `AppPaths::resolve`
47    // touches the filesystem and long before `open_rw`. Nothing has been written,
48    // so the caller can fix the payload and re-run with no state to undo — the
49    // same reason GAP-SG-215 moved the stream gate ahead of the first byte.
50    if args.strict_entity_types && !graph.type_warnings.is_empty() {
51        return Err(AppError::Validation(
52            crate::i18n::validation::strict_entity_type_folded(&graph.type_warnings),
53        ));
54    }
55
56    // v1.1.2 (Gap 2): boundary validation of BOTH payload ceilings — bytes
57    // (BodyTooLarge) and estimated tokens (TooManyTokens), exit 6 — reusing
58    // the same guard the REST embedding client keeps as defence in depth.
59    // The token cap used to fire only deep inside the embedding call.
60    crate::memory_guard::check_embedding_input_size(&raw_body)?;
61
62    // v1.0.22 P1: reject empty or whitespace-only body when no external graph is provided.
63    // Without this check, empty embeddings would be persisted, breaking recall semantics.
64    // GAP-08: skip this guard when --force-merge without --clear-body; the existing body
65    // will be preserved from the database, so the effective body will not be empty.
66    let body_will_be_preserved = args.force_merge && raw_body.trim().is_empty() && !args.clear_body;
67    if !entities_provided_externally
68        && graph.entities.is_empty()
69        && raw_body.trim().is_empty()
70        && !body_will_be_preserved
71        && !args.clear_body
72    {
73        return Err(AppError::Validation(crate::i18n::validation::empty_body()));
74    }
75
76    let metadata: serde_json::Value = if let Some(ref m) = args.metadata {
77        serde_json::from_str(m)?
78    } else if let Some(ref path) = args.metadata_file {
79        let file_size = std::fs::metadata(path).map_err(AppError::Io)?.len();
80        if file_size > MAX_MEMORY_BODY_LEN as u64 {
81            return Err(AppError::BodyTooLarge {
82                bytes: file_size,
83                limit: MAX_MEMORY_BODY_LEN as u64,
84            });
85        }
86        let content = std::fs::read_to_string(path).map_err(AppError::Io)?;
87        serde_json::from_str(&content)?
88    } else {
89        serde_json::json!({})
90    };
91
92    let mut body_hash = blake3::hash(raw_body.as_bytes()).to_hex().to_string();
93    let mut snippet: String = raw_body.chars().take(200).collect();
94
95    let paths = AppPaths::resolve(args.db.as_deref())?;
96    paths.ensure_dirs()?;
97
98    // v1.0.20: use .trim().is_empty() to reject bodies that are only whitespace.
99    let mut extraction_method: Option<String> = None;
100    let mut extracted_urls: Vec<crate::extraction::ExtractedUrl> = Vec::with_capacity(4);
101    if args.enable_ner && args.skip_extraction {
102        return Err(AppError::Validation(
103            crate::i18n::validation::enable_ner_skip_extraction_exclusive(),
104        ));
105    }
106    if args.skip_extraction && !args.enable_ner {
107        // v1.0.74: revert to v1.0.45 hidden no-op behavior. The v1.0.67
108        // commit (9ddb17b) promoted this to a hard validation error, which
109        // broke the "kept as a hidden no-op for backwards compatibility"
110        // promise documented in CHANGELOG v1.0.45 and started failing
111        // 5+ CI jobs whose E2E tests use this flag to skip the
112        // (since-removed) GLiNER-ONNX model download in CI environments.
113        tracing::warn!(
114            "--skip-extraction is deprecated since v1.0.45 and has no effect (NER is disabled by default); remove this flag to silence the warning"
115        );
116    }
117    if args.enable_ner && graph.entities.is_empty() && !raw_body.trim().is_empty() {
118        match crate::extraction::extract_graph_auto(&raw_body, &paths) {
119            Ok(extracted) => {
120                // v1.0.76: ExtractionResult is URL + entity + elapsed_ms;
121                // the LLM ExtractionBackend returns typed relationships
122                // separately. The default build is URL-only extraction.
123                extraction_method = Some("url-regex".to_string());
124                extracted_urls = extracted.urls;
125                // Convert ExtractedEntity → NewEntity (no offsets; the URL
126                // extractor declares no type, so the default applies).
127                graph.entities = extracted
128                    .entities
129                    .into_iter()
130                    .map(|e| NewEntity {
131                        name: e.name,
132                        entity_type: DEFAULT_ENTITY_TYPE.to_string(),
133                        description: None,
134                    })
135                    .collect();
136                graph.relationships.clear();
137                relationships_updated = false;
138
139                if graph.entities.len() > max_entities_per_memory() {
140                    graph.entities.truncate(max_entities_per_memory());
141                }
142                if graph.relationships.len() > max_relationships_per_memory() {
143                    relationships_updated = true;
144                    graph.relationships.truncate(max_relationships_per_memory());
145                }
146                normalize_and_validate_graph_input(&mut graph)?;
147            }
148            Err(e) => {
149                tracing::warn!(target: "remember", error = %e, "auto-extraction failed, graceful degradation");
150                extraction_method = Some("none:extraction-failed".to_string());
151            }
152        }
153    }
154
155    let mut conn = open_rw(&paths.db)?;
156    ensure_schema(&mut conn)?;
157
158    // --dry-run: emit planned action without any DB writes and return.
159    if args.dry_run {
160        let existing = memories::find_by_name(&conn, &namespace, &normalized_name)?;
161        let planned_action = if existing.is_some() && args.force_merge {
162            "would_update"
163        } else {
164            "would_create"
165        };
166        // GAP-SG-216: report what the validation actually learned. Until v1.2.8
167        // this envelope carried four fields and dropped every finding the parse
168        // had already produced — a run with three unusual `entity_type` labels
169        // emitted `warnings: null`, so the one mode whose entire purpose is
170        // "tell me what would happen" was the one mode that would not say.
171        // `purge` is the house pattern: its schema declares `dry_run` as a
172        // required member of the SAME envelope the real run emits.
173        output::emit_json(&serde_json::json!({
174            "dry_run": true,
175            "name": normalized_name,
176            "namespace": namespace,
177            "planned_action": planned_action,
178            "entities_parsed": graph.entities.len(),
179            "relationships_parsed": graph.relationships.len(),
180            "warnings": graph.type_warnings,
181        }))?;
182        return Ok(());
183    }
184
185    {
186        use crate::constants::MAX_NAMESPACES_ACTIVE;
187        let active_count: u32 = conn.query_row(
188            "SELECT COUNT(DISTINCT namespace) FROM memories WHERE deleted_at IS NULL",
189            [],
190            |r| r.get::<_, i64>(0).map(|v| v as u32),
191        )?;
192        let ns_exists: bool = conn.query_row(
193            "SELECT EXISTS(SELECT 1 FROM memories WHERE namespace = ?1 AND deleted_at IS NULL)",
194            rusqlite::params![namespace],
195            |r| r.get::<_, i64>(0).map(|v| v > 0),
196        )?;
197        if !ns_exists && active_count >= MAX_NAMESPACES_ACTIVE {
198            return Err(AppError::NamespaceError(
199                crate::i18n::errors_ops::active_namespace_limit_reached(
200                    MAX_NAMESPACES_ACTIVE,
201                    &namespace,
202                ),
203            ));
204        }
205    }
206
207    // M7: detect soft-deleted memory before the standard duplicate check.
208    if let Some((sd_id, true)) =
209        memories::find_by_name_any_state(&conn, &namespace, &normalized_name)?
210    {
211        if args.force_merge {
212            memories::clear_deleted_at(&conn, sd_id)?;
213        } else {
214            return Err(AppError::Duplicate(
215                errors_msg::duplicate_memory_soft_deleted(&normalized_name, &namespace),
216            ));
217        }
218    }
219
220    let existing_memory = memories::find_by_name(&conn, &namespace, &normalized_name)?;
221    if existing_memory.is_some() && !args.force_merge {
222        return Err(AppError::Duplicate(errors_msg::duplicate_memory(
223            &normalized_name,
224            &namespace,
225        )));
226    }
227
228    // GAP-10: resolve type and description.
229    // For CREATE path (new memory): both are required.
230    // For UPDATE path (--force-merge on existing memory): inherit from existing row when omitted.
231    let (resolved_type, resolved_description) = if existing_memory.is_none() {
232        // CREATE path — both fields are mandatory.
233        let t = args.r#type.ok_or_else(|| {
234            AppError::Validation(crate::i18n::validation::type_and_description_required())
235        })?;
236        let d = args.description.clone().ok_or_else(|| {
237            AppError::Validation(crate::i18n::validation::type_and_description_required())
238        })?;
239        (t.as_str().to_string(), d)
240    } else {
241        // UPDATE path (--force-merge) — inherit missing fields from stored row.
242        let existing_row = memories::read_by_name(&conn, &namespace, &normalized_name)?
243            .ok_or_else(|| {
244                AppError::NotFound(
245                    crate::i18n::validation::memory_named_not_found_in_namespace(
246                        &normalized_name,
247                        &namespace,
248                    ),
249                )
250            })?;
251        let t = args
252            .r#type
253            .map(|v| v.as_str().to_string())
254            .unwrap_or_else(|| existing_row.memory_type.clone());
255        let d = args
256            .description
257            .clone()
258            .unwrap_or_else(|| existing_row.description.clone());
259        (t, d)
260    };
261
262    // GAP-08/GAP-09: protect existing body from accidental destruction during --force-merge.
263    // When the caller omits a body (or passes an empty one) without --clear-body, silently
264    // preserve the existing body from the database.  This prevents a common scripting mistake
265    // where a cron job updates metadata fields and inadvertently wipes the stored content.
266    if body_will_be_preserved {
267        if let Some(existing_row) = memories::read_by_name(&conn, &namespace, &normalized_name)? {
268            if !existing_row.body.is_empty() {
269                tracing::debug!(target: "remember",
270                    name = %normalized_name,
271                    "GAP-08: empty body with --force-merge and no --clear-body; preserving existing body"
272                );
273                raw_body = existing_row.body;
274                body_hash = blake3::hash(raw_body.as_bytes()).to_hex().to_string();
275                snippet = raw_body.chars().take(200).collect();
276            }
277        }
278    }
279
280    let duplicate_hash_id = memories::find_by_hash(&conn, &namespace, &body_hash)?;
281
282    output::emit_progress_i18n(
283        &format!(
284            "Remember stage: validated input; available memory {} MB",
285            crate::memory_guard::available_memory_mb()
286        ),
287        &format!(
288            "Stage remember: input validated; available memory {} MB",
289            crate::memory_guard::available_memory_mb()
290        ),
291    );
292
293    let model_max_length = crate::tokenizer::get_model_max_length();
294    let total_passage_tokens = crate::tokenizer::count_passage_tokens(&raw_body)?;
295    let chunks_info = chunking::split_into_chunks_hierarchical(&raw_body);
296    let chunks_created = chunks_info.len();
297    // GAP-SG-40: `chunks_persisted` is no longer a pre-commit estimate. It is
298    // read back from `memory_chunks` AFTER the transaction commits (see below)
299    // so the reported count matches the observable database state. Single-chunk
300    // bodies store inline in the memories row and append no chunk rows.
301
302    output::emit_progress_i18n(
303        &format!(
304            "Remember stage: tokenizer counted {total_passage_tokens} passage tokens (model max {model_max_length}); chunking produced {} chunks; process RSS {} MB",
305            chunks_created,
306            crate::memory_guard::current_process_memory_mb().unwrap_or(0)
307        ),
308        &format!(
309            "Stage remember: tokenizer counted {total_passage_tokens} passage tokens (model max {model_max_length}); chunking produced {} chunks; process RSS {} MB",
310            chunks_created,
311            crate::memory_guard::current_process_memory_mb().unwrap_or(0)
312        ),
313    );
314
315    if chunks_created > crate::constants::REMEMBER_MAX_SAFE_MULTI_CHUNKS {
316        return Err(AppError::TooManyChunks {
317            chunks: chunks_created,
318            limit: crate::constants::REMEMBER_MAX_SAFE_MULTI_CHUNKS,
319        });
320    }
321
322    let embed_out = super::embed_phase::run_embed_phase(
323        &paths,
324        &raw_body,
325        &chunks_info,
326        &graph,
327        &args,
328        crate::cli::BackendChoice::new(llm_backend, embedding_backend),
329    )?;
330    let embedding = embed_out.embedding;
331    let backend_invoked_passage = embed_out.backend_invoked_passage;
332    let mut chunk_embeddings_cache = embed_out.chunk_embeddings_cache;
333    let graph_entity_embeddings = embed_out.graph_entity_embeddings;
334    let _skip_embed = embed_out.skip_embed;
335
336    let body_for_storage = raw_body;
337    let memory_type = resolved_type.as_str();
338    let new_memory = NewMemory {
339        namespace: namespace.clone(),
340        name: normalized_name.clone(),
341        memory_type: memory_type.to_string(),
342        description: resolved_description.clone(),
343        body: body_for_storage,
344        body_hash: body_hash.clone(),
345        session_id: args.session_id.clone(),
346        source: "agent".to_string(),
347        metadata,
348    };
349
350    let mut warnings = Vec::with_capacity(4);
351    // v1.2.8: a declared `entity_type` outside the canonical set is reported on
352    // the same envelope that reports success. The label is stored as written, so
353    // this is advice, not a correction: the caller learns their taxonomy is
354    // local to this payload while they can still align it.
355    warnings.append(&mut graph.type_warnings);
356    let mut entities_persisted = 0usize;
357    let mut relationships_persisted = 0usize;
358
359    let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
360
361    let mut skip_reindex = false;
362    let (memory_id, action, version) = match existing_memory {
363        Some((existing_id, _updated_at, _current_version)) => {
364            if let Some(hash_id) = duplicate_hash_id {
365                if hash_id != existing_id {
366                    warnings.push(format!(
367                        "identical body already exists as memory id {hash_id}"
368                    ));
369                }
370            }
371
372            // C1 fix: capture old values for FTS5 sync before update
373            let (old_fts_name, old_fts_desc, old_fts_body): (String, String, String) = tx
374                .query_row(
375                    "SELECT name, description, body FROM memories WHERE id = ?1",
376                    rusqlite::params![existing_id],
377                    |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
378                )?;
379
380            // G15: skip re-indexing when body hash matches (common in --force-merge loops)
381            let existing_body_hash: Option<String> = tx
382                .query_row(
383                    "SELECT body_hash FROM memories WHERE id = ?1",
384                    rusqlite::params![existing_id],
385                    |r| r.get(0),
386                )
387                .ok();
388            let body_unchanged = existing_body_hash.as_deref() == Some(&body_hash);
389            skip_reindex = body_unchanged;
390            if !body_unchanged {
391                storage_chunks::delete_chunks(&tx, existing_id)?;
392            }
393
394            let next_v = versions::next_version(&tx, existing_id)?;
395            memories::update(&tx, existing_id, &new_memory, args.expected_updated_at)?;
396
397            // C1 fix: sync FTS5 external-content index after update
398            // (trg_fts_au trigger is absent by design due to sqlite-vec conflict)
399            memories::sync_fts_after_update(
400                &tx,
401                existing_id,
402                &old_fts_name,
403                &old_fts_desc,
404                &old_fts_body,
405                &normalized_name,
406                &resolved_description,
407                &new_memory.body,
408            )?;
409
410            versions::insert_version(
411                &tx,
412                existing_id,
413                next_v,
414                &normalized_name,
415                memory_type,
416                &resolved_description,
417                &new_memory.body,
418                &serde_json::to_string(&new_memory.metadata)?,
419                None,
420                "edit",
421            )?;
422            if !body_unchanged {
423                if let Some(ref emb) = embedding {
424                    memories::upsert_vec(
425                        &tx,
426                        existing_id,
427                        &namespace,
428                        memory_type,
429                        emb,
430                        &normalized_name,
431                        &snippet,
432                    )?;
433                }
434            }
435            (existing_id, "updated".to_string(), next_v)
436        }
437        None => {
438            if let Some(hash_id) = duplicate_hash_id {
439                warnings.push(format!(
440                    "identical body already exists as memory id {hash_id}"
441                ));
442            }
443            let id = memories::insert(&tx, &new_memory)?;
444            versions::insert_version(
445                &tx,
446                id,
447                1,
448                &normalized_name,
449                memory_type,
450                &resolved_description,
451                &new_memory.body,
452                &serde_json::to_string(&new_memory.metadata)?,
453                None,
454                "create",
455            )?;
456            if let Some(ref emb) = embedding {
457                memories::upsert_vec(
458                    &tx,
459                    id,
460                    &namespace,
461                    memory_type,
462                    emb,
463                    &normalized_name,
464                    &snippet,
465                )?;
466            }
467            (id, "created".to_string(), 1)
468        }
469    };
470
471    // GAP-SG-51: when --force-merge --replace-graph updates an existing memory,
472    // clear its prior entity/relationship bindings BEFORE re-linking the supplied
473    // set. With an empty `entities`/`relationships` payload this zeroes the graph
474    // for that memory without a `forget`. New bindings (if any) are linked by the
475    // block further below.
476    if args.replace_graph && action == "updated" {
477        let (e_removed, r_removed) = entities::clear_memory_graph_bindings(&tx, memory_id)?;
478        if e_removed + r_removed > 0 {
479            warnings.push(format!(
480                "--replace-graph cleared {e_removed} entity binding(s) and {r_removed} relationship binding(s) before re-linking"
481            ));
482        }
483    }
484
485    if chunks_info.len() > 1 && !skip_reindex {
486        storage_chunks::insert_chunk_slices(&tx, memory_id, &new_memory.body, &chunks_info)?;
487
488        if let Some(chunk_embeddings) = chunk_embeddings_cache.take() {
489            for (i, emb) in chunk_embeddings.iter().enumerate() {
490                storage_chunks::upsert_chunk_vec(&tx, i as i64, memory_id, i as i32, emb)?;
491            }
492        }
493        output::emit_progress_i18n(
494            &format!(
495                "Remember stage: persisted chunk vectors; process RSS {} MB",
496                crate::memory_guard::current_process_memory_mb().unwrap_or(0)
497            ),
498            &format!(
499                "Etapa remember: vetores de chunks persistidos; RSS do processo {} MB",
500                crate::memory_guard::current_process_memory_mb().unwrap_or(0)
501            ),
502        );
503    }
504
505    if !graph.entities.is_empty() || !graph.relationships.is_empty() {
506        for entity in &graph.entities {
507            let entity_id = entities::upsert_entity(&tx, &namespace, entity)?;
508            let entity_embedding = &graph_entity_embeddings[entities_persisted];
509            entities::upsert_entity_vec(
510                &tx,
511                entity_id,
512                &namespace,
513                &entity.entity_type,
514                entity_embedding,
515                &entity.name,
516            )?;
517            entities::link_memory_entity(&tx, memory_id, entity_id)?;
518            entities_persisted += 1;
519        }
520        let entity_types: std::collections::HashMap<&str, &str> = graph
521            .entities
522            .iter()
523            .map(|entity| (entity.name.as_str(), entity.entity_type.as_str()))
524            .collect();
525
526        let mut affected_entity_ids: std::collections::HashSet<i64> =
527            std::collections::HashSet::new();
528        for entity in &graph.entities {
529            if let Some(eid) = entities::find_entity_id(&tx, &namespace, &entity.name)? {
530                affected_entity_ids.insert(eid);
531            }
532        }
533
534        for rel in &graph.relationships {
535            let source_entity = NewEntity {
536                name: rel.source.clone(),
537                entity_type: entity_types
538                    .get(rel.source.as_str())
539                    .copied()
540                    .unwrap_or(DEFAULT_ENTITY_TYPE)
541                    .to_string(),
542                description: None,
543            };
544            let target_entity = NewEntity {
545                name: rel.target.clone(),
546                entity_type: entity_types
547                    .get(rel.target.as_str())
548                    .copied()
549                    .unwrap_or(DEFAULT_ENTITY_TYPE)
550                    .to_string(),
551                description: None,
552            };
553            let source_id = entities::upsert_entity(&tx, &namespace, &source_entity)?;
554            let target_id = entities::upsert_entity(&tx, &namespace, &target_entity)?;
555            let rel_id = entities::upsert_relationship(&tx, &namespace, source_id, target_id, rel)?;
556            entities::link_memory_relationship(&tx, memory_id, rel_id)?;
557            affected_entity_ids.insert(source_id);
558            affected_entity_ids.insert(target_id);
559            relationships_persisted += 1;
560        }
561
562        for &eid in &affected_entity_ids {
563            entities::recalculate_degree(&tx, eid)?;
564        }
565    }
566    tx.commit()?;
567
568    super::finish::emit_remember_result(
569        super::finish::FinishContext {
570            conn: &conn,
571            paths: &paths,
572            args: &args,
573            graph: &graph,
574            new_memory: &new_memory,
575        },
576        super::finish::FinishIdentity {
577            memory_id,
578            namespace,
579            normalized_name,
580            original_name,
581            name_was_normalized,
582        },
583        super::finish::FinishOutcome {
584            action,
585            version,
586            entities_persisted,
587            relationships_persisted,
588            relationships_updated,
589            chunks_created,
590        },
591        super::finish::FinishExtraction {
592            extracted_urls,
593            extraction_method,
594            backend_invoked_passage,
595        },
596        warnings,
597        started,
598    )?;
599
600    Ok(())
601}
sqlite_graphrag/commands/remember/run.rs

sqlite_graphrag/commands/remember/
run.rs
