Skip to main content

plugmem_cli/
lib.rs

1#![doc = include_str!("../README.md")]
2//! `plugmem` — the command-line surface over the
3//! [temporal-memory engine](https://docs.rs/plugmem-core/latest), a thin wrapper around
4//! [`plugmem_host::Database`]. Parse the arguments, call one
5//! engine verb, render the result — human text by default, `--json` for
6//! tooling and agents. No memory logic lives here; that is the engine's.
7//!
8//! Exit codes: `0` success; `1` a soft miss (the target fact does not
9//! exist, or the database is locked by another process); `2` a usage or
10//! runtime error. This makes the binary scriptable as a gate.
11//!
12//! The logic is in this library (not `main.rs`) so it is unit-testable:
13//! [`run`] wires argv and the database, and `execute` runs one command
14//! against an open [`Database`] into any writer.
15
16mod cli;
17mod config;
18mod workspace;
19
20use std::collections::BTreeMap;
21use std::io::{self, BufRead, Write};
22use std::path::{Path, PathBuf};
23use std::process::ExitCode;
24use std::time::{SystemTime, UNIX_EPOCH};
25
26use clap::Parser;
27use plugmem_host::{
28    Database, ExportedFact, FactId, GuardedRememberOutcome, HostError, LinkInput, MaintenanceMode,
29    MaintenanceOptions, ReadOnlyDatabase, RecallQuery, RecallResult, RememberInput,
30    RememberOutcome, Settings, Stats, TagPage, TagQuery, UnlinkInput, VALID_TO_OPEN,
31};
32use serde_json::json;
33
34use crate::cli::{Cli, Command, HelpTopic, MaintainMode};
35use crate::config::read_batch_size;
36
37/// Environment variable naming the database file (below the `--db` flag).
38pub(crate) const ENV_DB: &str = "PLUGMEM_DB";
39/// Last-resort relative database name if the platform data directory is unavailable.
40pub(crate) const DEFAULT_DB: &str = "plugmem.db";
41
42/// A failure before or during a command: a runtime engine/host error, or a
43/// usage error (a malformed argument the parser could not catch).
44#[derive(Debug)]
45pub(crate) enum CliError {
46    Host(HostError),
47    Usage(String),
48}
49
50impl From<HostError> for CliError {
51    fn from(e: HostError) -> Self {
52        CliError::Host(e)
53    }
54}
55
56/// Wall-clock now in unix milliseconds (the engine keeps no clock).
57pub(crate) fn now_ms() -> u64 {
58    SystemTime::now()
59        .duration_since(UNIX_EPOCH)
60        .map(|d| d.as_millis() as u64)
61        .unwrap_or(0)
62}
63
64/// Parses argv and runs one command, mapping the result to a process exit
65/// code. The binary's `main` is a one-liner over this; the wiring itself is
66/// `run_parsed`, which is unit-testable (only `Cli::parse` is not).
67pub fn run() -> ExitCode {
68    let stdout = io::stdout();
69    ExitCode::from(run_parsed(Cli::parse(), &mut stdout.lock()))
70}
71
72/// The testable core of [`run`]: resolve settings and the database path,
73/// open the right handle, run the command into `out`, return the exit code
74/// (`0` ok, `1` soft miss / locked, `2` error). Errors go to stderr.
75fn run_parsed(cli: Cli, out: &mut impl Write) -> u8 {
76    if let Command::Help { topic } = &cli.command {
77        return execute_help(topic, cli.json, out);
78    }
79
80    // Read config.toml once: the shared loader builds engine/embedder/
81    // maintenance settings; the CLI reads its own `[maintenance].batch_size`
82    // from the same table (used by `import` below).
83    let table = match plugmem_host::read_config(cli.config.as_deref()) {
84        Ok(t) => t,
85        Err(e) => return report_err(&e.into()),
86    };
87    let cfg_batch_size = read_batch_size(table.as_ref());
88    let mut settings = match Settings::from_table(table.as_ref()) {
89        Ok(s) => s,
90        Err(e) => return report_err(&e.into()),
91    };
92    // Anything in config.toml nobody claimed. To stderr, not `out`: it is a
93    // note about the environment, and it must not land in the middle of `--json`
94    // output that something is piping.
95    for warning in &settings.warnings {
96        eprintln!("plugmem: {warning}");
97    }
98    // A workspace is opt-in: with no flag, no environment variable and no
99    // `[workspace].dir`, `root` is `None` and everything below behaves exactly
100    // as it did before workspaces existed — `--db` is a path, and the
101    // `workspace` group says there is nothing to manage.
102    let root = workspace::resolve_root(cli.workspace.as_deref(), &settings);
103    if let Command::Workspace { command } = &cli.command {
104        return match workspace::execute(command, root, settings, cli.json, out) {
105            Ok(code) => code,
106            Err(e) => {
107                let _ = out.flush();
108                report_err(&e)
109            }
110        };
111    }
112    let path = resolve_db_path(
113        cli.db.as_deref(),
114        settings.database_path.as_deref(),
115        root.as_ref(),
116    );
117    if let Err(e) = workspace::ensure_dir(&path, root.as_ref()) {
118        return report_err(&e);
119    }
120
121    // `recover` is a standalone salvage on file paths — it opens the source
122    // itself (under an exclusive lock) and writes a fresh destination, so it
123    // runs before the normal open. `scrub` is a byte-level container check over
124    // a read-only (shared-lock) open, which requires a checkpointed database.
125    match &cli.command {
126        Command::Recover { dst } => return do_recover(&path, dst, &settings, cli.json, out),
127        Command::Scrub => return do_scrub(&path, &settings, cli.json, out),
128        // The interactive session opens one handle and reads commands from
129        // stdin, so it is dispatched before the per-command open below. The
130        // read-only variant observes another process's writer over a shared
131        // mmap; the default variant opens the single writer handle.
132        Command::Repl { read_only: true } => {
133            return run_repl_ro(&path, settings, cli.json, io::stdin().lock(), out);
134        }
135        Command::Repl { read_only: false } => {
136            return run_repl(&path, settings, cli.json, io::stdin().lock(), out);
137        }
138        _ => {}
139    }
140
141    // Read-only commands open the snapshot zero-copy (mmap, shared lock) and
142    // coexist with a live writer process (Variant 2 MVCC) — they never take the
143    // writer lock. `verify` is a pure content check, so it belongs here too.
144    // `recall` embeds its text query *before* the open (mirroring the host's
145    // "embed outside the lock" rule) so it can search by vector on the read-only
146    // path, which carries no embedder. A dirty (un-checkpointed) journal forbids
147    // a read-only open, so those fall through to the read-write path.
148    let readonly_ok = matches!(
149        &cli.command,
150        Command::Show { .. }
151            | Command::Stats
152            | Command::Tags { .. }
153            | Command::Export
154            | Command::Verify
155            | Command::Recall { .. }
156    );
157    if readonly_ok {
158        let recall_vector = match embed_recall_query(&mut settings, &cli.command) {
159            Ok(v) => v,
160            Err(e) => return report_err(&e),
161        };
162        match Database::open_readonly(&path, settings.config.clone()) {
163            Ok(ro) => {
164                let recall_space = recall_vector.as_ref().and_then(|_| {
165                    settings
166                        .embedder
167                        .as_ref()
168                        .map(|embedder| embedder.space_id())
169                });
170                return execute_ro(
171                    &ro,
172                    &cli.command,
173                    recall_vector.as_deref(),
174                    recall_space,
175                    cli.json,
176                    out,
177                );
178            }
179            Err(HostError::Locked { path }) => return report_locked(&path),
180            // Any other failure — a missing snapshot (fresh db), a dirty
181            // journal (NeedsCheckpoint), or a corrupt image — is handled by
182            // the read-write path: it creates/checkpoints, or surfaces the
183            // same corruption as a typed error.
184            Err(_) => {}
185        }
186    }
187
188    // `cfg_batch_size` was read from the config table above (before `open`
189    // consumes `settings`); the `--batch` flag still wins over it.
190    let db = match settings.open(&path) {
191        Ok(db) => db,
192        Err(HostError::Locked { path }) => return report_locked(&path),
193        Err(e) => return report_err(&CliError::Host(e)),
194    };
195    // Import is dispatched here, not in `execute`: its batch size comes from the
196    // `--batch` flag or `[maintenance].batch_size` (flag > config > default).
197    if let Command::Import { file, batch } = &cli.command {
198        let batch_size = batch
199            .or(cfg_batch_size.map(|n| n as usize))
200            .unwrap_or(DEFAULT_IMPORT_BATCH)
201            .max(1);
202        return match do_import(&db, now_ms(), file, batch_size, out) {
203            Ok(report) => {
204                if cli.json {
205                    writeln!(
206                        out,
207                        "{}",
208                        json!({ "imported": report.facts, "edges": report.edges })
209                    )
210                    .ok();
211                } else if report.edges == 0 {
212                    writeln!(out, "imported {} facts", report.facts).ok();
213                } else {
214                    writeln!(
215                        out,
216                        "imported {} facts and {} edges",
217                        report.facts, report.edges
218                    )
219                    .ok();
220                }
221                0
222            }
223            Err(e) => {
224                let _ = out.flush();
225                report_err(&e)
226            }
227        };
228    }
229    match execute(&db, &cli.command, cli.json, now_ms(), out) {
230        Ok(code) => code,
231        Err(e) => {
232            let _ = out.flush();
233            report_err(&e)
234        }
235    }
236}
237
238/// Default facts-per-batch for `import` when neither `--batch` nor
239/// `[maintenance].batch_size` is set — safe for provider batch limits.
240const DEFAULT_IMPORT_BATCH: usize = 128;
241
242/// Writes one error, plus whatever follow-up it carries.
243///
244/// Every path that shows a failure goes through here — the one-shot commands
245/// and both repls — so a message cannot say one thing in one of them and
246/// something else in another. The follow-up matters for a pool ceiling in
247/// particular: on its own that error is a bare byte count.
248fn write_err(out: &mut impl Write, e: &CliError) {
249    let _ = match e {
250        CliError::Usage(msg) => writeln!(out, "plugmem: {msg}"),
251        CliError::Host(err) => {
252            writeln!(out, "plugmem: {err}").and_then(|()| match err.capacity_hint() {
253                Some(hint) => writeln!(out, "plugmem: {hint}"),
254                None => Ok(()),
255            })
256        }
257    };
258}
259
260/// Prints an error to stderr and returns its exit code (`2`).
261fn report_err(e: &CliError) -> u8 {
262    write_err(&mut std::io::stderr(), e);
263    2
264}
265
266/// Prints the locked message and returns its exit code (`1`).
267fn report_locked(path: &std::path::Path) -> u8 {
268    eprintln!(
269        "plugmem: database is locked by another process: {}",
270        path.display()
271    );
272    1
273}
274
275/// Database path precedence: `--db` flag > `$PLUGMEM_DB` >
276/// `[database].path` > the platform default.
277///
278/// With a workspace configured, the first two may name a memory instead of a
279/// path — see [`workspace::resolve_target`]. `[database].path` and the platform
280/// default are always paths: they are file settings, not names.
281fn resolve_db_path(
282    flag: Option<&str>,
283    config_path: Option<&std::path::Path>,
284    root: Option<&PathBuf>,
285) -> PathBuf {
286    flag.map(|value| workspace::resolve_target(value, root))
287        .or_else(|| {
288            std::env::var_os(ENV_DB).map(|v| workspace::resolve_target(&v.to_string_lossy(), root))
289        })
290        .or_else(|| config_path.map(PathBuf::from))
291        .or_else(plugmem_host::default_database_path)
292        .unwrap_or_else(|| PathBuf::from(DEFAULT_DB))
293}
294
295/// Render the opt-in detailed help topics without reading a config file or
296/// opening a database.
297fn execute_help(topic: &HelpTopic, json_output: bool, out: &mut impl Write) -> u8 {
298    match topic {
299        HelpTopic::Settings => {
300            if json_output {
301                let help = plugmem_host::settings_help();
302                let settings: Vec<_> = help
303                    .docs()
304                    .iter()
305                    .map(|doc| {
306                        json!({
307                            "section": doc.section,
308                            "key": doc.key,
309                            "type": doc.value_type,
310                            "default": doc.default,
311                            "description": doc.description,
312                            "scope": doc.scope.as_str(),
313                        })
314                    })
315                    .collect();
316                let value = json!({
317                    "topic": "settings",
318                    "config_path_precedence": help.config_path_precedence(),
319                    "default_config_path": plugmem_host::default_config_path()
320                        .map(|path| path.display().to_string()),
321                    "settings": settings,
322                });
323                writeln!(out, "{value}").ok();
324            } else {
325                write!(out, "{}", plugmem_host::settings_help().render_human()).ok();
326            }
327            0
328        }
329    }
330}
331
332/// Runs a read-only command over a zero-copy [`ReadOnlyDatabase`] (mmap,
333/// shared lock). Only the commands `run_parsed` routes here appear.
334fn execute_ro(
335    ro: &ReadOnlyDatabase,
336    cmd: &Command,
337    recall_vector: Option<&[f32]>,
338    recall_space: Option<&str>,
339    json: bool,
340    out: &mut impl Write,
341) -> u8 {
342    match cmd {
343        Command::Recall { .. } => {
344            match with_recall_query(cmd, now_ms(), recall_vector, |q| match recall_space {
345                Some(space) => ro.recall_in_space(q, space),
346                None => ro.recall(q),
347            }) {
348                Ok(res) => {
349                    render_recall(&res, json, out);
350                    0
351                }
352                Err(e) => report_err(&CliError::Host(e)),
353            }
354        }
355        Command::Show { id } => render_show(ro.get(FactId(*id)), *id, json, out),
356        Command::Stats => {
357            render_stats(&ro.stats(), json, out);
358            0
359        }
360        Command::Tags {
361            prefix,
362            cursor,
363            limit,
364        } => match ro.list_tags(TagQuery {
365            prefix: prefix.as_deref(),
366            cursor: cursor.as_deref(),
367            limit: *limit,
368        }) {
369            Ok(page) => {
370                render_tags(&page, json, out);
371                0
372            }
373            Err(e) => report_err(&CliError::Host(e)),
374        },
375        Command::Export => {
376            ro.export_each(|f| write_export_line(out, &f));
377            ro.export_edges_each(|src, rel, dst, fact| write_export_edge(out, src, rel, dst, fact));
378            0
379        }
380        // A clean image returns Ok; corruption is a typed error mapped to exit 2.
381        Command::Verify => match ro.verify() {
382            Ok(()) => {
383                if json {
384                    writeln!(out, "{}", json!({ "ok": true })).ok();
385                } else {
386                    writeln!(out, "integrity ok").ok();
387                }
388                0
389            }
390            Err(e) => report_err(&CliError::Host(e)),
391        },
392        _ => unreachable!("execute_ro only receives read-only commands"),
393    }
394}
395
396/// Runs one command against an open database, writing the result to `out`.
397/// Returns the process exit code (`0` ok, `1` soft miss). Split from
398/// [`run`] so tests drive it directly against a temp database.
399fn execute(
400    db: &Database,
401    cmd: &Command,
402    json: bool,
403    now: u64,
404    out: &mut impl Write,
405) -> Result<u8, CliError> {
406    match cmd {
407        Command::Remember {
408            text,
409            entity,
410            tags,
411            links,
412            meta,
413            valid_from,
414            vector,
415            guarded,
416        } => {
417            if *guarded {
418                let outcome = do_guarded_remember(
419                    db,
420                    now,
421                    text,
422                    entity,
423                    tags,
424                    links,
425                    meta,
426                    *valid_from,
427                    vector,
428                )?;
429                render_guarded_remember(&outcome, json, out);
430            } else {
431                let outcome = do_remember(
432                    db,
433                    now,
434                    text,
435                    entity,
436                    tags,
437                    links,
438                    meta,
439                    *valid_from,
440                    vector,
441                    None,
442                )?;
443                render_remember(&outcome, json, out);
444            }
445            Ok(0)
446        }
447        Command::Revise {
448            id,
449            text,
450            entity,
451            tags,
452            links,
453            meta,
454            valid_from,
455            vector,
456        } => {
457            let outcome = do_remember(
458                db,
459                now,
460                text,
461                entity,
462                tags,
463                links,
464                meta,
465                *valid_from,
466                vector,
467                Some(FactId(*id)),
468            )?;
469            render_remember(&outcome, json, out);
470            Ok(0)
471        }
472        Command::Recall { .. } => {
473            let res = with_recall_query(cmd, now, None, |q| db.recall(q))?;
474            render_recall(&res, json, out);
475            Ok(0)
476        }
477        Command::Forget { ids } => {
478            let fact_ids: Vec<FactId> = ids.iter().copied().map(FactId).collect();
479            let results = db.forget_many(now, &fact_ids)?;
480            if json {
481                if ids.len() == 1 {
482                    writeln!(out, "{}", json!({ "id": ids[0], "forgotten": results[0] })).ok();
483                } else {
484                    let items: Vec<_> = ids
485                        .iter()
486                        .zip(&results)
487                        .map(|(id, fresh)| json!({ "id": id, "forgotten": fresh }))
488                        .collect();
489                    writeln!(out, "{}", json!(items)).ok();
490                }
491            } else {
492                for (id, fresh) in ids.iter().zip(&results) {
493                    if *fresh {
494                        writeln!(out, "forgot fact {id}").ok();
495                    } else {
496                        writeln!(out, "fact {id} was already gone").ok();
497                    }
498                }
499            }
500            Ok(0)
501        }
502        Command::Tags {
503            prefix,
504            cursor,
505            limit,
506        } => {
507            let page = db.list_tags(TagQuery {
508                prefix: prefix.as_deref(),
509                cursor: cursor.as_deref(),
510                limit: *limit,
511            })?;
512            render_tags(&page, json, out);
513            Ok(0)
514        }
515        Command::RemoveTag { tag } => {
516            let report = db.remove_tag(now, tag)?;
517            if json {
518                writeln!(
519                    out,
520                    "{}",
521                    json!({ "tag": tag, "affected": report.affected })
522                )
523                .ok();
524            } else {
525                writeln!(
526                    out,
527                    "removed tag {tag:?} from {} current facts",
528                    report.affected
529                )
530                .ok();
531            }
532            Ok(0)
533        }
534        Command::Link {
535            src,
536            rel,
537            dst,
538            provenance,
539        } => {
540            db.link(LinkInput {
541                now,
542                src,
543                rel,
544                dst,
545                provenance: provenance.map(FactId),
546            })?;
547            if json {
548                writeln!(out, "{}", json!({ "src": src, "rel": rel, "dst": dst })).ok();
549            } else {
550                writeln!(out, "linked {src} -{rel}-> {dst}").ok();
551            }
552            Ok(0)
553        }
554        Command::Unlink { src, rel, dst } => {
555            let fresh = db.unlink(UnlinkInput { now, src, rel, dst })?;
556            if json {
557                writeln!(
558                    out,
559                    "{}",
560                    json!({ "src": src, "rel": rel, "dst": dst, "unlinked": fresh })
561                )
562                .ok();
563            } else if fresh {
564                writeln!(out, "unlinked {src} -{rel}-> {dst}").ok();
565            } else {
566                writeln!(out, "edge {src} -{rel}-> {dst} was already absent").ok();
567            }
568            Ok(0)
569        }
570        Command::Show { id } => Ok(render_show(db.get(FactId(*id)), *id, json, out)),
571        Command::Stats => {
572            render_stats(&db.stats(), json, out);
573            Ok(0)
574        }
575        Command::Export => {
576            db.export_each(|f| write_export_line(out, &f));
577            db.export_edges_each(|src, rel, dst, fact| write_export_edge(out, src, rel, dst, fact));
578            Ok(0)
579        }
580        Command::Maintain {
581            mode,
582            reembed,
583            batch_size,
584        } => {
585            if *reembed {
586                let report = db.reembed_with_batch(
587                    now,
588                    batch_size.unwrap_or(plugmem_host::DEFAULT_REEMBED_BATCH_SIZE),
589                )?;
590                if json {
591                    writeln!(
592                        out,
593                        "{}",
594                        json!({
595                            "embedded": report.embedded,
596                            "previous_dim": report.previous_dim,
597                            "new_dim": report.new_dim,
598                            "previous_space": report.previous_space,
599                            "new_space": report.new_space,
600                            "vector_bytes": report.vector_bytes,
601                            "hnsw_indexed": report.hnsw_indexed,
602                        })
603                    )
604                    .ok();
605                } else {
606                    writeln!(
607                        out,
608                        "reembedded: {} facts, {} -> {} dimensions, {:?} -> {:?}, {} vector bytes, hnsw {}",
609                        report.embedded,
610                        report.previous_dim,
611                        report.new_dim,
612                        report.previous_space,
613                        report.new_space,
614                        report.vector_bytes,
615                        report.hnsw_indexed,
616                    )
617                    .ok();
618                }
619                return Ok(0);
620            }
621            let report = db.maintain_with_options(now, maintenance_options(*mode))?;
622            if json {
623                writeln!(
624                    out,
625                    "{}",
626                    json!({
627                        "purged": report.purged,
628                        "bytes_before": report.bytes_before,
629                        "bytes_after": report.bytes_after,
630                        "no_op": report.no_op,
631                        "tombstones_before": report.tombstones_before,
632                        "facts_before": report.facts_before,
633                        "facts_after": report.facts_after,
634                        "vectors_before": report.vectors_before,
635                        "vectors_after": report.vectors_after,
636                        "hnsw_indexed_before": report.hnsw_indexed_before,
637                        "hnsw_indexed_after": report.hnsw_indexed_after,
638                        "structural_compacted": report.structural_compacted,
639                        "bm25_compacted": report.bm25_compacted,
640                        "bm25_reindexed": report.bm25_reindexed,
641                        "hnsw_rebuilt": report.hnsw_rebuilt,
642                        "hnsw_remapped": report.hnsw_remapped,
643                        "hnsw_inserted": report.hnsw_inserted,
644                        "edges_compacted": report.edges_compacted,
645                        "edges_before": report.edges_before,
646                        "edge_versions_before": report.edge_versions_before,
647                    })
648                )
649                .ok();
650            } else {
651                writeln!(
652                    out,
653                    "maintained: purged {}, {} -> {} bytes, hnsw +{}, bm25 {}{}{}",
654                    report.purged,
655                    report.bytes_before,
656                    report.bytes_after,
657                    report.hnsw_inserted,
658                    if report.bm25_reindexed {
659                        "reindexed"
660                    } else if report.bm25_compacted {
661                        "compacted"
662                    } else {
663                        "unchanged"
664                    },
665                    if report.edges_compacted {
666                        ", edges repacked"
667                    } else {
668                        ""
669                    },
670                    if report.no_op { " (no-op)" } else { "" }
671                )
672                .ok();
673            }
674            Ok(0)
675        }
676        Command::Checkpoint => {
677            db.checkpoint(now)?;
678            if json {
679                writeln!(out, "{}", json!({ "ok": true })).ok();
680            } else {
681                writeln!(out, "checkpointed: journal flushed to snapshot").ok();
682            }
683            Ok(0)
684        }
685        Command::Verify => {
686            // A clean image returns Ok; corruption is a typed error the caller
687            // maps to exit 2.
688            db.verify()?;
689            if json {
690                writeln!(out, "{}", json!({ "ok": true })).ok();
691            } else {
692                writeln!(out, "integrity ok").ok();
693            }
694            Ok(0)
695        }
696        // Handled in `run_parsed` (Import needs `settings` for its batch size).
697        Command::Scrub
698        | Command::Recover { .. }
699        | Command::Repl { .. }
700        | Command::Import { .. }
701        | Command::Workspace { .. }
702        | Command::Help { .. } => {
703            unreachable!("this command is dispatched before execute")
704        }
705    }
706}
707
708/// Salvages `src` into a fresh `dst`: `Database::recover` opens
709/// the source under an exclusive lock, drops the content-corrupt facts, and
710/// writes a clean disk-first copy. The source is left untouched.
711fn do_recover(src: &Path, dst: &Path, settings: &Settings, json: bool, out: &mut impl Write) -> u8 {
712    match Database::recover(src, dst, settings.config.clone(), now_ms()) {
713        Ok(r) => {
714            if json {
715                writeln!(
716                    out,
717                    "{}",
718                    json!({
719                        "kept": r.kept,
720                        "dropped_text": r.dropped_text,
721                        "dropped_vector": r.dropped_vector,
722                        "dropped_metadata": r.dropped_metadata,
723                        "dst": dst.display().to_string(),
724                    })
725                )
726                .ok();
727            } else {
728                writeln!(
729                    out,
730                    "recovered to {}: kept {}, dropped {} text + {} vector + {} metadata",
731                    dst.display(),
732                    r.kept,
733                    r.dropped_text,
734                    r.dropped_vector,
735                    r.dropped_metadata
736                )
737                .ok();
738            }
739            0
740        }
741        Err(HostError::Locked { path }) => report_locked(&path),
742        Err(e) => report_err(&CliError::Host(e)),
743    }
744}
745
746/// Runs a byte-level container scrub over a read-only (shared-lock) open. A
747/// clean image exits 0; the first damaged section is a typed error (exit 2). A
748/// dirty journal forbids the read-only open — the reported `NeedsCheckpoint`
749/// tells the caller to run `maintain` first.
750fn do_scrub(path: &Path, settings: &Settings, json: bool, out: &mut impl Write) -> u8 {
751    let ro = match Database::open_readonly(path, settings.config.clone()) {
752        Ok(ro) => ro,
753        Err(HostError::Locked { path }) => return report_locked(&path),
754        Err(e) => return report_err(&CliError::Host(e)),
755    };
756    let scrub = match ro.scrub() {
757        Ok(s) => s,
758        Err(e) => return report_err(&CliError::Host(e)),
759    };
760    let mut done = 0u64;
761    let mut total = 0u64;
762    for step in scrub {
763        match step {
764            Ok(p) => {
765                done = p.done_bytes;
766                total = p.total_bytes;
767            }
768            Err(e) => return report_err(&CliError::Host(e)),
769        }
770    }
771    if json {
772        writeln!(out, "{}", json!({ "ok": true, "bytes": done })).ok();
773    } else {
774        writeln!(out, "scrub ok: {done}/{total} bytes verified").ok();
775    }
776    0
777}
778
779/// Parses a single REPL line: the subcommand grammar, with no leading binary
780/// name (the line is `recall tokio`, not `plugmem recall tokio`).
781#[derive(Parser)]
782#[command(
783    no_binary_name = true,
784    name = "plugmem",
785    disable_help_subcommand = true
786)]
787struct ReplLine {
788    #[command(subcommand)]
789    command: Command,
790}
791
792/// Splits a REPL line into tokens, honoring single/double quotes so
793/// `remember "two words"` is one argument. No escape handling — a quote runs to
794/// its match or the end of the line.
795fn split_line(line: &str) -> Vec<String> {
796    let mut tokens = Vec::new();
797    let mut cur = String::new();
798    let mut quote: Option<char> = None;
799    let mut has = false;
800    for c in line.chars() {
801        match quote {
802            Some(q) => {
803                if c == q {
804                    quote = None;
805                } else {
806                    cur.push(c);
807                }
808            }
809            None if c == '"' || c == '\'' => {
810                quote = Some(c);
811                has = true;
812            }
813            None if c.is_whitespace() => {
814                if has {
815                    tokens.push(std::mem::take(&mut cur));
816                    has = false;
817                }
818            }
819            None => {
820                cur.push(c);
821                has = true;
822            }
823        }
824    }
825    if has {
826        tokens.push(cur);
827    }
828    tokens
829}
830
831/// Runs the interactive session over one open writer handle: read a line, parse
832/// it as a subcommand, run it against the in-memory engine, repeat. The engine
833/// stays resident, so each command is host-speed (no per-command reload). The
834/// session checkpoints on exit, leaving a read-ready file. Prompts and the
835/// banner go to stderr so stdout carries only command output.
836fn run_repl(
837    path: &Path,
838    settings: Settings,
839    json: bool,
840    input: impl BufRead,
841    out: &mut impl Write,
842) -> u8 {
843    let db = match settings.open(path) {
844        Ok(db) => db,
845        Err(HostError::Locked { path }) => return report_locked(&path),
846        Err(e) => return report_err(&CliError::Host(e)),
847    };
848    eprintln!("plugmem repl — one open handle, host speed. `help` for verbs, `exit` to quit.");
849    eprint!("plugmem> ");
850    for line in input.lines() {
851        let Ok(line) = line else { break };
852        let line = line.trim();
853        if line.is_empty() {
854            eprint!("plugmem> ");
855            continue;
856        }
857        if line == "exit" || line == "quit" {
858            break;
859        } else if line == "help" {
860            writeln!(
861                out,
862                "verbs: remember recall revise forget tags remove-tag link unlink show stats maintain checkpoint \
863                 verify export import  (scrub/recover stay one-shot)  exit"
864            )
865            .ok();
866        } else {
867            run_repl_line(&db, line, json, out);
868        }
869        eprint!("plugmem> ");
870    }
871    eprintln!();
872    // Leave the database checkpointed (read-ready) for the next opener.
873    match db.checkpoint(now_ms()) {
874        Ok(()) => 0,
875        Err(e) => report_err(&CliError::Host(e)),
876    }
877}
878
879/// Parses and runs one non-meta REPL line, reporting errors to `out` without
880/// ending the session.
881fn run_repl_line(db: &Database, line: &str, json: bool, out: &mut impl Write) {
882    let cmd = match ReplLine::try_parse_from(split_line(line)) {
883        Ok(r) => r.command,
884        // clap's message (usage / unknown command / `--help`) — print, continue.
885        Err(e) => {
886            let _ = writeln!(out, "{e}");
887            return;
888        }
889    };
890    match &cmd {
891        Command::Repl { .. } => {
892            let _ = writeln!(out, "already in a repl session");
893        }
894        Command::Scrub | Command::Recover { .. } => {
895            let _ = writeln!(
896                out,
897                "scrub/recover are one-shot commands; run them outside the repl"
898            );
899        }
900        _ => {
901            if let Err(e) = execute(db, &cmd, json, now_ms(), out) {
902                write_err(out, &e);
903            }
904        }
905    }
906}
907
908/// Runs the interactive session **read-only** over one open
909/// [`ReadOnlyDatabase`] (a shared, zero-copy mmap): it observes another
910/// process's writer at the generation it opened on. Only the read verbs run;
911/// writes and one-shot commands are refused. Two extra meta-verbs make the
912/// cross-process freshness observable by hand — `generation` prints the pinned
913/// snapshot number, and `refresh` advances to the writer's latest published
914/// checkpoint (see [`ReadOnlyDatabase::refresh`](plugmem_host::ReadOnlyDatabase::refresh)).
915///
916/// These two verbs exist **only** in this mode. A normal (writer) `repl` and
917/// any one-shot command already see the freshest data — read-your-writes over
918/// the overlay, or a fresh open per command — so there is nothing to refresh
919/// there. This session never writes: it does not checkpoint on exit.
920fn run_repl_ro(
921    path: &Path,
922    mut settings: Settings,
923    json: bool,
924    input: impl BufRead,
925    out: &mut impl Write,
926) -> u8 {
927    let mut ro = match Database::open_readonly(path, settings.config.clone()) {
928        Ok(ro) => ro,
929        Err(HostError::Locked { path }) => return report_locked(&path),
930        // A dirty (un-checkpointed) journal, a fresh database with no published
931        // generation, or a corrupt image — surfaced as a typed error.
932        Err(e) => return report_err(&CliError::Host(e)),
933    };
934    eprintln!(
935        "plugmem repl --read-only — observing generation {} of another process's writer. \
936         `help` for verbs, `refresh`/`generation` for cross-process freshness, `exit` to quit.",
937        ro.generation()
938    );
939    eprint!("plugmem(ro)> ");
940    for line in input.lines() {
941        let Ok(line) = line else { break };
942        let line = line.trim();
943        if line.is_empty() {
944            eprint!("plugmem(ro)> ");
945            continue;
946        }
947        match line {
948            "exit" | "quit" => break,
949            "help" => {
950                writeln!(
951                    out,
952                    "read verbs: recall show stats export verify  \
953                     freshness: generation refresh  exit  \
954                     (writes and scrub/recover are refused in a read-only session)"
955                )
956                .ok();
957            }
958            // Freshness meta-verbs — only meaningful for a read-only observer of
959            // another process's writer (a writer repl sees its own writes at once).
960            "generation" => {
961                let g = ro.generation();
962                if json {
963                    writeln!(out, "{}", json!({ "generation": g })).ok();
964                } else {
965                    writeln!(out, "generation {g}").ok();
966                }
967            }
968            "refresh" => match ro.refresh() {
969                Ok(advanced) => {
970                    let g = ro.generation();
971                    if json {
972                        writeln!(out, "{}", json!({ "advanced": advanced, "generation": g })).ok();
973                    } else if advanced {
974                        writeln!(out, "refreshed → generation {g}").ok();
975                    } else {
976                        writeln!(out, "already current → generation {g}").ok();
977                    }
978                }
979                Err(e) => write_err(out, &CliError::Host(e)),
980            },
981            _ => run_repl_ro_line(&ro, &mut settings, line, json, out),
982        }
983        eprint!("plugmem(ro)> ");
984    }
985    eprintln!();
986    // Read-only: nothing to checkpoint, the writer owns the file.
987    0
988}
989
990/// Parses and runs one non-meta line of a read-only repl, refusing anything but
991/// the read verbs (writes/one-shot are not available without the writer lock).
992fn run_repl_ro_line(
993    ro: &ReadOnlyDatabase,
994    settings: &mut Settings,
995    line: &str,
996    json: bool,
997    out: &mut impl Write,
998) {
999    let cmd = match ReplLine::try_parse_from(split_line(line)) {
1000        Ok(r) => r.command,
1001        Err(e) => {
1002            let _ = writeln!(out, "{e}");
1003            return;
1004        }
1005    };
1006    let readable = matches!(
1007        &cmd,
1008        Command::Show { .. }
1009            | Command::Stats
1010            | Command::Tags { .. }
1011            | Command::Export
1012            | Command::Verify
1013            | Command::Recall { .. }
1014    );
1015    if !readable {
1016        let _ = writeln!(
1017            out,
1018            "read-only session: only recall/show/stats/tags/export/verify run \
1019             (plus refresh/generation); writes and one-shot commands need a writer handle"
1020        );
1021        return;
1022    }
1023    // Embed a text recall query up front, exactly like the one-shot read-only
1024    // path — the read-only handle carries no embedder of its own.
1025    let recall_vector = match embed_recall_query(settings, &cmd) {
1026        Ok(v) => v,
1027        Err(e) => {
1028            write_err(out, &e);
1029            return;
1030        }
1031    };
1032    let recall_space = recall_vector.as_ref().and_then(|_| {
1033        settings
1034            .embedder
1035            .as_ref()
1036            .map(|embedder| embedder.space_id())
1037    });
1038    let _ = execute_ro(ro, &cmd, recall_vector.as_deref(), recall_space, json, out);
1039}
1040
1041/// Embeds a `recall` command's text query into a vector using the configured
1042/// embedder, so the read-only path (which carries no embedder) can still search
1043/// by meaning while a writer process holds the database. Returns `None` when the
1044/// command is not `recall`, carries no query text, or no embedder is configured
1045/// — recall then falls back to lexical/structural sources. Mirrors the host's
1046/// "embed before the lock" rule; the embed happens before the open
1047/// so a locked database only costs the embed on the rare read-write fallback.
1048fn embed_recall_query(
1049    settings: &mut Settings,
1050    cmd: &Command,
1051) -> Result<Option<Vec<f32>>, CliError> {
1052    let Command::Recall {
1053        query: Some(text),
1054        vector,
1055        ..
1056    } = cmd
1057    else {
1058        return Ok(None);
1059    };
1060    // An explicit `--vector` replaces the embedder, so there is nothing to
1061    // embed and no call to make.
1062    if !vector.is_empty() {
1063        return Ok(None);
1064    }
1065    let Some(embedder) = settings.embedder.as_ref() else {
1066        return Ok(None);
1067    };
1068    let mut vectors = embedder.embed(&[text.as_str()]).map_err(CliError::Host)?;
1069    Ok(vectors.pop())
1070}
1071
1072/// Builds the [`RecallQuery`] for a `recall` command and passes it to `f`.
1073/// A closure (not a return) because the query borrows temporary tag/entity
1074/// slices that must outlive the call. Used by both the read-write and
1075/// read-only paths.
1076fn with_recall_query<R>(
1077    cmd: &Command,
1078    now: u64,
1079    override_vector: Option<&[f32]>,
1080    f: impl FnOnce(RecallQuery<'_>) -> R,
1081) -> R {
1082    let Command::Recall {
1083        query,
1084        tags,
1085        entities,
1086        as_of,
1087        range,
1088        k,
1089        closed,
1090        token_budget,
1091        ef,
1092        graph_depth,
1093        vector,
1094    } = cmd
1095    else {
1096        unreachable!("with_recall_query called on a non-recall command");
1097    };
1098    let tag_refs: Vec<&str> = tags.iter().map(String::as_str).collect();
1099    let ent_refs: Vec<&str> = entities.iter().map(String::as_str).collect();
1100    let range_pair = range.as_ref().map(|v| (v[0], v[1]));
1101    // Precedence, matching the host: an explicit `--vector` wins outright and
1102    // nothing is sent to the embedder. Otherwise `override_vector` carries the
1103    // text the CLI embedded for the read-only path; on the read-write path both
1104    // are `None` and the host embeds inside `recall`.
1105    let explicit = (!vector.is_empty()).then_some(vector.as_slice());
1106    let q = RecallQuery {
1107        now,
1108        text: query.as_deref(),
1109        vector: explicit.or(override_vector),
1110        tags: &tag_refs,
1111        entities: &ent_refs,
1112        as_of: *as_of,
1113        range: range_pair,
1114        k: *k,
1115        token_budget: *token_budget,
1116        include_closed: *closed,
1117        ef: *ef,
1118        graph_depth: *graph_depth,
1119    };
1120    f(q)
1121}
1122
1123/// Renders a recall result — the engine's block (human) or facts + block
1124/// (JSON).
1125fn render_recall(res: &RecallResult, json: bool, out: &mut impl Write) {
1126    if json {
1127        let facts: Vec<_> = res
1128            .facts
1129            .iter()
1130            .map(|f| {
1131                json!({
1132                    "id": f.id.0,
1133                    "score": f.score,
1134                    "sources": f.sources,
1135                    "recorded_at": f.recorded_at,
1136                    "valid_from": f.valid_from,
1137                    "valid_to": open_or(f.valid_to),
1138                })
1139            })
1140            .collect();
1141        // The edges the graph source walked. The MCP server and the napi
1142        // binding have always returned these; the CLI built its JSON by hand
1143        // and dropped them, which made `--provenance` write-only from here:
1144        // recordable, then unreadable.
1145        let edges: Vec<_> = res
1146            .edges
1147            .iter()
1148            .map(|e| {
1149                json!({
1150                    "src": e.src.0,
1151                    "rel": e.rel.0,
1152                    "dst": e.dst.0,
1153                    "provenance": (e.provenance != FactId::NONE).then_some(e.provenance.0),
1154                })
1155            })
1156            .collect();
1157        writeln!(
1158            out,
1159            "{}",
1160            json!({
1161                "facts": facts,
1162                "edges": edges,
1163                "rendered": res.rendered,
1164                "truncated": res.truncated,
1165            })
1166        )
1167        .ok();
1168    } else if res.rendered.is_empty() {
1169        writeln!(out, "(nothing recalled)").ok();
1170    } else {
1171        writeln!(out, "{}", res.rendered).ok();
1172    }
1173}
1174
1175/// Renders one fact's card. Returns the exit code (`0` found, `1` missing).
1176fn render_show(
1177    fact: Option<plugmem_host::FactSnapshot>,
1178    id: u32,
1179    json: bool,
1180    out: &mut impl Write,
1181) -> u8 {
1182    let Some(fact) = fact else {
1183        if json {
1184            writeln!(out, "{}", json!({ "id": id, "found": false })).ok();
1185        } else {
1186            writeln!(out, "fact {id} not found").ok();
1187        }
1188        return 1;
1189    };
1190    let r = &fact.record;
1191    if json {
1192        writeln!(
1193            out,
1194            "{}",
1195            json!({
1196                "id": r.id.0,
1197                "text": fact.text,
1198                "recorded_at": r.recorded_at,
1199                "valid_from": r.valid_from,
1200                "valid_to": open_or(r.valid_to),
1201                "closed": r.is_closed(),
1202                "tombstone": r.is_tombstone(),
1203                "revises": (r.revises != FactId::NONE).then_some(r.revises.0),
1204                "metadata": fact.metadata,
1205            })
1206        )
1207        .ok();
1208    } else {
1209        writeln!(out, "fact {}", r.id.0).ok();
1210        writeln!(out, "  text        {}", fact.text).ok();
1211        writeln!(out, "  recorded_at {}", r.recorded_at).ok();
1212        write!(out, "  valid       [{}, ", r.valid_from).ok();
1213        match r.valid_to {
1214            VALID_TO_OPEN => writeln!(out, "open)").ok(),
1215            to => writeln!(out, "{to})").ok(),
1216        };
1217        if r.revises != FactId::NONE {
1218            writeln!(out, "  revises     fact {}", r.revises.0).ok();
1219        }
1220        if !fact.metadata.is_empty() {
1221            let rendered = fact
1222                .metadata
1223                .iter()
1224                .map(|(k, v)| format!("{k}={v}"))
1225                .collect::<Vec<_>>()
1226                .join(", ");
1227            writeln!(out, "  metadata    {rendered}").ok();
1228        }
1229        if r.is_tombstone() {
1230            writeln!(out, "  state       tombstoned").ok();
1231        }
1232    }
1233    0
1234}
1235
1236/// Translates the command-line mode into engine options.
1237///
1238/// `auto` keeps the bounded HNSW budget that makes it safe to run often;
1239/// every explicit mode takes the budget the engine defines for it.
1240fn maintenance_options(mode: MaintainMode) -> MaintenanceOptions {
1241    match MaintenanceMode::from(mode) {
1242        MaintenanceMode::Auto => MaintenanceOptions::auto(),
1243        MaintenanceMode::Full => MaintenanceOptions::full(),
1244        mode => MaintenanceOptions {
1245            mode,
1246            ..MaintenanceOptions::auto()
1247        },
1248    }
1249}
1250
1251/// Renders engine size counters.
1252fn render_stats(s: &Stats, json: bool, out: &mut impl Write) {
1253    if json {
1254        writeln!(
1255            out,
1256            "{}",
1257            json!({
1258                "facts": s.facts,
1259                "entities": s.entities,
1260                "terms": s.terms,
1261                "edges": s.edges,
1262                "edge_versions": s.edge_versions,
1263                "vectors": s.vectors,
1264                "hnsw_indexed": s.hnsw_indexed,
1265                "next_fact": s.next_fact,
1266                "next_entity": s.next_entity,
1267                "next_edge": s.next_edge,
1268                "pool_bytes": s.pool_bytes,
1269                "shards": {
1270                    "facts": s.shards.facts,
1271                    "entities": s.shards.entities,
1272                    "edges": s.shards.edges,
1273                    "temporal": s.shards.temporal,
1274                    "postings": s.shards.postings,
1275                },
1276            })
1277        )
1278        .ok();
1279    } else {
1280        writeln!(out, "facts       {}", s.facts).ok();
1281        writeln!(out, "entities    {}", s.entities).ok();
1282        writeln!(out, "terms       {}", s.terms).ok();
1283        writeln!(out, "edges       {}", s.edges).ok();
1284        writeln!(out, "edge_vers   {}", s.edge_versions).ok();
1285        writeln!(out, "vectors     {}", s.vectors).ok();
1286        writeln!(out, "hnsw_idx    {}", s.hnsw_indexed).ok();
1287        writeln!(out, "next_fact   {}", s.next_fact).ok();
1288        writeln!(out, "next_edge   {}", s.next_edge).ok();
1289        writeln!(out, "pool_bytes  {}", s.pool_bytes).ok();
1290        // The engine picks these from what it holds and moves them during
1291        // `maintain`; they are state to read, not a setting to choose.
1292        writeln!(
1293            out,
1294            "shards      facts {} entities {} edges {} temporal {} postings {}",
1295            s.shards.facts, s.shards.entities, s.shards.edges, s.shards.temporal, s.shards.postings,
1296        )
1297        .ok();
1298    }
1299}
1300
1301/// Renders one bounded page. Human output is line-oriented and keeps the
1302/// opaque continuation token separate so scripts can feed it back verbatim.
1303fn render_tags(page: &TagPage, json_output: bool, out: &mut impl Write) {
1304    if json_output {
1305        writeln!(
1306            out,
1307            "{}",
1308            json!({
1309                "items": page.items.iter().map(|item| json!({
1310                    "name": item.name,
1311                    "count": item.count,
1312                })).collect::<Vec<_>>(),
1313                "next_cursor": page.next_cursor,
1314            })
1315        )
1316        .ok();
1317        return;
1318    }
1319    for item in &page.items {
1320        writeln!(out, "{}\t{}", item.count, item.name).ok();
1321    }
1322    if let Some(cursor) = &page.next_cursor {
1323        writeln!(out, "next_cursor\t{cursor}").ok();
1324    }
1325}
1326
1327/// Writes one exported fact as a JSONL line. The unit of the streaming export
1328/// — the same shape with or without `--json` (JSONL is already machine-readable).
1329///
1330/// `kind` is what makes the format extensible: a reader dispatches on it, and a
1331/// line without one is a fact, which is exactly how files written before edges
1332/// existed still load.
1333fn write_export_line(out: &mut impl Write, f: &ExportedFact) {
1334    writeln!(
1335        out,
1336        "{}",
1337        json!({
1338            "kind": "fact",
1339            "id": f.id,
1340            "text": f.text,
1341            "entity": f.entity,
1342            "tags": f.tags,
1343            "metadata": f.metadata,
1344            "recorded_at": f.recorded_at,
1345            "valid_from": f.valid_from,
1346        })
1347    )
1348    .ok();
1349}
1350
1351/// Writes one edge line. Emitted **after** every fact line, so an importer
1352/// reading forward has already seen the fact a `provenance` names and can
1353/// translate its id without buffering or a second pass.
1354fn write_export_edge(out: &mut impl Write, src: &str, rel: &str, dst: &str, provenance: FactId) {
1355    writeln!(
1356        out,
1357        "{}",
1358        json!({
1359            "kind": "edge",
1360            "src": src,
1361            "rel": rel,
1362            "dst": dst,
1363            "provenance": (provenance != FactId::NONE).then_some(provenance.0),
1364        })
1365    )
1366    .ok();
1367}
1368
1369/// Renders a whole slice of exported facts as JSONL (test helper — the runtime
1370/// path streams via [`write_export_line`]).
1371#[cfg(test)]
1372fn render_export(facts: &[ExportedFact], _json: bool, out: &mut impl Write) {
1373    for f in facts {
1374        write_export_line(out, f);
1375    }
1376}
1377
1378/// Loads facts from a JSONL file (as written by `export`) in **streamed
1379/// batches** of `batch_size`: the file is read line-by-line (memory bounded to
1380/// a batch, not the whole file), and each full batch is one
1381/// [`remember_many`](Database::remember_many) — one embedder round-trip and one
1382/// journal fsync, instead of per fact. Returns the count imported. A malformed
1383/// line is a usage error naming its 1-based number.
1384fn do_import(
1385    db: &Database,
1386    now: u64,
1387    file: &std::path::Path,
1388    batch_size: usize,
1389    _out: &mut impl Write,
1390) -> Result<ImportReport, CliError> {
1391    let f = std::fs::File::open(file)
1392        .map_err(|e| CliError::Usage(format!("reading {}: {e}", file.display())))?;
1393    let reader = io::BufReader::new(f);
1394    let mut count = 0usize;
1395    let mut batch: Vec<ParsedFact> = Vec::with_capacity(batch_size);
1396    // Old fact id -> the id this database gave it. Sparse (an exporting
1397    // database has burned ids), so a map rather than a dense vector: a file
1398    // whose edges carry no provenance never grows it at all.
1399    let mut remap: BTreeMap<u32, FactId> = BTreeMap::new();
1400    let mut edges = 0usize;
1401
1402    for (i, line) in reader.lines().enumerate() {
1403        let line = line.map_err(|e| CliError::Usage(format!("line {}: {e}", i + 1)))?;
1404        let line = line.trim();
1405        if line.is_empty() {
1406            continue;
1407        }
1408        match parse_import_line(line, i + 1)? {
1409            ImportLine::Fact(fact) => {
1410                batch.push(fact);
1411                if batch.len() >= batch_size {
1412                    count += flush_import_batch(db, now, &batch, &mut remap)?;
1413                    batch.clear();
1414                }
1415            }
1416            ImportLine::Edge(edge) => {
1417                // Edges follow every fact in a file this CLI wrote, so the
1418                // pending batch has to land before an edge can name a fact
1419                // from it. Flushing here costs one extra batch write per file,
1420                // not per edge.
1421                if !batch.is_empty() {
1422                    count += flush_import_batch(db, now, &batch, &mut remap)?;
1423                    batch.clear();
1424                }
1425                db.link(LinkInput {
1426                    now,
1427                    src: &edge.src,
1428                    rel: &edge.rel,
1429                    dst: &edge.dst,
1430                    // A provenance naming a fact this file did not carry (it
1431                    // was closed, or forgotten before the export) links without
1432                    // one rather than pointing at an unrelated id.
1433                    provenance: edge.provenance.and_then(|old| remap.get(&old).copied()),
1434                })?;
1435                edges += 1;
1436            }
1437        }
1438    }
1439    count += flush_import_batch(db, now, &batch, &mut remap)?;
1440    Ok(ImportReport {
1441        facts: count,
1442        edges,
1443    })
1444}
1445
1446/// What an import wrote. Edges are counted separately because a file may carry
1447/// only facts (anything written before edges were in the format) and reporting
1448/// "0 edges" for those would read as a loss rather than as their absence.
1449#[derive(Debug, PartialEq, Eq)]
1450struct ImportReport {
1451    facts: usize,
1452    edges: usize,
1453}
1454
1455/// One line of an import file: the two shapes `export` writes.
1456enum ImportLine {
1457    Fact(ParsedFact),
1458    Edge(ParsedEdge),
1459}
1460
1461/// One parsed edge line. Its `provenance` is the fact id **in the exporting
1462/// database**; the importer translates it through the ids it just assigned.
1463struct ParsedEdge {
1464    src: String,
1465    rel: String,
1466    dst: String,
1467    provenance: Option<u32>,
1468}
1469
1470/// One parsed JSONL fact, owned so a whole batch can be buffered before its
1471/// `remember_many`. `id` is the exporting database's id, kept only so edges in
1472/// the same file can be pointed at the fact once it has a new one.
1473struct ParsedFact {
1474    id: Option<u32>,
1475    text: String,
1476    entity: Option<String>,
1477    tags: Vec<String>,
1478    metadata: Vec<(String, String)>,
1479    valid_from: Option<u64>,
1480}
1481
1482/// Parses one JSONL line. Dispatches on `kind`; a line without one is a fact,
1483/// which is how files written before edges existed still load. Bad JSON, an
1484/// unknown `kind`, or a fact missing its `text` is a usage error naming the
1485/// 1-based line.
1486fn parse_import_line(line: &str, lineno: usize) -> Result<ImportLine, CliError> {
1487    let v: serde_json::Value =
1488        serde_json::from_str(line).map_err(|e| CliError::Usage(format!("line {lineno}: {e}")))?;
1489    match v["kind"].as_str() {
1490        None | Some("fact") => parse_import_fact(&v, lineno).map(ImportLine::Fact),
1491        Some("edge") => parse_import_edge(&v, lineno).map(ImportLine::Edge),
1492        Some(other) => Err(CliError::Usage(format!(
1493            "line {lineno}: unknown kind \"{other}\" (expected \"fact\" or \"edge\")"
1494        ))),
1495    }
1496}
1497
1498/// Parses an edge line. All three endpoints are required — an edge missing one
1499/// is not a partial edge, it is a broken file.
1500fn parse_import_edge(v: &serde_json::Value, lineno: usize) -> Result<ParsedEdge, CliError> {
1501    let field = |key: &str| -> Result<String, CliError> {
1502        v[key]
1503            .as_str()
1504            .map(String::from)
1505            .ok_or_else(|| CliError::Usage(format!("line {lineno}: edge missing string \"{key}\"")))
1506    };
1507    Ok(ParsedEdge {
1508        src: field("src")?,
1509        rel: field("rel")?,
1510        dst: field("dst")?,
1511        provenance: v["provenance"].as_u64().and_then(|n| u32::try_from(n).ok()),
1512    })
1513}
1514
1515/// Parses a fact line into an owned fact.
1516fn parse_import_fact(v: &serde_json::Value, lineno: usize) -> Result<ParsedFact, CliError> {
1517    let text = v["text"]
1518        .as_str()
1519        .ok_or_else(|| CliError::Usage(format!("line {lineno}: missing string \"text\"")))?
1520        .to_string();
1521    let entity = v["entity"].as_str().map(String::from);
1522    let tags = v["tags"]
1523        .as_array()
1524        .map(|a| {
1525            a.iter()
1526                .filter_map(|t| t.as_str().map(String::from))
1527                .collect()
1528        })
1529        .unwrap_or_default();
1530    // Metadata: an object of string values. Keys are sorted (via `BTreeMap`) so
1531    // the imported pairs are canonical; non-string values are skipped.
1532    let metadata = v["metadata"]
1533        .as_object()
1534        .map(|m| {
1535            m.iter()
1536                .filter_map(|(k, val)| val.as_str().map(|s| (k.clone(), s.to_string())))
1537                .collect::<BTreeMap<_, _>>()
1538                .into_iter()
1539                .collect()
1540        })
1541        .unwrap_or_default();
1542    let valid_from = v["valid_from"].as_u64();
1543    Ok(ParsedFact {
1544        id: v["id"].as_u64().and_then(|n| u32::try_from(n).ok()),
1545        text,
1546        entity,
1547        tags,
1548        metadata,
1549        valid_from,
1550    })
1551}
1552
1553/// Writes one batch of parsed facts via `remember_many` (one embed round-trip,
1554/// one fsync). Returns how many were written; an empty batch is a no-op.
1555fn flush_import_batch(
1556    db: &Database,
1557    now: u64,
1558    batch: &[ParsedFact],
1559    remap: &mut BTreeMap<u32, FactId>,
1560) -> Result<usize, CliError> {
1561    if batch.is_empty() {
1562        return Ok(0);
1563    }
1564    // Per-fact `&[&str]` tag slices and `&[(&str,&str)]` metadata pairs must
1565    // outlive the `remember_many` call.
1566    let tag_refs: Vec<Vec<&str>> = batch
1567        .iter()
1568        .map(|p| p.tags.iter().map(String::as_str).collect())
1569        .collect();
1570    let meta_refs: Vec<Vec<(&str, &str)>> = batch
1571        .iter()
1572        .map(|p| {
1573            p.metadata
1574                .iter()
1575                .map(|(k, v)| (k.as_str(), v.as_str()))
1576                .collect()
1577        })
1578        .collect();
1579    let inputs: Vec<RememberInput> = batch
1580        .iter()
1581        .zip(&tag_refs)
1582        .zip(&meta_refs)
1583        .map(|((p, tags), meta)| RememberInput {
1584            entity: p.entity.as_deref(),
1585            tags,
1586            metadata: (!meta.is_empty()).then_some(meta.as_slice()),
1587            valid_from: p.valid_from,
1588            ..RememberInput::text(now, &p.text)
1589        })
1590        .collect();
1591    // `remember_many` returns outcomes in input order, so each parsed fact
1592    // learns the id this database gave it — the only thing an edge needs.
1593    let outcomes = db.remember_many(inputs)?;
1594    for (parsed, outcome) in batch.iter().zip(&outcomes) {
1595        if let Some(old) = parsed.id {
1596            remap.insert(old, outcome.id);
1597        }
1598    }
1599    Ok(batch.len())
1600}
1601
1602/// Shared `remember`/`revise` body: build the input and dispatch.
1603#[allow(clippy::too_many_arguments)]
1604fn do_remember(
1605    db: &Database,
1606    now: u64,
1607    text: &str,
1608    entity: &Option<String>,
1609    tags: &[String],
1610    links: &[String],
1611    meta: &[String],
1612    valid_from: Option<u64>,
1613    vector: &[f32],
1614    revise: Option<FactId>,
1615) -> Result<RememberOutcome, CliError> {
1616    let tag_refs: Vec<&str> = tags.iter().map(String::as_str).collect();
1617    let link_pairs = parse_links(links)?;
1618    let link_refs: Vec<(&str, &str)> = link_pairs
1619        .iter()
1620        .map(|(r, e)| (r.as_str(), e.as_str()))
1621        .collect();
1622    // A `BTreeMap` dedups keys (last `--meta` for a key wins) and sorts them;
1623    // the engine re-canonicalizes regardless, but this keeps the borrowed pairs
1624    // clean and dup-free.
1625    let meta_map = parse_meta(meta)?;
1626    let meta_refs: Vec<(&str, &str)> = meta_map
1627        .iter()
1628        .map(|(k, v)| (k.as_str(), v.as_str()))
1629        .collect();
1630    let input = RememberInput {
1631        entity: entity.as_deref(),
1632        tags: &tag_refs,
1633        links: &link_refs,
1634        metadata: (!meta_refs.is_empty()).then_some(meta_refs.as_slice()),
1635        valid_from,
1636        // An explicit `--vector` is authoritative: the host embeds only when
1637        // this is `None`, so passing one skips the provider entirely. The
1638        // engine checks the length against `dim`.
1639        vector: (!vector.is_empty()).then_some(vector),
1640        ..RememberInput::text(now, text)
1641    };
1642    match revise {
1643        Some(target) => Ok(db.revise(target, input)?),
1644        None => Ok(db.remember(input)?),
1645    }
1646}
1647
1648/// Guarded counterpart of [`do_remember`]. Kept separate at the final
1649/// dispatch so `revise` cannot accidentally acquire conditional semantics.
1650#[allow(clippy::too_many_arguments)]
1651fn do_guarded_remember(
1652    db: &Database,
1653    now: u64,
1654    text: &str,
1655    entity: &Option<String>,
1656    tags: &[String],
1657    links: &[String],
1658    meta: &[String],
1659    valid_from: Option<u64>,
1660    vector: &[f32],
1661) -> Result<GuardedRememberOutcome, CliError> {
1662    let tag_refs: Vec<&str> = tags.iter().map(String::as_str).collect();
1663    let link_pairs = parse_links(links)?;
1664    let link_refs: Vec<(&str, &str)> = link_pairs
1665        .iter()
1666        .map(|(relation, target)| (relation.as_str(), target.as_str()))
1667        .collect();
1668    let meta_map = parse_meta(meta)?;
1669    let meta_refs: Vec<(&str, &str)> = meta_map
1670        .iter()
1671        .map(|(key, value)| (key.as_str(), value.as_str()))
1672        .collect();
1673    Ok(db.remember_guarded(RememberInput {
1674        entity: entity.as_deref(),
1675        tags: &tag_refs,
1676        links: &link_refs,
1677        metadata: (!meta_refs.is_empty()).then_some(meta_refs.as_slice()),
1678        valid_from,
1679        vector: (!vector.is_empty()).then_some(vector),
1680        ..RememberInput::text(now, text)
1681    })?)
1682}
1683
1684/// Parses `--meta KEY=VALUE` strings into a sorted, deduped map (last value per
1685/// key wins).
1686fn parse_meta(meta: &[String]) -> Result<BTreeMap<String, String>, CliError> {
1687    let mut map = BTreeMap::new();
1688    for s in meta {
1689        let (k, v) = s
1690            .split_once('=')
1691            .filter(|(k, _)| !k.is_empty())
1692            .ok_or_else(|| CliError::Usage(format!("bad --meta `{s}` — expected KEY=VALUE")))?;
1693        map.insert(k.to_string(), v.to_string());
1694    }
1695    Ok(map)
1696}
1697
1698/// Parses `--link REL:ENTITY` strings into `(rel, entity)` pairs.
1699fn parse_links(links: &[String]) -> Result<Vec<(String, String)>, CliError> {
1700    links
1701        .iter()
1702        .map(|s| {
1703            s.split_once(':')
1704                .filter(|(r, e)| !r.is_empty() && !e.is_empty())
1705                .map(|(r, e)| (r.to_string(), e.to_string()))
1706                .ok_or_else(|| CliError::Usage(format!("bad --link `{s}` — expected REL:ENTITY")))
1707        })
1708        .collect()
1709}
1710
1711/// Renders a `remember`/`revise` outcome (shared shape).
1712fn render_remember(outcome: &RememberOutcome, json: bool, out: &mut impl Write) {
1713    if json {
1714        let similar: Vec<_> = outcome
1715            .similar
1716            .iter()
1717            .map(|s| json!({ "id": s.id.0, "score": s.score, "reason": format!("{:?}", s.reason) }))
1718            .collect();
1719        writeln!(
1720            out,
1721            "{}",
1722            json!({
1723                "id": outcome.id.0,
1724                "entity": outcome.entity.map(|e| e.0),
1725                "similar": similar,
1726            })
1727        )
1728        .ok();
1729    } else {
1730        writeln!(out, "remembered fact {}", outcome.id.0).ok();
1731        for s in &outcome.similar {
1732            writeln!(
1733                out,
1734                "  ~ similar to fact {} ({:?}, {:.2})",
1735                s.id.0, s.reason, s.score
1736            )
1737            .ok();
1738        }
1739    }
1740}
1741
1742fn render_guarded_remember(outcome: &GuardedRememberOutcome, json: bool, out: &mut impl Write) {
1743    match outcome {
1744        GuardedRememberOutcome::Stored { outcome, checked } => {
1745            if json {
1746                writeln!(
1747                    out,
1748                    "{}",
1749                    json!({
1750                        "status": "stored",
1751                        "checked": checked,
1752                        "outcome": {
1753                            "id": outcome.id.0,
1754                            "entity": outcome.entity.map(|entity| entity.0),
1755                            "similar": [],
1756                        }
1757                    })
1758                )
1759                .ok();
1760            } else {
1761                writeln!(out, "remembered fact {}", outcome.id.0).ok();
1762                // Said plainly, because the alternative is a caller who asked
1763                // for a duplicate check, did not get one, and cannot tell.
1764                if !checked {
1765                    writeln!(
1766                        out,
1767                        "  ! stored WITHOUT a duplicate check: the check is scoped to \
1768                         --entity, and none was given"
1769                    )
1770                    .ok();
1771                }
1772            }
1773        }
1774        GuardedRememberOutcome::Blocked { similar } => {
1775            if json {
1776                let similar: Vec<_> = similar
1777                    .iter()
1778                    .map(|item| {
1779                        json!({
1780                            "id": item.id.0,
1781                            "score": item.score,
1782                            "reason": format!("{:?}", item.reason),
1783                        })
1784                    })
1785                    .collect();
1786                writeln!(
1787                    out,
1788                    "{}",
1789                    json!({ "status": "blocked", "similar": similar })
1790                )
1791                .ok();
1792            } else {
1793                writeln!(out, "not remembered: similar facts require a decision").ok();
1794                for item in similar {
1795                    writeln!(
1796                        out,
1797                        "  ~ fact {} ({:?}, {:.2})",
1798                        item.id.0, item.reason, item.score
1799                    )
1800                    .ok();
1801                }
1802            }
1803        }
1804    }
1805}
1806
1807/// `VALID_TO_OPEN` → JSON `null`, a real bound → the number.
1808fn open_or(valid_to: u64) -> Option<u64> {
1809    (valid_to != VALID_TO_OPEN).then_some(valid_to)
1810}
1811
1812#[cfg(test)]
1813mod tests {
1814    use plugmem_host::Config;
1815
1816    use super::*;
1817
1818    /// A stub embedder returning a fixed vector per input — no network.
1819    struct StubEmbedder;
1820    impl plugmem_host::Embedder for StubEmbedder {
1821        fn space_id(&self) -> &str {
1822            "cli-stub"
1823        }
1824
1825        fn dim(&self) -> usize {
1826            3
1827        }
1828        fn embed(&self, texts: &[&str]) -> Result<Vec<Vec<f32>>, HostError> {
1829            Ok(texts.iter().map(|_| vec![0.1, 0.2, 0.3]).collect())
1830        }
1831    }
1832
1833    fn recall_cmd(query: Option<&str>) -> Command {
1834        Command::Recall {
1835            query: query.map(str::to_owned),
1836            tags: vec![],
1837            entities: vec![],
1838            as_of: None,
1839            range: None,
1840            k: 0,
1841            closed: false,
1842            token_budget: None,
1843            ef: None,
1844            graph_depth: None,
1845            vector: Vec::new(),
1846        }
1847    }
1848
1849    fn settings_with(embedder: Option<Box<dyn plugmem_host::Embedder>>) -> Settings {
1850        Settings {
1851            database_path: None,
1852            config: Config::default(),
1853            embedder,
1854            snapshot_every_ops: None,
1855            snapshot_journal_bytes: None,
1856            maintain_every_forgets: None,
1857            fsync: None,
1858            workspace: plugmem_host::WorkspaceSettings {
1859                dir: None,
1860                limits: plugmem_host::WorkspaceLimits::default(),
1861            },
1862            warnings: Vec::new(),
1863        }
1864    }
1865
1866    #[test]
1867    fn embed_recall_query_embeds_recall_text_only_when_an_embedder_is_set() {
1868        // recall text + embedder → a vector.
1869        let mut with = settings_with(Some(Box::new(StubEmbedder)));
1870        assert_eq!(
1871            embed_recall_query(&mut with, &recall_cmd(Some("tokio"))).unwrap(),
1872            Some(vec![0.1, 0.2, 0.3])
1873        );
1874
1875        // no embedder → None (recall falls back to lexical/structural sources).
1876        let mut without = settings_with(None);
1877        assert_eq!(
1878            embed_recall_query(&mut without, &recall_cmd(Some("tokio"))).unwrap(),
1879            None
1880        );
1881
1882        // recall with no query text → None (nothing to embed).
1883        let mut with_empty = settings_with(Some(Box::new(StubEmbedder)));
1884        assert_eq!(
1885            embed_recall_query(&mut with_empty, &recall_cmd(None)).unwrap(),
1886            None
1887        );
1888
1889        // a non-recall command → None even with an embedder configured.
1890        let mut with_stats = settings_with(Some(Box::new(StubEmbedder)));
1891        assert_eq!(
1892            embed_recall_query(&mut with_stats, &Command::Stats).unwrap(),
1893            None
1894        );
1895    }
1896
1897    #[test]
1898    fn split_line_honors_quotes_and_whitespace() {
1899        assert_eq!(split_line("remember hello"), ["remember", "hello"]);
1900        assert_eq!(
1901            split_line(r#"remember "two words" --tag x"#),
1902            ["remember", "two words", "--tag", "x"]
1903        );
1904        assert_eq!(split_line("  recall   'a b'  "), ["recall", "a b"]);
1905        assert_eq!(split_line(""), Vec::<String>::new());
1906        // An empty quoted string is a real (empty) argument.
1907        assert_eq!(split_line(r#"remember """#), ["remember", ""]);
1908    }
1909
1910    #[test]
1911    fn repl_runs_over_one_handle_and_checkpoints_on_exit() {
1912        let (db, tmp) = TempDb::open();
1913        let path = tmp.0.join("m.plugmem");
1914        drop(db); // release the writer lock so run_repl can open it
1915
1916        let settings = settings_with(None);
1917        // Multi-word text is quoted, same grammar as the one-shot CLI.
1918        let script = b"remember \"hello tokio world\"\nrecall tokio\nrevise 0 \"goodbye tokio\"\nbadcmd\nexit\n";
1919        let mut out = Vec::new();
1920        let code = run_repl(&path, settings, false, &script[..], &mut out);
1921        let text = String::from_utf8(out).unwrap();
1922
1923        assert_eq!(code, 0);
1924        assert!(text.contains("remembered fact 0"), "{text}");
1925        assert!(text.contains("tokio"), "{text}");
1926        // A bad line is reported but does not end the session (revise ran after).
1927        assert!(text.contains("unrecognized subcommand"), "{text}");
1928
1929        // Checkpointed on exit → a fresh read-only open sees the data with a
1930        // clean journal. The revise chain leaves two facts: the closed original
1931        // and its active successor.
1932        let ro = Database::open_readonly(&path, Config::default()).unwrap();
1933        assert_eq!(ro.stats().facts, 2, "original + successor after the revise");
1934    }
1935
1936    #[test]
1937    fn read_only_repl_observes_a_writer_reports_freshness_and_refuses_writes() {
1938        let (db, tmp) = TempDb::open();
1939        let path = tmp.0.join("m.plugmem");
1940        // Seed and publish generation 1, then keep the writer open and live —
1941        // the read-only repl observes it cross-process (Variant 2 MVCC).
1942        let mut sink = Vec::new();
1943        execute(
1944            &db,
1945            &remember("seed fact tokio", None, &[]),
1946            false,
1947            1_000,
1948            &mut sink,
1949        )
1950        .unwrap();
1951        db.checkpoint(1_001).unwrap();
1952
1953        let settings = settings_with(None);
1954        // A read verb, both freshness verbs, and a write (must be refused).
1955        let script = b"generation\nstats\nrefresh\nremember \"nope\"\nexit\n";
1956        let mut out = Vec::new();
1957        let code = run_repl_ro(&path, settings, false, &script[..], &mut out);
1958        let text = String::from_utf8(out).unwrap();
1959
1960        assert_eq!(code, 0);
1961        assert!(text.contains("generation 1"), "generation verb: {text}");
1962        assert!(text.contains("fact"), "stats ran: {text}");
1963        // The writer published nothing after the reader opened, so refresh is a
1964        // no-op that stays on generation 1.
1965        assert!(
1966            text.contains("already current → generation 1"),
1967            "refresh no-op: {text}"
1968        );
1969        // A write verb is refused without ending the session (exit still ran).
1970        assert!(text.contains("read-only session"), "write refused: {text}");
1971
1972        // The read-only session never wrote: the writer is still on generation 1
1973        // with its single seeded fact, untouched by the repl.
1974        assert_eq!(db.stats().facts, 1);
1975    }
1976
1977    #[test]
1978    fn read_only_repl_refresh_advances_after_the_writer_checkpoints() {
1979        let (db, tmp) = TempDb::open();
1980        let path = tmp.0.join("m.plugmem");
1981        let mut sink = Vec::new();
1982        execute(&db, &remember("first", None, &[]), false, 1_000, &mut sink).unwrap();
1983        db.checkpoint(1_001).unwrap();
1984
1985        // A reader hook that publishes a *new* generation the first time the repl
1986        // pulls a line, so the subsequent `refresh` deterministically advances —
1987        // exercising the "refreshed" branch without a background thread.
1988        struct HookOnFirstRead<'a> {
1989            script: std::io::Cursor<&'a [u8]>,
1990            db: &'a Database,
1991            fired: bool,
1992        }
1993        impl std::io::Read for HookOnFirstRead<'_> {
1994            fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
1995                if !self.fired {
1996                    self.fired = true;
1997                    // Publish generation 2 before the first command is read, so
1998                    // the reader (opened on gen 1) sees something newer.
1999                    let mut s = Vec::new();
2000                    execute(
2001                        self.db,
2002                        &remember("second", None, &[]),
2003                        false,
2004                        2_000,
2005                        &mut s,
2006                    )
2007                    .unwrap();
2008                    self.db.checkpoint(2_001).unwrap();
2009                }
2010                self.script.read(buf)
2011            }
2012        }
2013        let reader = std::io::BufReader::new(HookOnFirstRead {
2014            script: std::io::Cursor::new(b"refresh\nstats\nexit\n" as &[u8]),
2015            db: &db,
2016            fired: false,
2017        });
2018
2019        let mut out = Vec::new();
2020        let code = run_repl_ro(&path, settings_with(None), false, reader, &mut out);
2021        let text = String::from_utf8(out).unwrap();
2022
2023        assert_eq!(code, 0);
2024        // Opened on gen 1, the writer published gen 2, refresh advanced onto it.
2025        assert!(text.contains("refreshed → generation 2"), "advance: {text}");
2026        // And the advanced reader now sees the writer's second fact.
2027        assert!(text.contains("fact"), "stats after refresh: {text}");
2028        assert_eq!(db.stats().facts, 2);
2029    }
2030
2031    #[test]
2032    fn read_only_repl_freshness_verbs_emit_json() {
2033        let (db, tmp) = TempDb::open();
2034        let path = tmp.0.join("m.plugmem");
2035        let mut sink = Vec::new();
2036        execute(&db, &remember("j", None, &[]), false, 1_000, &mut sink).unwrap();
2037        db.checkpoint(1_001).unwrap();
2038
2039        let script = b"generation\nrefresh\nexit\n";
2040        let mut out = Vec::new();
2041        let code = run_repl_ro(&path, settings_with(None), true, &script[..], &mut out);
2042        let text = String::from_utf8(out).unwrap();
2043
2044        assert_eq!(code, 0);
2045        assert!(
2046            text.contains(r#""generation":1"#),
2047            "generation json: {text}"
2048        );
2049        assert!(text.contains(r#""advanced":false"#), "refresh json: {text}");
2050    }
2051
2052    /// A throwaway database on a unique temp path; removed on drop.
2053    struct TempDb(PathBuf);
2054    impl TempDb {
2055        fn open() -> (Database, Self) {
2056            let dir = std::env::temp_dir().join(format!(
2057                "plugmem-cli-{}-{}",
2058                std::process::id(),
2059                now_ms_unique()
2060            ));
2061            std::fs::create_dir_all(&dir).unwrap();
2062            let path = dir.join("m.plugmem");
2063            let (db, _) = Database::open(&path, Config::default()).unwrap();
2064            (db, TempDb(dir))
2065        }
2066    }
2067    impl Drop for TempDb {
2068        fn drop(&mut self) {
2069            let _ = std::fs::remove_dir_all(&self.0);
2070        }
2071    }
2072
2073    /// A strictly-increasing counter so temp dirs never collide within a run
2074    /// (the wall clock alone can repeat at millisecond resolution).
2075    fn now_ms_unique() -> String {
2076        use std::sync::atomic::{AtomicU64, Ordering};
2077        static N: AtomicU64 = AtomicU64::new(0);
2078        format!("{}-{}", now_ms(), N.fetch_add(1, Ordering::Relaxed))
2079    }
2080
2081    fn run_cmd(db: &Database, cmd: &Command, json: bool, now: u64) -> (u8, String) {
2082        let mut buf = Vec::new();
2083        let code = execute(db, cmd, json, now, &mut buf).expect("execute");
2084        (code, String::from_utf8(buf).unwrap())
2085    }
2086
2087    fn remember(text: &str, entity: Option<&str>, tags: &[&str]) -> Command {
2088        Command::Remember {
2089            text: text.into(),
2090            entity: entity.map(Into::into),
2091            tags: tags.iter().map(|t| (*t).into()).collect(),
2092            links: Vec::new(),
2093            meta: Vec::new(),
2094            valid_from: None,
2095            vector: Vec::new(),
2096            guarded: false,
2097        }
2098    }
2099
2100    fn remember_with_meta(text: &str, meta: &[&str]) -> Command {
2101        Command::Remember {
2102            text: text.into(),
2103            entity: None,
2104            tags: Vec::new(),
2105            links: Vec::new(),
2106            meta: meta.iter().map(|m| (*m).into()).collect(),
2107            valid_from: None,
2108            vector: Vec::new(),
2109            guarded: false,
2110        }
2111    }
2112
2113    #[test]
2114    fn guarded_remember_flag_reports_blocked_without_a_write() {
2115        let (db, _temp) = TempDb::open();
2116        let mut first = remember("likes green tea every morning", Some("user"), &[]);
2117        let Command::Remember { guarded, .. } = &mut first else {
2118            unreachable!()
2119        };
2120        *guarded = true;
2121        let (_, stored) = run_cmd(&db, &first, true, 1_000);
2122        let stored: serde_json::Value = serde_json::from_str(&stored).unwrap();
2123        assert_eq!(stored["status"], "stored");
2124        assert_eq!(stored["outcome"]["id"], 0);
2125        assert_eq!(stored["checked"], true, "an entity was named");
2126
2127        let mut second = remember("likes green tea each morning", Some("user"), &[]);
2128        let Command::Remember { guarded, .. } = &mut second else {
2129            unreachable!()
2130        };
2131        *guarded = true;
2132        let (_, blocked) = run_cmd(&db, &second, true, 2_000);
2133        let blocked: serde_json::Value = serde_json::from_str(&blocked).unwrap();
2134        assert_eq!(blocked["status"], "blocked");
2135        assert_eq!(blocked["similar"][0]["id"], 0);
2136        assert_eq!(db.stats().facts, 1);
2137    }
2138
2139    /// `--guarded` with no `--entity` stores unguarded, and prints that it did.
2140    ///
2141    /// The detector is scoped to the fact's entity, so there is no candidate
2142    /// set to compare against and nothing can be blocked - now or after any
2143    /// number of later writes. Without the notice the caller sees a plain
2144    /// "remembered fact N" and has no way to learn the check they asked for
2145    /// never ran.
2146    #[test]
2147    fn guarded_without_an_entity_says_it_stored_unchecked() {
2148        let (db, _temp) = TempDb::open();
2149        let text = "the cache is disabled because it raced with the warmup task";
2150        for at in [1_000, 2_000, 3_000] {
2151            let mut cmd = remember(text, None, &[]);
2152            let Command::Remember { guarded, .. } = &mut cmd else {
2153                unreachable!()
2154            };
2155            *guarded = true;
2156            let (_, plain) = run_cmd(&db, &cmd, false, at);
2157            assert!(plain.contains("remembered fact"), "{plain}");
2158            assert!(plain.contains("WITHOUT a duplicate check"), "{plain}");
2159        }
2160        assert_eq!(db.stats().facts, 3, "identical facts, as remember would");
2161
2162        let mut cmd = remember(text, None, &[]);
2163        let Command::Remember { guarded, .. } = &mut cmd else {
2164            unreachable!()
2165        };
2166        *guarded = true;
2167        let (_, json) = run_cmd(&db, &cmd, true, 4_000);
2168        let json: serde_json::Value = serde_json::from_str(&json).unwrap();
2169        assert_eq!(json["status"], "stored");
2170        assert_eq!(json["checked"], false);
2171    }
2172
2173    #[test]
2174    fn meta_flag_renders_sorted_in_show_and_export_and_rejects_bad_input() {
2175        let (db, _t) = TempDb::open();
2176        // Keys given out of order; last value for a repeated key wins.
2177        let cmd = remember_with_meta("a scan", &["uri=s3://b/x", "page=2", "page=3"]);
2178        assert_eq!(run_cmd(&db, &cmd, false, 1_000).0, 0);
2179
2180        // show (human): sorted `key=value`, last-write-wins on `page`.
2181        let (_, human) = run_cmd(&db, &Command::Show { id: 0 }, false, 2_000);
2182        assert!(
2183            human.contains("metadata    page=3, uri=s3://b/x"),
2184            "{human}"
2185        );
2186        // show (json): a metadata object.
2187        let (_, jshow) = run_cmd(&db, &Command::Show { id: 0 }, true, 2_000);
2188        let v: serde_json::Value = serde_json::from_str(&jshow).unwrap();
2189        assert_eq!(v["metadata"]["page"], "3");
2190        assert_eq!(v["metadata"]["uri"], "s3://b/x");
2191
2192        // export: the JSONL line carries the same object.
2193        let (_, exp) = run_cmd(&db, &Command::Export, false, 2_000);
2194        let line: serde_json::Value = serde_json::from_str(exp.lines().next().unwrap()).unwrap();
2195        assert_eq!(line["metadata"]["uri"], "s3://b/x");
2196
2197        // A `--meta` without `=` is a usage error.
2198        assert!(matches!(
2199            parse_meta(&["noequals".to_string()]),
2200            Err(CliError::Usage(_))
2201        ));
2202        assert!(parse_meta(&["=noKey".to_string()]).is_err());
2203    }
2204
2205    #[test]
2206    fn remember_then_recall_human_and_json() {
2207        let (db, _t) = TempDb::open();
2208        let (code, out) = run_cmd(
2209            &db,
2210            &remember("prefers tokio", Some("user"), &["pref"]),
2211            false,
2212            1_000,
2213        );
2214        assert_eq!(code, 0);
2215        assert!(out.starts_with("remembered fact 0"), "{out}");
2216
2217        // human recall
2218        let recall = Command::Recall {
2219            query: Some("tokio".into()),
2220            tags: Vec::new(),
2221            entities: Vec::new(),
2222            as_of: None,
2223            range: None,
2224            k: 0,
2225            closed: false,
2226            token_budget: None,
2227            ef: None,
2228            graph_depth: None,
2229            vector: Vec::new(),
2230        };
2231        let (code, out) = run_cmd(&db, &recall, false, 2_000);
2232        assert_eq!(code, 0);
2233        assert!(out.contains("tokio"), "{out}");
2234
2235        // json recall
2236        let (code, out) = run_cmd(&db, &recall, true, 2_000);
2237        assert_eq!(code, 0);
2238        let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2239        assert!(!v["facts"].as_array().unwrap().is_empty(), "{out}");
2240    }
2241
2242    #[test]
2243    fn recall_empty_is_ok_with_a_note() {
2244        let (db, _t) = TempDb::open();
2245        let recall = Command::Recall {
2246            query: Some("nothing here".into()),
2247            tags: Vec::new(),
2248            entities: Vec::new(),
2249            as_of: None,
2250            range: None,
2251            k: 0,
2252            closed: false,
2253            token_budget: None,
2254            ef: None,
2255            graph_depth: None,
2256            vector: Vec::new(),
2257        };
2258        let (code, out) = run_cmd(&db, &recall, false, 1_000);
2259        assert_eq!(code, 0);
2260        assert!(out.contains("nothing recalled"), "{out}");
2261    }
2262
2263    #[test]
2264    fn revise_closes_the_predecessor_and_conflict_is_surfaced() {
2265        let (db, _t) = TempDb::open();
2266        run_cmd(
2267            &db,
2268            &remember("lives in Moscow", Some("user"), &[]),
2269            false,
2270            1_000,
2271        );
2272        // a near-duplicate surfaces a similar hint
2273        let (_, out) = run_cmd(
2274            &db,
2275            &remember("lives in Moscow now", Some("user"), &[]),
2276            false,
2277            1_500,
2278        );
2279        assert!(out.contains("similar to fact"), "{out}");
2280
2281        let revise = Command::Revise {
2282            id: 0,
2283            text: "lives in Berlin".into(),
2284            entity: Some("user".into()),
2285            tags: Vec::new(),
2286            links: Vec::new(),
2287            meta: Vec::new(),
2288            valid_from: None,
2289            vector: Vec::new(),
2290        };
2291        let (code, out) = run_cmd(&db, &revise, false, 2_000);
2292        assert_eq!(code, 0);
2293        assert!(out.starts_with("remembered fact"), "{out}");
2294    }
2295
2296    #[test]
2297    fn show_found_and_missing() {
2298        let (db, _t) = TempDb::open();
2299        run_cmd(&db, &remember("a note", None, &[]), false, 1_000);
2300
2301        let (code, out) = run_cmd(&db, &Command::Show { id: 0 }, false, 2_000);
2302        assert_eq!(code, 0);
2303        assert!(
2304            out.contains("a note") && out.contains("recorded_at 1000"),
2305            "{out}"
2306        );
2307
2308        let (code, out) = run_cmd(&db, &Command::Show { id: 999 }, false, 2_000);
2309        assert_eq!(code, 1, "missing id is a soft miss");
2310        assert!(out.contains("not found"), "{out}");
2311
2312        // json card
2313        let (_, out) = run_cmd(&db, &Command::Show { id: 0 }, true, 2_000);
2314        let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2315        assert_eq!(v["text"], "a note");
2316        assert_eq!(v["valid_to"], serde_json::Value::Null); // open interval
2317    }
2318
2319    #[test]
2320    fn forget_then_maintain_purges() {
2321        let (db, _t) = TempDb::open();
2322        run_cmd(&db, &remember("temp", None, &[]), false, 1_000);
2323
2324        let (code, out) = run_cmd(&db, &Command::Forget { ids: vec![0] }, false, 2_000);
2325        assert_eq!(code, 0);
2326        assert!(out.contains("forgot fact 0"), "{out}");
2327        // second forget is idempotent
2328        let (_, out) = run_cmd(&db, &Command::Forget { ids: vec![0] }, false, 2_100);
2329        assert!(out.contains("already gone"), "{out}");
2330
2331        let (code, out) = run_cmd(
2332            &db,
2333            &Command::Maintain {
2334                mode: MaintainMode::Auto,
2335                reembed: false,
2336                batch_size: None,
2337            },
2338            false,
2339            3_000,
2340        );
2341        assert_eq!(code, 0);
2342        assert!(out.contains("purged 1"), "{out}");
2343    }
2344
2345    #[test]
2346    fn forget_many_ids_one_line_per_id() {
2347        let (db, _t) = TempDb::open();
2348        run_cmd(&db, &remember("a", None, &[]), false, 1_000);
2349        run_cmd(&db, &remember("b", None, &[]), false, 1_100);
2350        run_cmd(&db, &remember("c", None, &[]), false, 1_200);
2351
2352        let (code, out) = run_cmd(&db, &Command::Forget { ids: vec![0, 1] }, false, 2_000);
2353        assert_eq!(code, 0);
2354        let lines: Vec<_> = out.lines().collect();
2355        assert_eq!(lines, vec!["forgot fact 0", "forgot fact 1"]);
2356
2357        // A second batch mixes an already-gone id with a fresh one.
2358        let (code, out) = run_cmd(&db, &Command::Forget { ids: vec![0, 2] }, false, 2_100);
2359        assert_eq!(code, 0);
2360        let lines: Vec<_> = out.lines().collect();
2361        assert_eq!(lines, vec!["fact 0 was already gone", "forgot fact 2"]);
2362    }
2363
2364    #[test]
2365    fn forget_many_ids_json_is_an_array() {
2366        let (db, _t) = TempDb::open();
2367        run_cmd(&db, &remember("a", None, &[]), false, 1_000);
2368        run_cmd(&db, &remember("b", None, &[]), false, 1_100);
2369
2370        let (_, out) = run_cmd(&db, &Command::Forget { ids: vec![0, 1] }, true, 2_000);
2371        let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2372        let arr = v.as_array().expect("json array for multi-id forget");
2373        assert_eq!(arr.len(), 2);
2374        assert_eq!(arr[0]["id"], 0);
2375        assert_eq!(arr[0]["forgotten"], true);
2376        assert_eq!(arr[1]["id"], 1);
2377        assert_eq!(arr[1]["forgotten"], true);
2378    }
2379
2380    #[test]
2381    fn explicit_reembed_has_human_and_json_reports() {
2382        let (plain, tmp) = TempDb::open();
2383        let path = tmp.0.join("m.plugmem");
2384        drop(plain);
2385        let mut config = Config::default();
2386        config.dim = 3;
2387        let (db, _) = Database::builder(config)
2388            .embedder(Box::new(StubEmbedder))
2389            .open(path)
2390            .unwrap();
2391        run_cmd(&db, &remember("vector fact", None, &["kept"]), false, 1);
2392
2393        let command = Command::Maintain {
2394            mode: MaintainMode::Auto,
2395            reembed: true,
2396            batch_size: Some(1),
2397        };
2398        let (code, human) = run_cmd(&db, &command, false, 2);
2399        assert_eq!(code, 0);
2400        assert!(human.contains("reembedded: 1 facts"));
2401        let (code, json) = run_cmd(&db, &command, true, 3);
2402        assert_eq!(code, 0);
2403        let report: serde_json::Value = serde_json::from_str(json.trim()).unwrap();
2404        assert_eq!(report["new_space"], "cli-stub");
2405        assert_eq!(report["embedded"], 1);
2406    }
2407
2408    #[test]
2409    fn readonly_cli_refuses_a_same_dimension_different_space() {
2410        let (plain, tmp) = TempDb::open();
2411        let path = tmp.0.join("m.plugmem");
2412        drop(plain);
2413        let mut config = Config::default();
2414        config.dim = 3;
2415        let (db, _) = Database::builder(config.clone())
2416            .embedder(Box::new(StubEmbedder))
2417            .open(&path)
2418            .unwrap();
2419        db.remember(RememberInput::text(1, "stored vector"))
2420            .unwrap();
2421        db.checkpoint(2).unwrap();
2422        let ro = Database::open_readonly(&path, config).unwrap();
2423        let cmd = recall_cmd(Some("stored"));
2424        let mut out = Vec::new();
2425        let code = execute_ro(
2426            &ro,
2427            &cmd,
2428            Some(&[0.1, 0.2, 0.3]),
2429            Some("other-model"),
2430            false,
2431            &mut out,
2432        );
2433        assert_eq!(code, 2);
2434    }
2435
2436    #[test]
2437    fn tags_pages_and_remove_tag_have_human_and_json_shapes() {
2438        let (db, _t) = TempDb::open();
2439        run_cmd(&db, &remember("one", None, &["drop", "keep"]), false, 1_000);
2440        run_cmd(&db, &remember("two", None, &["drop"]), false, 1_100);
2441
2442        let (code, out) = run_cmd(
2443            &db,
2444            &Command::Tags {
2445                prefix: None,
2446                cursor: None,
2447                limit: 1,
2448            },
2449            false,
2450            2_000,
2451        );
2452        assert_eq!(code, 0);
2453        assert!(out.starts_with("2\tdrop\n"), "{out}");
2454        assert!(out.contains("next_cursor\t"), "{out}");
2455
2456        let (_, out) = run_cmd(&db, &Command::RemoveTag { tag: "drop".into() }, true, 3_000);
2457        let removed: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2458        assert_eq!(removed["affected"], 2);
2459
2460        let (_, out) = run_cmd(
2461            &db,
2462            &Command::Tags {
2463                prefix: None,
2464                cursor: None,
2465                limit: 0,
2466            },
2467            true,
2468            4_000,
2469        );
2470        let page: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2471        assert_eq!(page["items"][0]["name"], "keep");
2472        assert_eq!(db.export().len(), 2, "facts remain current after retagging");
2473    }
2474
2475    #[test]
2476    fn link_and_stats_and_json() {
2477        let (db, _t) = TempDb::open();
2478        run_cmd(
2479            &db,
2480            &remember("uses tokio", Some("plugmem"), &[]),
2481            false,
2482            1_000,
2483        );
2484        let link = Command::Link {
2485            src: "plugmem".into(),
2486            rel: "depends_on".into(),
2487            dst: "tokio".into(),
2488            provenance: None,
2489        };
2490        let (code, out) = run_cmd(&db, &link, false, 2_000);
2491        assert_eq!(code, 0);
2492        assert!(out.contains("plugmem -depends_on-> tokio"), "{out}");
2493        let unlink = Command::Unlink {
2494            src: "plugmem".into(),
2495            rel: "depends_on".into(),
2496            dst: "tokio".into(),
2497        };
2498        let (code, out) = run_cmd(&db, &unlink, false, 2_500);
2499        assert_eq!(code, 0);
2500        assert!(
2501            out.contains("unlinked plugmem -depends_on-> tokio"),
2502            "{out}"
2503        );
2504
2505        let (code, out) = run_cmd(&db, &Command::Stats, true, 3_000);
2506        assert_eq!(code, 0);
2507        let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2508        assert_eq!(v["facts"], 1);
2509        assert_eq!(v["edges"], 0);
2510        assert_eq!(v["edge_versions"], 1);
2511    }
2512
2513    #[test]
2514    fn bad_link_is_a_usage_error() {
2515        let (db, _t) = TempDb::open();
2516        let cmd = Command::Remember {
2517            text: "x".into(),
2518            entity: Some("user".into()),
2519            tags: Vec::new(),
2520            links: vec!["not-a-pair".into()],
2521            meta: Vec::new(),
2522            valid_from: None,
2523            vector: Vec::new(),
2524            guarded: false,
2525        };
2526        let mut buf = Vec::new();
2527        let err = execute(&db, &cmd, false, 1_000, &mut buf).unwrap_err();
2528        assert!(matches!(err, CliError::Usage(_)));
2529    }
2530
2531    #[test]
2532    fn as_of_time_travel_via_recall() {
2533        let (db, _t) = TempDb::open();
2534        run_cmd(
2535            &db,
2536            &remember("lives in Moscow", Some("user"), &[]),
2537            false,
2538            1_000,
2539        );
2540        let revise = Command::Revise {
2541            id: 0,
2542            text: "lives in Berlin".into(),
2543            entity: Some("user".into()),
2544            tags: Vec::new(),
2545            links: Vec::new(),
2546            meta: Vec::new(),
2547            valid_from: None,
2548            vector: Vec::new(),
2549        };
2550        run_cmd(&db, &revise, false, 2_000);
2551
2552        let as_of = Command::Recall {
2553            query: Some("lives".into()),
2554            tags: Vec::new(),
2555            entities: vec!["user".into()],
2556            as_of: Some(1_500),
2557            range: None,
2558            k: 0,
2559            closed: false,
2560            token_budget: None,
2561            ef: None,
2562            graph_depth: None,
2563            vector: Vec::new(),
2564        };
2565        let (_, out) = run_cmd(&db, &as_of, false, 3_000);
2566        assert!(out.contains("Moscow"), "as-of 1500 → Moscow: {out}");
2567    }
2568
2569    #[test]
2570    fn every_command_has_a_json_shape() {
2571        let (db, _t) = TempDb::open();
2572        // remember --json: id + similar array
2573        let (_, out) = run_cmd(
2574            &db,
2575            &remember("uses tokio", Some("plugmem"), &["pref"]),
2576            true,
2577            1_000,
2578        );
2579        let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2580        assert_eq!(v["id"], 0);
2581        assert!(v["similar"].is_array());
2582
2583        // revise --json
2584        let revise = Command::Revise {
2585            id: 0,
2586            text: "uses tokio now".into(),
2587            entity: Some("plugmem".into()),
2588            tags: Vec::new(),
2589            links: Vec::new(),
2590            meta: Vec::new(),
2591            valid_from: None,
2592            vector: Vec::new(),
2593        };
2594        let (_, out) = run_cmd(&db, &revise, true, 1_500);
2595        assert!(serde_json::from_str::<serde_json::Value>(out.trim()).is_ok());
2596
2597        // link --json
2598        let link = Command::Link {
2599            src: "plugmem".into(),
2600            rel: "depends_on".into(),
2601            dst: "tokio".into(),
2602            provenance: None,
2603        };
2604        let (_, out) = run_cmd(&db, &link, true, 2_000);
2605        let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2606        assert_eq!(v["rel"], "depends_on");
2607        let unlink = Command::Unlink {
2608            src: "plugmem".into(),
2609            rel: "depends_on".into(),
2610            dst: "tokio".into(),
2611        };
2612        let (_, out) = run_cmd(&db, &unlink, true, 2_100);
2613        let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2614        assert_eq!(v["unlinked"], true);
2615
2616        // forget --json then maintain --json
2617        let (_, out) = run_cmd(&db, &Command::Forget { ids: vec![1] }, true, 2_500);
2618        let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2619        assert_eq!(v["forgotten"], true);
2620        let (_, out) = run_cmd(
2621            &db,
2622            &Command::Maintain {
2623                mode: MaintainMode::Auto,
2624                reembed: false,
2625                batch_size: None,
2626            },
2627            true,
2628            3_000,
2629        );
2630        let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2631        assert!(v["purged"].as_u64().unwrap() >= 1);
2632
2633        // show --json of a missing id
2634        let (code, out) = run_cmd(&db, &Command::Show { id: 999 }, true, 3_500);
2635        assert_eq!(code, 1);
2636        let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2637        assert_eq!(v["found"], false);
2638
2639        // recall --json with a range window (covers the range/closed paths)
2640        let recall = Command::Recall {
2641            query: None,
2642            tags: Vec::new(),
2643            entities: vec!["plugmem".into()],
2644            as_of: None,
2645            range: Some(vec![0, 10_000]),
2646            k: 4,
2647            closed: true,
2648            token_budget: None,
2649            ef: None,
2650            graph_depth: None,
2651            vector: Vec::new(),
2652        };
2653        let (_, out) = run_cmd(&db, &recall, true, 4_000);
2654        assert!(serde_json::from_str::<serde_json::Value>(out.trim()).is_ok());
2655    }
2656
2657    #[test]
2658    fn graph_depth_is_a_per_call_flag_over_the_configured_default() {
2659        // A chain a -> b -> c -> d with one fact each, so the number of facts
2660        // recalled *is* the number of hops taken.
2661        let (db, _t) = TempDb::open();
2662        for (i, (entity, next)) in [("a", "b"), ("b", "c"), ("c", "d"), ("d", "e")]
2663            .iter()
2664            .enumerate()
2665        {
2666            let cmd = Command::Remember {
2667                text: format!("fact on {entity}"),
2668                entity: Some((*entity).into()),
2669                tags: Vec::new(),
2670                links: vec![format!("leads_to:{next}")],
2671                meta: Vec::new(),
2672                valid_from: None,
2673                vector: Vec::new(),
2674                guarded: false,
2675            };
2676            run_cmd(&db, &cmd, false, 1_000 + i as u64);
2677        }
2678
2679        let reached = |depth: Option<u32>| {
2680            let recall = Command::Recall {
2681                query: None,
2682                tags: Vec::new(),
2683                entities: vec!["a".into()],
2684                as_of: None,
2685                range: None,
2686                k: 64,
2687                closed: false,
2688                token_budget: Some(4096),
2689                ef: None,
2690                graph_depth: depth,
2691                vector: Vec::new(),
2692            };
2693            let (_, out) = run_cmd(&db, &recall, false, 5_000);
2694            out.lines().filter(|l| l.starts_with("- [f")).count()
2695        };
2696
2697        assert_eq!(reached(None), 3, "the configured default is 2 hops");
2698        assert_eq!(reached(Some(0)), 1, "no expansion: the anchor's own fact");
2699        assert_eq!(reached(Some(1)), 2);
2700        assert_eq!(reached(Some(3)), 4);
2701        // No hop ceiling, and an absurd depth terminates: the walk ends when a
2702        // pass adds no entity, not when a counter runs out.
2703        assert_eq!(reached(Some(99)), 4);
2704        assert_eq!(reached(Some(u32::MAX)), 4);
2705    }
2706
2707    #[test]
2708    fn stats_human_lists_the_counters() {
2709        let (db, _t) = TempDb::open();
2710        run_cmd(&db, &remember("a", None, &[]), false, 1_000);
2711        let (code, out) = run_cmd(&db, &Command::Stats, false, 2_000);
2712        assert_eq!(code, 0);
2713        assert!(out.contains("facts") && out.contains("pool_bytes"), "{out}");
2714    }
2715
2716    #[test]
2717    fn verify_command_renders_human_and_json() {
2718        let (db, _t) = TempDb::open();
2719        run_cmd(&db, &remember("clean", None, &[]), false, 1_000);
2720
2721        let (code, out) = run_cmd(&db, &Command::Verify, false, 2_000);
2722        assert_eq!(code, 0);
2723        assert_eq!(out.trim(), "integrity ok");
2724
2725        let (code, out) = run_cmd(&db, &Command::Verify, true, 2_100);
2726        assert_eq!(code, 0);
2727        let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2728        assert_eq!(v["ok"], true);
2729    }
2730
2731    #[test]
2732    fn show_json_of_a_revised_predecessor_is_closed() {
2733        let (db, _t) = TempDb::open();
2734        run_cmd(&db, &remember("v1", Some("e"), &[]), false, 1_000);
2735        let revise = Command::Revise {
2736            id: 0,
2737            text: "v2".into(),
2738            entity: Some("e".into()),
2739            tags: Vec::new(),
2740            links: Vec::new(),
2741            meta: Vec::new(),
2742            valid_from: None,
2743            vector: Vec::new(),
2744        };
2745        run_cmd(&db, &revise, false, 2_000);
2746        // the successor records `revises`; its card names the predecessor
2747        let (_, out) = run_cmd(&db, &Command::Show { id: 1 }, false, 3_000);
2748        assert!(out.contains("revises     fact 0"), "{out}");
2749    }
2750
2751    #[test]
2752    fn resolve_db_path_prefers_the_flag() {
2753        let p = "/tmp/explicit.plugmem";
2754        assert_eq!(resolve_db_path(Some(p), None, None), PathBuf::from(p));
2755        let configured = std::path::Path::new("/tmp/configured.plugmem");
2756        assert_eq!(
2757            resolve_db_path(None, Some(configured), None),
2758            PathBuf::from(configured)
2759        );
2760        // With no flag/config it falls back to $PLUGMEM_DB or the platform default — we
2761        // only assert the code path runs and yields some path.
2762        let _ = resolve_db_path(None, None, None);
2763    }
2764
2765    #[test]
2766    fn a_bare_name_is_a_memory_only_when_a_workspace_is_configured() {
2767        let root = PathBuf::from("/srv/bot");
2768
2769        // Without a workspace, everything is a path — this is the guard on the
2770        // default: the old behaviour of `--db` is not allowed to shift.
2771        assert_eq!(
2772            resolve_db_path(Some("work"), None, None),
2773            PathBuf::from("work")
2774        );
2775
2776        // With one, a bare name resolves inside it...
2777        assert_eq!(
2778            resolve_db_path(Some("work"), None, Some(&root)),
2779            PathBuf::from("/srv/bot/db/work.plugmem")
2780        );
2781        // ...and anything that is not a name stays a path, so an explicit file
2782        // is still reachable from inside a workspace.
2783        for path in ["./work", "work.plugmem", "/srv/other.plugmem", "../up"] {
2784            assert_eq!(
2785                resolve_db_path(Some(path), None, Some(&root)),
2786                PathBuf::from(path),
2787                "{path}"
2788            );
2789        }
2790
2791        // `[database].path` is a file setting, never a name.
2792        let configured = std::path::Path::new("work");
2793        assert_eq!(
2794            resolve_db_path(None, Some(configured), Some(&root)),
2795            PathBuf::from("work")
2796        );
2797    }
2798
2799    #[test]
2800    fn settings_help_runs_without_opening_a_database() {
2801        let cli = Cli::try_parse_from(["plugmem-cli", "help", "settings"]).unwrap();
2802        let mut output = Vec::new();
2803        assert_eq!(run_parsed(cli, &mut output), 0);
2804        let output = String::from_utf8(output).unwrap();
2805        assert!(output.contains("plugmem settings"));
2806        assert!(output.contains("[database]"));
2807        assert!(output.contains("path (path string"));
2808
2809        let cli = Cli::try_parse_from(["plugmem-cli", "--json", "help", "settings"]).unwrap();
2810        let mut output = Vec::new();
2811        assert_eq!(run_parsed(cli, &mut output), 0);
2812        let output: serde_json::Value = serde_json::from_slice(&output).unwrap();
2813        assert_eq!(output["topic"], "settings");
2814        assert!(output["config_path_precedence"].is_array());
2815        assert!(output["settings"].as_array().unwrap().len() > 10);
2816    }
2817
2818    #[test]
2819    fn run_parsed_opens_runs_and_reports() {
2820        let dir = std::env::temp_dir().join(format!(
2821            "plugmem-run-{}-{}",
2822            std::process::id(),
2823            now_ms_unique()
2824        ));
2825        std::fs::create_dir_all(&dir).unwrap();
2826        let path = dir.join("m.plugmem");
2827        let cli = Cli {
2828            db: Some(path.display().to_string()),
2829            workspace: None,
2830            config: None,
2831            json: false,
2832            command: Command::Stats,
2833        };
2834        let mut buf = Vec::new();
2835        let code = run_parsed(cli, &mut buf);
2836        assert_eq!(code, 0);
2837        assert!(String::from_utf8(buf).unwrap().contains("facts"));
2838        let _ = std::fs::remove_dir_all(&dir);
2839    }
2840
2841    #[test]
2842    fn recover_and_scrub_render_json_and_human_shapes() {
2843        let (db, tmp) = TempDb::open();
2844        let path = tmp.0.join("m.plugmem");
2845        run_cmd(&db, &remember("recoverable fact", None, &[]), false, 1_000);
2846        run_cmd(&db, &Command::Checkpoint, false, 2_000);
2847        drop(db);
2848
2849        let settings = settings_with(None);
2850        let mut out = Vec::new();
2851        assert_eq!(do_scrub(&path, &settings, true, &mut out), 0);
2852        let scrub: serde_json::Value = serde_json::from_slice(&out).unwrap();
2853        assert_eq!(scrub["ok"], true);
2854        assert!(scrub["bytes"].as_u64().unwrap() > 0);
2855        let mut out = Vec::new();
2856        assert_eq!(do_scrub(&path, &settings, false, &mut out), 0);
2857        let out = String::from_utf8(out).unwrap();
2858        assert!(out.contains("scrub ok:"), "{out}");
2859
2860        let json_dst = tmp.0.join("copy-json.plugmem");
2861        let mut out = Vec::new();
2862        assert_eq!(do_recover(&path, &json_dst, &settings, true, &mut out), 0);
2863        let recover: serde_json::Value = serde_json::from_slice(&out).unwrap();
2864        assert_eq!(recover["kept"], 1);
2865        assert_eq!(recover["dropped_text"], 0);
2866        assert_eq!(recover["dst"], json_dst.display().to_string());
2867
2868        let human_dst = tmp.0.join("copy-human.plugmem");
2869        let mut out = Vec::new();
2870        assert_eq!(do_recover(&path, &human_dst, &settings, false, &mut out), 0);
2871        let out = String::from_utf8(out).unwrap();
2872        assert!(out.contains("recovered to"), "{out}");
2873        assert!(out.contains("kept 1"), "{out}");
2874    }
2875
2876    #[test]
2877    fn readonly_dispatcher_renders_every_read_shape() {
2878        let (db, tmp) = TempDb::open();
2879        let path = tmp.0.join("m.plugmem");
2880        run_cmd(
2881            &db,
2882            &remember("readonly tokio fact", Some("plugmem"), &["pref"]),
2883            false,
2884            1_000,
2885        );
2886        run_cmd(&db, &Command::Checkpoint, false, 2_000);
2887        let ro = Database::open_readonly(&path, Config::default()).unwrap();
2888
2889        let mut out = Vec::new();
2890        assert_eq!(
2891            execute_ro(&ro, &Command::Stats, None, None, true, &mut out),
2892            0
2893        );
2894        let stats: serde_json::Value = serde_json::from_slice(&out).unwrap();
2895        assert_eq!(stats["facts"], 1);
2896
2897        let mut out = Vec::new();
2898        assert_eq!(
2899            execute_ro(&ro, &Command::Show { id: 0 }, None, None, false, &mut out),
2900            0
2901        );
2902        let text = String::from_utf8(out).unwrap();
2903        assert!(text.contains("readonly tokio fact"), "{text}");
2904
2905        let mut out = Vec::new();
2906        assert_eq!(
2907            execute_ro(&ro, &Command::Export, None, None, false, &mut out),
2908            0
2909        );
2910        let exported: serde_json::Value =
2911            serde_json::from_str(String::from_utf8(out).unwrap().lines().next().unwrap()).unwrap();
2912        assert_eq!(exported["text"], "readonly tokio fact");
2913
2914        let mut out = Vec::new();
2915        let recall = Command::Recall {
2916            query: Some("tokio".into()),
2917            tags: vec!["pref".into()],
2918            entities: vec!["plugmem".into()],
2919            as_of: None,
2920            range: None,
2921            k: 1,
2922            closed: false,
2923            token_budget: None,
2924            ef: None,
2925            graph_depth: None,
2926            vector: Vec::new(),
2927        };
2928        assert_eq!(execute_ro(&ro, &recall, None, None, false, &mut out), 0);
2929        let text = String::from_utf8(out).unwrap();
2930        assert!(text.contains("tokio"), "{text}");
2931
2932        let mut out = Vec::new();
2933        assert_eq!(
2934            execute_ro(&ro, &Command::Verify, None, None, true, &mut out),
2935            0
2936        );
2937        let verify: serde_json::Value = serde_json::from_slice(&out).unwrap();
2938        assert_eq!(verify["ok"], true);
2939    }
2940
2941    #[test]
2942    fn run_parsed_on_a_locked_database_returns_one() {
2943        let (_held, dir) = {
2944            let dir = std::env::temp_dir().join(format!(
2945                "plugmem-lock-{}-{}",
2946                std::process::id(),
2947                now_ms_unique()
2948            ));
2949            std::fs::create_dir_all(&dir).unwrap();
2950            let path = dir.join("m.plugmem");
2951            (Database::open(&path, Config::default()).unwrap(), dir)
2952        };
2953        let cli = Cli {
2954            db: Some(dir.join("m.plugmem").display().to_string()),
2955            workspace: None,
2956            config: None,
2957            json: false,
2958            command: Command::Stats,
2959        };
2960        let mut buf = Vec::new();
2961        assert_eq!(run_parsed(cli, &mut buf), 1);
2962        let _ = std::fs::remove_dir_all(&dir);
2963    }
2964
2965    #[test]
2966    fn run_parsed_propagates_a_usage_error_as_two() {
2967        let dir = std::env::temp_dir().join(format!(
2968            "plugmem-usage-{}-{}",
2969            std::process::id(),
2970            now_ms_unique()
2971        ));
2972        std::fs::create_dir_all(&dir).unwrap();
2973        let cli = Cli {
2974            db: Some(dir.join("m.plugmem").display().to_string()),
2975            workspace: None,
2976            config: None,
2977            json: false,
2978            command: Command::Remember {
2979                text: "x".into(),
2980                entity: None,
2981                tags: Vec::new(),
2982                links: vec!["bad".into()],
2983                meta: Vec::new(),
2984                valid_from: None,
2985                vector: Vec::new(),
2986                guarded: false,
2987            },
2988        };
2989        let mut buf = Vec::new();
2990        assert_eq!(run_parsed(cli, &mut buf), 2);
2991        let _ = std::fs::remove_dir_all(&dir);
2992    }
2993
2994    /// A scratch directory (no db) for config/checkpoint tests; removed on drop.
2995    struct Scratch(PathBuf);
2996    impl Scratch {
2997        fn new(tag: &str) -> Self {
2998            let dir = std::env::temp_dir().join(format!(
2999                "plugmem-cli-{tag}-{}-{}",
3000                std::process::id(),
3001                now_ms_unique()
3002            ));
3003            std::fs::create_dir_all(&dir).unwrap();
3004            Scratch(dir)
3005        }
3006    }
3007    impl Drop for Scratch {
3008        fn drop(&mut self) {
3009            let _ = std::fs::remove_dir_all(&self.0);
3010        }
3011    }
3012
3013    #[test]
3014    fn export_import_roundtrip_preserves_open_facts() {
3015        // A deliberately nested scenario: entities, multi-tag facts, a
3016        // revision (closes its predecessor), a forget (tombstone), and an
3017        // explicit valid_from — export must dump exactly the open facts, and
3018        // import must reconstruct that set faithfully.
3019        let (a, _ta) = TempDb::open();
3020        run_cmd(
3021            &a,
3022            &Command::Remember {
3023                text: "prefers tokio".into(),
3024                entity: Some("user".into()),
3025                tags: vec!["pref".into(), "lang".into()],
3026                links: Vec::new(),
3027                meta: vec!["uri=s3://b/x".into(), "src=chat".into()],
3028                valid_from: Some(500),
3029                vector: Vec::new(),
3030                guarded: false,
3031            },
3032            false,
3033            1_000,
3034        );
3035        run_cmd(
3036            &a,
3037            &remember("lives in Moscow", Some("user"), &[]),
3038            false,
3039            1_100,
3040        ); // id 1
3041        run_cmd(
3042            &a,
3043            &Command::Revise {
3044                id: 1,
3045                text: "lives in Berlin".into(),
3046                entity: Some("user".into()),
3047                tags: vec!["geo".into()],
3048                links: Vec::new(),
3049                meta: Vec::new(),
3050                valid_from: None,
3051                vector: Vec::new(),
3052            },
3053            false,
3054            1_200,
3055        ); // id 2 open, id 1 closed
3056        run_cmd(&a, &remember("junk", None, &[]), false, 1_300); // id 3
3057        run_cmd(&a, &Command::Forget { ids: vec![3] }, false, 1_400); // tombstone id 3
3058        run_cmd(
3059            &a,
3060            &remember("uses rust", Some("plugmem"), &["lang"]),
3061            false,
3062            1_500,
3063        ); // id 4
3064
3065        // Export A into a JSONL file.
3066        let mut dump = Vec::new();
3067        render_export(&a.export(), false, &mut dump);
3068        let scratch = Scratch::new("roundtrip");
3069        let file = scratch.0.join("dump.jsonl");
3070        std::fs::write(&file, &dump).unwrap();
3071
3072        // Import into a fresh B.
3073        let (b, _tb) = TempDb::open();
3074        let n = do_import(&b, 9_000, &file, 128, &mut Vec::new()).unwrap();
3075
3076        // Both sides, compared as sets keyed by the preserved fields.
3077        let key = |f: &ExportedFact| {
3078            let mut tags = f.tags.clone();
3079            tags.sort();
3080            (f.text.clone(), f.entity.clone(), tags, f.valid_from)
3081        };
3082        let mut ak: Vec<_> = a.export().iter().map(key).collect();
3083        let mut bk: Vec<_> = b.export().iter().map(key).collect();
3084        ak.sort();
3085        bk.sort();
3086        assert_eq!(n.facts, ak.len());
3087        assert_eq!(
3088            ak, bk,
3089            "roundtrip must preserve text/entity/tags/valid_from"
3090        );
3091
3092        // Spot-checks: the open facts survive with their metadata; the closed
3093        // revision and the tombstone do not.
3094        let b_open = b.export();
3095        assert!(b_open.iter().any(|f| f.text == "prefers tokio"
3096            && f.valid_from == 500
3097            && f.entity.as_deref() == Some("user")
3098            && f.tags == vec!["pref".to_string(), "lang".to_string()]
3099            && f.metadata.get("uri").map(String::as_str) == Some("s3://b/x")
3100            && f.metadata.get("src").map(String::as_str) == Some("chat")));
3101        assert!(b_open.iter().any(|f| f.text == "lives in Berlin"));
3102        assert!(b_open.iter().any(|f| f.text == "uses rust"));
3103        assert!(!b_open.iter().any(|f| f.text.contains("Moscow")));
3104        assert!(!b_open.iter().any(|f| f.text == "junk"));
3105    }
3106
3107    #[test]
3108    fn export_command_emits_jsonl_regardless_of_json_flag() {
3109        let (db, _t) = TempDb::open();
3110        run_cmd(&db, &remember("a fact", Some("e"), &["t"]), false, 1_000);
3111        for json in [false, true] {
3112            let (code, out) = run_cmd(&db, &Command::Export, json, 2_000);
3113            assert_eq!(code, 0);
3114            let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
3115            assert_eq!(v["text"], "a fact");
3116            assert_eq!(v["entity"], "e");
3117            assert_eq!(v["tags"][0], "t");
3118        }
3119    }
3120
3121    #[test]
3122    fn import_command_counts_and_rejects_bad_lines() {
3123        let (db, _t) = TempDb::open();
3124        let scratch = Scratch::new("import");
3125        let good = scratch.0.join("in.jsonl");
3126        std::fs::write(
3127            &good,
3128            "{\"text\":\"from jsonl\",\"entity\":\"user\",\"tags\":[\"x\"],\"valid_from\":42}\n\n{\"text\":\"second\"}\n",
3129        )
3130        .unwrap();
3131        // A tiny batch size exercises the streaming/chunking path (two batches).
3132        let n = do_import(&db, 9_000, &good, 1, &mut Vec::new()).unwrap();
3133        assert_eq!(n.facts, 2, "both facts imported, blank line skipped");
3134
3135        let bad = scratch.0.join("bad.jsonl");
3136        std::fs::write(&bad, "not json at all\n").unwrap();
3137        let err = do_import(&db, 9_000, &bad, 128, &mut Vec::new()).unwrap_err();
3138        assert!(matches!(err, CliError::Usage(_)));
3139    }
3140
3141    #[test]
3142    fn import_batch_size_does_not_change_the_result() {
3143        // The chunk size is a performance knob only: importing the same file with
3144        // batch 1 and batch 100 yields the identical fact set.
3145        let scratch = Scratch::new("import-batch");
3146        let file = scratch.0.join("facts.jsonl");
3147        let mut jsonl = String::new();
3148        for i in 0..5 {
3149            jsonl.push_str(&format!("{{\"text\":\"fact number {i}\"}}\n"));
3150        }
3151        std::fs::write(&file, &jsonl).unwrap();
3152
3153        let (a, _ta) = TempDb::open();
3154        let (b, _tb) = TempDb::open();
3155        let na = do_import(&a, 9_000, &file, 1, &mut Vec::new()).unwrap();
3156        let nb = do_import(&b, 9_000, &file, 100, &mut Vec::new()).unwrap();
3157
3158        assert_eq!(na.facts, 5);
3159        assert_eq!(nb.facts, 5);
3160        let texts = |db: &Database| {
3161            let mut t: Vec<_> = db.export().into_iter().map(|f| f.text).collect();
3162            t.sort();
3163            t
3164        };
3165        assert_eq!(texts(&a), texts(&b), "batch size must not change the facts");
3166    }
3167
3168    #[test]
3169    fn config_table_feeds_settings_and_the_cli_batch_size() {
3170        // The CLI reads config.toml once (host `read_config`), builds the
3171        // shared `Settings`, and pulls its own `batch_size` from the same
3172        // table — the exact flow of `run_parsed`.
3173        let scratch = Scratch::new("settings");
3174        let cfgfile = scratch.0.join("config.toml");
3175        std::fs::write(
3176            &cfgfile,
3177            "[engine]\ndim = 512\n[embedder]\nenabled = false\n\
3178             [maintenance]\nsnapshot_every_ops = 64\nbatch_size = 200\n",
3179        )
3180        .unwrap();
3181        let table = plugmem_host::read_config(Some(&cfgfile)).unwrap();
3182        let s = Settings::from_table(table.as_ref()).unwrap();
3183        assert_eq!(s.config.dim, 512);
3184        assert!(s.embedder.is_none());
3185        assert_eq!(s.snapshot_every_ops, Some(64));
3186        assert_eq!(read_batch_size(table.as_ref()), Some(200));
3187
3188        // An explicit --config that does not exist is a usage error.
3189        assert!(plugmem_host::read_config(Some(&scratch.0.join("nope.toml"))).is_err());
3190    }
3191
3192    #[test]
3193    fn checkpoint_command_flushes_the_journal_and_enables_the_readonly_path() {
3194        let scratch = Scratch::new("checkpoint-cmd");
3195        let path = scratch.0.join("m.plugmem");
3196
3197        // A remember through the read-write path leaves a dirty journal.
3198        let remember = Cli {
3199            db: Some(path.display().to_string()),
3200            workspace: None,
3201            config: None,
3202            json: false,
3203            command: Command::Remember {
3204                text: "hello tokio".into(),
3205                entity: None,
3206                tags: Vec::new(),
3207                links: Vec::new(),
3208                meta: Vec::new(),
3209                valid_from: None,
3210                vector: Vec::new(),
3211                guarded: false,
3212            },
3213        };
3214        assert_eq!(run_parsed(remember, &mut Vec::new()), 0);
3215
3216        // The new command: human shape.
3217        let checkpoint = |json| Cli {
3218            db: Some(path.display().to_string()),
3219            workspace: None,
3220            config: None,
3221            json,
3222            command: Command::Checkpoint,
3223        };
3224        let mut buf = Vec::new();
3225        assert_eq!(run_parsed(checkpoint(false), &mut buf), 0);
3226        assert!(String::from_utf8(buf).unwrap().contains("checkpointed"));
3227
3228        // json shape.
3229        let mut buf = Vec::new();
3230        assert_eq!(run_parsed(checkpoint(true), &mut buf), 0);
3231        let v: serde_json::Value =
3232            serde_json::from_str(String::from_utf8(buf).unwrap().trim()).unwrap();
3233        assert_eq!(v["ok"], true);
3234
3235        // The journal is now clean, so scrub (a shared-lock, read-only open)
3236        // succeeds — it would fail `NeedsCheckpoint` on a dirty journal.
3237        let scrub = Cli {
3238            db: Some(path.display().to_string()),
3239            workspace: None,
3240            config: None,
3241            json: false,
3242            command: Command::Scrub,
3243        };
3244        let mut buf = Vec::new();
3245        assert_eq!(run_parsed(scrub, &mut buf), 0);
3246        assert!(String::from_utf8(buf).unwrap().contains("scrub ok"));
3247    }
3248
3249    #[test]
3250    fn run_parsed_uses_the_readonly_path_after_a_checkpoint() {
3251        let scratch = Scratch::new("ro-route");
3252        let path = scratch.0.join("m.plugmem");
3253        {
3254            let (db, _) = Database::open(&path, Config::default()).unwrap();
3255            db.remember(RememberInput::text(1_000, "hello tokio"))
3256                .unwrap();
3257            db.checkpoint(2_000).unwrap(); // empty journal → open_readonly succeeds
3258        }
3259        // stats routes through open_readonly (mmap, shared)
3260        let cli = Cli {
3261            db: Some(path.display().to_string()),
3262            workspace: None,
3263            config: None,
3264            json: false,
3265            command: Command::Stats,
3266        };
3267        let mut buf = Vec::new();
3268        assert_eq!(run_parsed(cli, &mut buf), 0);
3269        assert!(String::from_utf8(buf).unwrap().contains("facts"));
3270
3271        // recall with no embedder also uses the read-only path
3272        let cli = Cli {
3273            db: Some(path.display().to_string()),
3274            workspace: None,
3275            config: None,
3276            json: false,
3277            command: Command::Recall {
3278                query: Some("tokio".into()),
3279                tags: Vec::new(),
3280                entities: Vec::new(),
3281                as_of: None,
3282                range: None,
3283                k: 0,
3284                closed: false,
3285                token_budget: None,
3286                ef: None,
3287                graph_depth: None,
3288                vector: Vec::new(),
3289            },
3290        };
3291        let mut buf = Vec::new();
3292        assert_eq!(run_parsed(cli, &mut buf), 0);
3293        assert!(String::from_utf8(buf).unwrap().contains("tokio"));
3294    }
3295}