Skip to main content

memnite_cli/
app.rs

1use std::collections::{BTreeMap, HashSet};
2use std::io::{BufRead, Write};
3use std::path::Path;
4
5use memnite_core::{
6    detect_conflicts, new_event_id, next_lamport, normalize_project, replay, Anchor, Event,
7    EventKind, Log, MemoryData, MemoryPatch, Relation, Scope, Status,
8};
9use memnite_ingest::{slugify, IngestSource, IngestSummary};
10use memnite_stale::hash_region;
11use memnite_store::{EventMetaRow, MemoryRow, Store};
12
13use crate::error::CliError;
14use crate::spec::{
15    AddSpec, DoctorReport, EventCtx, ImportSummary, RelationSpec, SearchQuery, UpdatePatch,
16};
17
18/// An open Memnite instance: the event log + its SQLite projection.
19pub struct App {
20    pub(crate) log: Log,
21    pub(crate) store: Store,
22}
23
24impl App {
25    /// Open (creating if needed) a Memnite data dir: `<dir>/log/` + `<dir>/memnite.db`.
26    pub fn open(data_dir: impl AsRef<Path>) -> Result<Self, CliError> {
27        let data_dir = data_dir.as_ref();
28        let log = Log::open(data_dir.join("log"))?;
29        let store = Store::open(data_dir.join("memnite.db"))?;
30        Ok(App { log, store })
31    }
32
33    /// Create a memory: stamp anchor hashes from `root`, append a `MemoryAdded`
34    /// event to the log, and project it. Returns the projected `MemoryRow`.
35    pub fn add(&self, spec: AddSpec, root: &Path, ctx: &EventCtx) -> Result<MemoryRow, CliError> {
36        let mut anchors = Vec::with_capacity(spec.anchors.len());
37        for a in &spec.anchors {
38            let content_hash = hash_region(root, &a.path, a.line_start, a.line_end)?;
39            anchors.push(Anchor {
40                path: a.path.clone(),
41                symbol: a.symbol.clone(),
42                line_start: a.line_start,
43                line_end: a.line_end,
44                content_hash,
45            });
46        }
47
48        // Canonicalize the project once, up front: the event stores the canonical
49        // name and the upsert key matches by it, so name variants collapse to one
50        // bucket. Pure (no I/O, no new event) — resolution from the environment is
51        // the caller's job (see project_resolve).
52        let project = normalize_project(&spec.project);
53
54        // topic_key upsert: if a live memory shares (topic_key, project, scope),
55        // update it in place instead of adding a new one. Look up before the spec
56        // fields are moved into MemoryData.
57        let target = match &spec.topic_key {
58            Some(t) => self.store.find_live_by_topic(t, &project, &spec.scope)?,
59            None => None,
60        };
61
62        let data = MemoryData {
63            title: spec.title,
64            body: spec.body,
65            mem_type: spec.mem_type,
66            scope: spec.scope,
67            project,
68            topic_key: spec.topic_key,
69            anchors,
70            tags: spec.tags,
71        };
72        let (memory_id, kind) = match target {
73            Some(id) => (id, EventKind::MemoryUpdated(data)),
74            None => (
75                format!("mem_{}", new_event_id()),
76                EventKind::MemoryAdded(data),
77            ),
78        };
79
80        let event = Event {
81            event_id: new_event_id(),
82            lamport: next_lamport(&self.log.read_all()?),
83            ts: ctx.ts.clone(),
84            engine: ctx.engine.clone(),
85            machine: ctx.machine.clone(),
86            memory_id: memory_id.clone(),
87            kind,
88        };
89
90        self.log.append(&event)?;
91        self.store.apply_event(&event)?;
92        self.get(&memory_id)?
93            .ok_or_else(|| CliError::NotFound(memory_id))
94    }
95
96    /// Persist a session summary as a memory (`mem_type = "session_summary"`).
97    /// Model-driven: the caller passes the distilled text. Plain add — summaries
98    /// accumulate as history (no topic_key upsert).
99    pub fn session_summary(
100        &self,
101        summary: String,
102        project: String,
103        scope: Scope,
104        ctx: &EventCtx,
105    ) -> Result<MemoryRow, CliError> {
106        let spec = AddSpec {
107            title: format!("Session summary {}", ctx.ts),
108            body: summary,
109            mem_type: "session_summary".to_string(),
110            scope,
111            project,
112            topic_key: None,
113            anchors: vec![],
114            tags: vec![],
115        };
116        // No anchors → `root` is unused by `add`; a placeholder path is safe.
117        self.add(spec, Path::new("."), ctx)
118    }
119
120    /// Fetch one memory by id.
121    pub fn get(&self, memory_id: &str) -> Result<Option<MemoryRow>, CliError> {
122        Ok(self.store.get(memory_id)?)
123    }
124
125    /// Rebuild the projection from the full log. Returns the event count folded.
126    pub fn rebuild(&self) -> Result<usize, CliError> {
127        let events = self.log.read_all()?;
128        self.store.rebuild(&events)?;
129        Ok(events.len())
130    }
131
132    /// Full-text search over memories, narrowed by optional filters. User text
133    /// is sanitized into FTS5 syntax; a blank query returns no results. Accepts
134    /// anything convertible into `SearchQuery` (e.g. a plain `&str`).
135    pub fn search(&self, query: impl Into<SearchQuery>) -> Result<Vec<MemoryRow>, CliError> {
136        use crate::fts::sanitize_fts;
137        let query = query.into();
138        let fts = sanitize_fts(&query.text, query.match_any);
139        if fts.is_empty() {
140            return Ok(Vec::new());
141        }
142        let filters = memnite_store::SearchFilters {
143            mem_type: query.mem_type,
144            project: query.project.map(|p| normalize_project(&p)),
145            scope: query.scope,
146        };
147        Ok(self.store.search(&fts, &filters)?)
148    }
149
150    /// All memories currently marked stale.
151    pub fn list_stale(&self) -> Result<Vec<MemoryRow>, CliError> {
152        Ok(self.store.list_stale()?)
153    }
154
155    /// Tombstone a memory (append a `MemoryDeleted` event). Errors if the id is
156    /// unknown. Re-deleting an already-deleted memory is a no-op (delete is
157    /// terminal in the fold), so no duplicate event is appended.
158    pub fn delete(&self, memory_id: &str, ctx: &EventCtx) -> Result<(), CliError> {
159        let current = self
160            .store
161            .get(memory_id)?
162            .ok_or_else(|| CliError::NotFound(memory_id.to_string()))?;
163        if current.status == "deleted" {
164            return Ok(());
165        }
166        let event = Event {
167            event_id: new_event_id(),
168            lamport: next_lamport(&self.log.read_all()?),
169            ts: ctx.ts.clone(),
170            engine: ctx.engine.clone(),
171            machine: ctx.machine.clone(),
172            memory_id: memory_id.to_string(),
173            kind: EventKind::MemoryDeleted,
174        };
175        self.log.append(&event)?;
176        self.store.apply_event(&event)?;
177        Ok(())
178    }
179
180    /// Mark a memory as reviewed: append a `MemoryReviewed` event, resetting its
181    /// decay clock (`updated_ts`) without changing its status. Errors `NotFound` if
182    /// the id is unknown or already deleted (a tombstone cannot be reviewed).
183    pub fn mark(&self, memory_id: &str, ctx: &EventCtx) -> Result<(), CliError> {
184        let current = self
185            .store
186            .get(memory_id)?
187            .ok_or_else(|| CliError::NotFound(memory_id.to_string()))?;
188        if current.status == "deleted" {
189            return Err(CliError::NotFound(memory_id.to_string()));
190        }
191        let event = Event {
192            event_id: new_event_id(),
193            lamport: next_lamport(&self.log.read_all()?),
194            ts: ctx.ts.clone(),
195            engine: ctx.engine.clone(),
196            machine: ctx.machine.clone(),
197            memory_id: memory_id.to_string(),
198            kind: EventKind::MemoryReviewed,
199        };
200        self.log.append(&event)?;
201        self.store.apply_event(&event)?;
202        Ok(())
203    }
204
205    /// Live memories whose decay verdict (given `now`, an RFC3339 timestamp) is
206    /// `NeedsReview`, optionally narrowed by exact `project` / `mem_type`. Read-time
207    /// only — nothing is written and the log is untouched.
208    pub fn review_list(
209        &self,
210        now: &str,
211        project: Option<&str>,
212        mem_type: Option<&str>,
213    ) -> Result<Vec<MemoryRow>, CliError> {
214        use memnite_core::{decay_state, DecayState};
215        // Canonicalize the filter so a name variant matches the stored (canonical) project.
216        let project = project.map(normalize_project);
217        let mut out = Vec::new();
218        for row in self.store.list_all()? {
219            if row.status == "deleted" {
220                continue;
221            }
222            if project.as_deref().is_some_and(|p| row.project != p) {
223                continue;
224            }
225            if mem_type.is_some_and(|t| row.mem_type != t) {
226                continue;
227            }
228            if decay_state(&row.mem_type, &row.updated_ts, now) == DecayState::NeedsReview {
229                out.push(row);
230            }
231        }
232        Ok(out)
233    }
234
235    /// Assert a relation `from_id → to_id`. Appends a `RelationAsserted` event and
236    /// projects it. Both memories must exist, be live (not deleted), and differ.
237    /// `spec.judged_by` records who classified it ("agent" for interactive/manual,
238    /// an engine name for the judge).
239    pub fn relate(
240        &self,
241        from_id: &str,
242        to_id: &str,
243        spec: RelationSpec,
244        ctx: &EventCtx,
245    ) -> Result<(), CliError> {
246        if from_id == to_id {
247            return Err(CliError::Invalid(
248                "cannot relate a memory to itself".to_string(),
249            ));
250        }
251        let from = self
252            .store
253            .get(from_id)?
254            .ok_or_else(|| CliError::NotFound(from_id.to_string()))?;
255        let to = self
256            .store
257            .get(to_id)?
258            .ok_or_else(|| CliError::NotFound(to_id.to_string()))?;
259        if from.status == "deleted" || to.status == "deleted" {
260            return Err(CliError::Invalid(format!(
261                "cannot relate a deleted memory ({from_id} or {to_id})"
262            )));
263        }
264        if !(0.0..=1.0).contains(&spec.confidence) {
265            return Err(CliError::Invalid(format!(
266                "confidence must be in [0.0, 1.0], got {}",
267                spec.confidence
268            )));
269        }
270
271        let event = Event {
272            event_id: new_event_id(),
273            lamport: next_lamport(&self.log.read_all()?),
274            ts: ctx.ts.clone(),
275            engine: ctx.engine.clone(),
276            machine: ctx.machine.clone(),
277            memory_id: from_id.to_string(),
278            kind: EventKind::RelationAsserted {
279                to_id: to_id.to_string(),
280                relation: spec.relation,
281                confidence: spec.confidence,
282                reason: spec.reason,
283                judged_by: spec.judged_by,
284            },
285        };
286        self.log.append(&event)?;
287        self.store.apply_event(&event)?;
288        Ok(())
289    }
290
291    /// Relation edges touching a memory (as source or target).
292    pub fn relations_for(
293        &self,
294        memory_id: &str,
295    ) -> Result<Vec<memnite_store::RelationRow>, CliError> {
296        Ok(self.store.relations_for(memory_id)?)
297    }
298
299    /// Every relation edge.
300    pub fn list_relations(&self) -> Result<Vec<memnite_store::RelationRow>, CliError> {
301        Ok(self.store.list_relations()?)
302    }
303
304    /// Candidate memories worth relating to `memory_id`: FTS over its title within
305    /// the same project + scope, excluding self / deleted / already-related pairs.
306    /// Uses OR-match (broad recall). Returns up to `limit` rows. Callers treat a
307    /// failure as "no candidates" (non-blocking) at the surface layer.
308    pub fn find_candidates(&self, memory_id: &str, limit: u32) -> Result<Vec<MemoryRow>, CliError> {
309        use crate::fts::sanitize_fts;
310        let Some(src) = self.store.get(memory_id)? else {
311            return Ok(Vec::new());
312        };
313        let fts = sanitize_fts(&src.title, true);
314        if fts.is_empty() {
315            return Ok(Vec::new());
316        }
317        Ok(self
318            .store
319            .find_candidates(&fts, &src.project, &src.scope, memory_id, limit)?)
320    }
321
322    /// Batch scan: for each live memory, find candidates and ask the injected
323    /// `judge` to classify each pair. Non-`not_conflict` verdicts are asserted as
324    /// `RelationAsserted` events stamped with `judged_by` (provenance is the
325    /// caller's, never a hardcoded literal — keeps the audit trail honest under a
326    /// swapped judge). Judge errors are swallowed per pair (never abort); the first
327    /// one is sampled into `ScanSummary::first_error`.
328    ///
329    /// `since` limits the scan to memories updated at/after it. It must be the same
330    /// RFC3339 form as the stored `updated_ts` — the comparison is lexical, so a
331    /// differently-formatted timestamp will filter incorrectly.
332    pub fn conflicts_scan(
333        &self,
334        judge: impl Fn(
335            &memnite_judge::MemoryView,
336            &memnite_judge::MemoryView,
337        ) -> Result<memnite_judge::Verdict, memnite_judge::JudgeError>,
338        judged_by: &str,
339        since: Option<String>,
340        limit: u32,
341        ctx: &EventCtx,
342    ) -> Result<crate::spec::ScanSummary, CliError> {
343        let mut summary = crate::spec::ScanSummary::default();
344        let mut lamport = next_lamport(&self.log.read_all()?);
345
346        for row in self.store.list_all()? {
347            if row.status == "deleted" {
348                continue;
349            }
350            if let Some(s) = &since {
351                if row.updated_ts.as_str() < s.as_str() {
352                    continue;
353                }
354            }
355            for cand in self.find_candidates(&row.memory_id, limit)? {
356                summary.pairs += 1;
357                let a = memnite_judge::MemoryView {
358                    id: row.memory_id.clone(),
359                    title: row.title.clone(),
360                    body: row.body.clone(),
361                };
362                let b = memnite_judge::MemoryView {
363                    id: cand.memory_id.clone(),
364                    title: cand.title.clone(),
365                    body: cand.body.clone(),
366                };
367                match judge(&a, &b) {
368                    Ok(v) if v.relation == Relation::NotConflict => summary.not_conflict += 1,
369                    Ok(v) => {
370                        let event = Event {
371                            event_id: new_event_id(),
372                            lamport,
373                            ts: ctx.ts.clone(),
374                            engine: ctx.engine.clone(),
375                            machine: ctx.machine.clone(),
376                            memory_id: row.memory_id.clone(),
377                            kind: EventKind::RelationAsserted {
378                                to_id: cand.memory_id.clone(),
379                                relation: v.relation,
380                                confidence: v.confidence,
381                                reason: v.reason,
382                                judged_by: judged_by.to_string(),
383                            },
384                        };
385                        self.log.append(&event)?;
386                        self.store.apply_event(&event)?;
387                        lamport += 1;
388                        summary.asserted += 1;
389                    }
390                    Err(e) => {
391                        if summary.first_error.is_none() {
392                            summary.first_error = Some(e.to_string());
393                        }
394                        summary.errors += 1;
395                    }
396                }
397            }
398        }
399        Ok(summary)
400    }
401
402    /// Run the staleness checker over every anchored, non-deleted memory. When a
403    /// memory's verdict changes its status, append the verdict event and project
404    /// it. Memories without anchors (nothing to verify) are skipped.
405    pub fn check(
406        &self,
407        root: &Path,
408        ctx: &EventCtx,
409    ) -> Result<crate::spec::CheckSummary, CliError> {
410        use crate::spec::CheckSummary;
411        use memnite_stale::{check_memory, verdict_to_kind, Verdict};
412
413        let mut summary = CheckSummary::default();
414        // Read the log once; track lamport locally (incremented per emitted event)
415        // instead of re-reading the full log for every changed memory.
416        let mut lamport = next_lamport(&self.log.read_all()?);
417
418        for row in self.store.list_all()? {
419            if row.status == "deleted" || row.anchors.is_empty() {
420                continue;
421            }
422
423            let anchors: Vec<memnite_core::Anchor> = row
424                .anchors
425                .iter()
426                .map(|a| memnite_core::Anchor {
427                    path: a.path.clone(),
428                    symbol: a.symbol.clone(),
429                    line_start: a.line_start,
430                    line_end: a.line_end,
431                    content_hash: a.content_hash.clone(),
432                })
433                .collect();
434
435            let verdict = check_memory(root, &anchors)?;
436            let new_status = match verdict {
437                Verdict::Stable => "stable",
438                Verdict::Stale(_) => "stale",
439            };
440
441            if new_status == row.status {
442                summary.unchanged += 1;
443                continue;
444            }
445
446            let event = Event {
447                event_id: new_event_id(),
448                lamport,
449                ts: ctx.ts.clone(),
450                engine: ctx.engine.clone(),
451                machine: ctx.machine.clone(),
452                memory_id: row.memory_id.clone(),
453                kind: verdict_to_kind(&verdict),
454            };
455            self.log.append(&event)?;
456            self.store.apply_event(&event)?;
457            lamport += 1;
458
459            match verdict {
460                Verdict::Stable => summary.stable += 1,
461                Verdict::Stale(_) => summary.stale += 1,
462            }
463        }
464
465        Ok(summary)
466    }
467
468    /// Recent memories for a project (newest-updated first, deleted excluded).
469    pub fn context(&self, project: &str, limit: u32) -> Result<Vec<MemoryRow>, CliError> {
470        Ok(self.store.context(&normalize_project(project), limit)?)
471    }
472
473    /// Event history of one memory, oldest first.
474    pub fn timeline(&self, memory_id: &str) -> Result<Vec<EventMetaRow>, CliError> {
475        Ok(self.store.timeline(memory_id)?)
476    }
477
478    /// Test-only: read the raw event log. Public for integration tests that must
479    /// assert on event payloads not surfaced by the projection (e.g. tags).
480    #[doc(hidden)]
481    pub fn read_log_for_test(&self) -> Vec<Event> {
482        self.log.read_all().unwrap_or_default()
483    }
484
485    /// Test-only: wipe the projection to an empty state without touching the log.
486    /// Lets integration tests simulate projection/log divergence for `doctor`.
487    #[doc(hidden)]
488    pub fn rebuild_from_empty_for_test(&self) {
489        self.store.rebuild(&[]).unwrap();
490    }
491
492    /// Compare the log fold against the projection and report divergences. A
493    /// clean projection (after any normal op, since each op projects incrementally)
494    /// returns zero mismatches and `cursor_ok == true`.
495    pub fn doctor(&self) -> Result<DoctorReport, CliError> {
496        let events = self.log.read_all()?;
497        let folded = replay(&events);
498        let projected = self.store.list_all()?;
499
500        let proj: BTreeMap<&str, &MemoryRow> = projected
501            .iter()
502            .map(|m| (m.memory_id.as_str(), m))
503            .collect();
504
505        let mut mismatches = Vec::new();
506        for (id, m) in &folded {
507            match proj.get(id.as_str()) {
508                None => mismatches.push(format!("{id}: in log, missing from projection")),
509                Some(p) => {
510                    let want = status_str(m.status);
511                    if p.status != want {
512                        mismatches.push(format!("{id}: status log={want} proj={}", p.status));
513                    }
514                    if p.title != m.title {
515                        mismatches.push(format!("{id}: title differs"));
516                    }
517                    if p.last_event_id != m.last_event_id {
518                        mismatches.push(format!("{id}: last_event_id differs"));
519                    }
520                }
521            }
522        }
523        for p in &projected {
524            if !folded.contains_key(&p.memory_id) {
525                mismatches.push(format!("{}: in projection, missing from log", p.memory_id));
526            }
527        }
528
529        let max_lamport = events.iter().map(|e| e.lamport).max().unwrap_or(0);
530        let cursor_ok = self.store.sync_cursor()?.1 == max_lamport;
531        let conflicts = detect_conflicts(&events);
532
533        Ok(DoctorReport {
534            log_count: folded.len(),
535            proj_count: projected.len(),
536            mismatches,
537            cursor_ok,
538            conflicts,
539        })
540    }
541
542    /// Update a memory by patch: emits a sparse `MemoryPatched` delta containing
543    /// only the fields `patch` provides — untouched fields are left `None` and
544    /// carry no value over the wire, enabling field-level LWW merge downstream
545    /// (see `memnite_core::replay`). The LOG FOLD (not the projection) is only
546    /// consulted to validate the memory exists and is not deleted; it is not used
547    /// to fill in omitted fields. Anchors in the patch are re-hashed from `root`;
548    /// omitted anchors are left untouched. Errors if the id is unknown. Returns
549    /// the new projected row.
550    pub fn update(
551        &self,
552        memory_id: &str,
553        patch: UpdatePatch,
554        root: &Path,
555        ctx: &EventCtx,
556    ) -> Result<MemoryRow, CliError> {
557        let events = self.log.read_all()?;
558        let state = replay(&events);
559        let base = state
560            .get(memory_id)
561            .ok_or_else(|| CliError::NotFound(memory_id.to_string()))?;
562
563        if base.status == Status::Deleted {
564            return Err(CliError::Invalid(format!(
565                "{memory_id} is deleted; cannot update"
566            )));
567        }
568
569        // anchors: re-hash from `root` only if the patch provides them; omitted
570        // (None) means "do not touch anchors".
571        let anchors = match patch.anchors {
572            Some(specs) => {
573                let mut out = Vec::with_capacity(specs.len());
574                for a in &specs {
575                    let content_hash = hash_region(root, &a.path, a.line_start, a.line_end)?;
576                    out.push(Anchor {
577                        path: a.path.clone(),
578                        symbol: a.symbol.clone(),
579                        line_start: a.line_start,
580                        line_end: a.line_end,
581                        content_hash,
582                    });
583                }
584                Some(out)
585            }
586            None => None,
587        };
588
589        let delta = MemoryPatch {
590            title: patch.title,
591            body: patch.body,
592            mem_type: patch.mem_type,
593            scope: patch.scope,
594            project: patch.project.map(|p| normalize_project(&p)),
595            topic_key: patch.topic_key,
596            anchors,
597            tags: patch.tags,
598        };
599
600        let event = Event {
601            event_id: new_event_id(),
602            lamport: next_lamport(&events),
603            ts: ctx.ts.clone(),
604            engine: ctx.engine.clone(),
605            machine: ctx.machine.clone(),
606            memory_id: memory_id.to_string(),
607            kind: EventKind::MemoryPatched(delta),
608        };
609        self.log.append(&event)?;
610        self.store.apply_event(&event)?;
611        self.get(memory_id)?
612            .ok_or_else(|| CliError::NotFound(memory_id.to_string()))
613    }
614
615    /// Import an NDJSON event bundle: append only events whose `event_id` is not
616    /// already in the local log (dedup), then rebuild the projection from the
617    /// full log. Rebuild — not incremental apply — because imported events may
618    /// carry a `lamport` below the local max and must be re-folded in
619    /// `(lamport, event_id)` order. Corrupt lines are skipped and counted.
620    pub fn import(&self, reader: impl BufRead) -> Result<ImportSummary, CliError> {
621        let mut events = self.log.read_all()?;
622        let mut seen: HashSet<String> = events.iter().map(|e| e.event_id.clone()).collect();
623        let mut sum = ImportSummary::default();
624        for line in reader.lines() {
625            let line = line?;
626            if line.trim().is_empty() {
627                continue;
628            }
629            match serde_json::from_str::<Event>(&line) {
630                Err(_) => sum.corrupt += 1,
631                Ok(ev) => {
632                    if seen.contains(&ev.event_id) {
633                        sum.duplicates += 1;
634                    } else {
635                        self.log.append(&ev)?;
636                        seen.insert(ev.event_id.clone());
637                        events.push(ev);
638                        sum.imported += 1;
639                    }
640                }
641            }
642        }
643        if sum.imported > 0 {
644            self.store.rebuild(&events)?;
645        }
646        Ok(sum)
647    }
648
649    /// Write the event log as an NDJSON bundle (one event per line), sorted by
650    /// `(lamport, event_id)` for deterministic, diff-friendly bundles. With
651    /// `since`, only events whose `lamport >= since` are written. Returns the
652    /// number of events written.
653    pub fn export(&self, since: Option<u64>, mut writer: impl Write) -> Result<usize, CliError> {
654        let mut evs = self.log.read_all()?;
655        if let Some(s) = since {
656            evs.retain(|e| e.lamport >= s);
657        }
658        evs.sort_by(|a, b| {
659            a.lamport
660                .cmp(&b.lamport)
661                .then_with(|| a.event_id.cmp(&b.event_id))
662        });
663        for e in &evs {
664            writeln!(writer, "{}", serde_json::to_string(e)?)?;
665        }
666        Ok(evs.len())
667    }
668
669    /// Ingest memories from any `IngestSource` (source-agnostic). Each item gets a
670    /// deterministic `memory_id` (`mem_ing_<source>_<key-slug>`); a new id is
671    /// added, a changed one (by projected title/body/type/topic) is updated, an
672    /// unchanged one is skipped. Idempotent.
673    pub fn ingest(
674        &self,
675        source: &dyn IngestSource,
676        root: &Path,
677        project: &str,
678        ctx: &EventCtx,
679    ) -> Result<IngestSummary, CliError> {
680        let mems = source.collect(root)?;
681        let mut summary = IngestSummary::default();
682        // Read the log once; track the lamport locally (incremented only when an
683        // event is actually emitted) instead of re-reading the log per item.
684        let mut lamport = next_lamport(&self.log.read_all()?);
685        for m in mems {
686            let id = format!("mem_ing_{}_{}", source.name(), slugify(&m.source_key));
687            let existing = self.store.get(&id)?;
688            if let Some(row) = &existing {
689                if row.title == m.title
690                    && row.body == m.body
691                    && row.mem_type == m.mem_type
692                    && row.topic_key == m.topic_key
693                {
694                    summary.unchanged += 1;
695                    continue;
696                }
697            }
698            let data = MemoryData {
699                title: m.title,
700                body: m.body,
701                mem_type: m.mem_type,
702                scope: m.scope,
703                project: project.to_string(),
704                topic_key: m.topic_key,
705                anchors: vec![],
706                tags: vec![],
707            };
708            let kind = if existing.is_some() {
709                EventKind::MemoryUpdated(data)
710            } else {
711                EventKind::MemoryAdded(data)
712            };
713            let event = Event {
714                event_id: new_event_id(),
715                lamport,
716                ts: ctx.ts.clone(),
717                engine: ctx.engine.clone(),
718                machine: ctx.machine.clone(),
719                memory_id: id,
720                kind,
721            };
722            self.log.append(&event)?;
723            self.store.apply_event(&event)?;
724            lamport += 1;
725            if existing.is_some() {
726                summary.updated += 1;
727            } else {
728                summary.added += 1;
729            }
730        }
731        Ok(summary)
732    }
733}
734
735/// Core `Status` → the text form stored in the projection's `status` column.
736fn status_str(s: Status) -> &'static str {
737    match s {
738        Status::Stable => "stable",
739        Status::Stale => "stale",
740        Status::Deleted => "deleted",
741    }
742}