1#![doc = include_str!("../README.md")]
5mod cli;
20mod config;
21mod workspace;
22
23use std::collections::BTreeMap;
24use std::io::{self, BufRead, Write};
25use std::path::{Path, PathBuf};
26use std::process::ExitCode;
27use std::time::{SystemTime, UNIX_EPOCH};
28
29use clap::Parser;
30use plugmem_host::{
31 Database, ExportedFact, FactId, HostError, LinkInput, MaintenanceMode, MaintenanceOptions,
32 ReadOnlyDatabase, RecallQuery, RecallResult, RememberInput, RememberOutcome, Settings, Stats,
33 UnlinkInput, VALID_TO_OPEN,
34};
35use serde_json::json;
36
37use crate::cli::{Cli, Command, HelpTopic, MaintainMode};
38use crate::config::read_batch_size;
39
40pub(crate) const ENV_DB: &str = "PLUGMEM_DB";
42pub(crate) const DEFAULT_DB: &str = "plugmem.db";
44
45#[derive(Debug)]
48pub(crate) enum CliError {
49 Host(HostError),
50 Usage(String),
51}
52
53impl From<HostError> for CliError {
54 fn from(e: HostError) -> Self {
55 CliError::Host(e)
56 }
57}
58
59pub(crate) fn now_ms() -> u64 {
61 SystemTime::now()
62 .duration_since(UNIX_EPOCH)
63 .map(|d| d.as_millis() as u64)
64 .unwrap_or(0)
65}
66
67pub fn run() -> ExitCode {
71 let stdout = io::stdout();
72 ExitCode::from(run_parsed(Cli::parse(), &mut stdout.lock()))
73}
74
75fn run_parsed(cli: Cli, out: &mut impl Write) -> u8 {
79 if let Command::Help { topic } = &cli.command {
80 return execute_help(topic, cli.json, out);
81 }
82
83 let table = match plugmem_host::read_config(cli.config.as_deref()) {
87 Ok(t) => t,
88 Err(e) => return report_err(&e.into()),
89 };
90 let cfg_batch_size = read_batch_size(table.as_ref());
91 let mut settings = match Settings::from_table(table.as_ref()) {
92 Ok(s) => s,
93 Err(e) => return report_err(&e.into()),
94 };
95 for warning in &settings.warnings {
99 eprintln!("plugmem: {warning}");
100 }
101 let root = workspace::resolve_root(cli.workspace.as_deref(), &settings);
106 if let Command::Workspace { command } = &cli.command {
107 return match workspace::execute(command, root, settings, cli.json, out) {
108 Ok(code) => code,
109 Err(e) => {
110 let _ = out.flush();
111 report_err(&e)
112 }
113 };
114 }
115 let path = resolve_db_path(
116 cli.db.as_deref(),
117 settings.database_path.as_deref(),
118 root.as_ref(),
119 );
120 if let Err(e) = workspace::ensure_dir(&path, root.as_ref()) {
121 return report_err(&e);
122 }
123
124 match &cli.command {
129 Command::Recover { dst } => return do_recover(&path, dst, &settings, cli.json, out),
130 Command::Scrub => return do_scrub(&path, &settings, cli.json, out),
131 Command::Repl { read_only: true } => {
136 return run_repl_ro(&path, settings, cli.json, io::stdin().lock(), out);
137 }
138 Command::Repl { read_only: false } => {
139 return run_repl(&path, settings, cli.json, io::stdin().lock(), out);
140 }
141 _ => {}
142 }
143
144 let readonly_ok = matches!(
152 &cli.command,
153 Command::Show { .. }
154 | Command::Stats
155 | Command::Export
156 | Command::Verify
157 | Command::Recall { .. }
158 );
159 if readonly_ok {
160 let recall_vector = match embed_recall_query(&mut settings, &cli.command) {
161 Ok(v) => v,
162 Err(e) => return report_err(&e),
163 };
164 match Database::open_readonly(&path, settings.config.clone()) {
165 Ok(ro) => {
166 return execute_ro(&ro, &cli.command, recall_vector.as_deref(), cli.json, out);
167 }
168 Err(HostError::Locked { path }) => return report_locked(&path),
169 Err(_) => {}
174 }
175 }
176
177 let db = match settings.open(&path) {
180 Ok(db) => db,
181 Err(HostError::Locked { path }) => return report_locked(&path),
182 Err(e) => return report_err(&CliError::Host(e)),
183 };
184 if let Command::Import { file, batch } = &cli.command {
187 let batch_size = batch
188 .or(cfg_batch_size.map(|n| n as usize))
189 .unwrap_or(DEFAULT_IMPORT_BATCH)
190 .max(1);
191 return match do_import(&db, now_ms(), file, batch_size, out) {
192 Ok(report) => {
193 if cli.json {
194 writeln!(
195 out,
196 "{}",
197 json!({ "imported": report.facts, "edges": report.edges })
198 )
199 .ok();
200 } else if report.edges == 0 {
201 writeln!(out, "imported {} facts", report.facts).ok();
202 } else {
203 writeln!(
204 out,
205 "imported {} facts and {} edges",
206 report.facts, report.edges
207 )
208 .ok();
209 }
210 0
211 }
212 Err(e) => {
213 let _ = out.flush();
214 report_err(&e)
215 }
216 };
217 }
218 match execute(&db, &cli.command, cli.json, now_ms(), out) {
219 Ok(code) => code,
220 Err(e) => {
221 let _ = out.flush();
222 report_err(&e)
223 }
224 }
225}
226
227const DEFAULT_IMPORT_BATCH: usize = 128;
230
231fn write_err(out: &mut impl Write, e: &CliError) {
238 let _ = match e {
239 CliError::Usage(msg) => writeln!(out, "plugmem: {msg}"),
240 CliError::Host(err) => {
241 writeln!(out, "plugmem: {err}").and_then(|()| match err.capacity_hint() {
242 Some(hint) => writeln!(out, "plugmem: {hint}"),
243 None => Ok(()),
244 })
245 }
246 };
247}
248
249fn report_err(e: &CliError) -> u8 {
251 write_err(&mut std::io::stderr(), e);
252 2
253}
254
255fn report_locked(path: &std::path::Path) -> u8 {
257 eprintln!(
258 "plugmem: database is locked by another process: {}",
259 path.display()
260 );
261 1
262}
263
264fn resolve_db_path(
271 flag: Option<&str>,
272 config_path: Option<&std::path::Path>,
273 root: Option<&PathBuf>,
274) -> PathBuf {
275 flag.map(|value| workspace::resolve_target(value, root))
276 .or_else(|| {
277 std::env::var_os(ENV_DB).map(|v| workspace::resolve_target(&v.to_string_lossy(), root))
278 })
279 .or_else(|| config_path.map(PathBuf::from))
280 .or_else(plugmem_host::default_database_path)
281 .unwrap_or_else(|| PathBuf::from(DEFAULT_DB))
282}
283
284fn execute_help(topic: &HelpTopic, json_output: bool, out: &mut impl Write) -> u8 {
287 match topic {
288 HelpTopic::Settings => {
289 if json_output {
290 let help = plugmem_host::settings_help();
291 let settings: Vec<_> = help
292 .docs()
293 .iter()
294 .map(|doc| {
295 json!({
296 "section": doc.section,
297 "key": doc.key,
298 "type": doc.value_type,
299 "default": doc.default,
300 "description": doc.description,
301 "scope": doc.scope.as_str(),
302 })
303 })
304 .collect();
305 let value = json!({
306 "topic": "settings",
307 "config_path_precedence": help.config_path_precedence(),
308 "default_config_path": plugmem_host::default_config_path()
309 .map(|path| path.display().to_string()),
310 "settings": settings,
311 });
312 writeln!(out, "{value}").ok();
313 } else {
314 write!(out, "{}", plugmem_host::settings_help().render_human()).ok();
315 }
316 0
317 }
318 }
319}
320
321fn execute_ro(
324 ro: &ReadOnlyDatabase,
325 cmd: &Command,
326 recall_vector: Option<&[f32]>,
327 json: bool,
328 out: &mut impl Write,
329) -> u8 {
330 match cmd {
331 Command::Recall { .. } => {
332 match with_recall_query(cmd, now_ms(), recall_vector, |q| ro.recall(q)) {
333 Ok(res) => {
334 render_recall(&res, json, out);
335 0
336 }
337 Err(e) => report_err(&CliError::Host(e)),
338 }
339 }
340 Command::Show { id } => render_show(ro.get(FactId(*id)), *id, json, out),
341 Command::Stats => {
342 render_stats(&ro.stats(), json, out);
343 0
344 }
345 Command::Export => {
346 ro.export_each(|f| write_export_line(out, &f));
347 ro.export_edges_each(|src, rel, dst, fact| write_export_edge(out, src, rel, dst, fact));
348 0
349 }
350 Command::Verify => match ro.verify() {
352 Ok(()) => {
353 if json {
354 writeln!(out, "{}", json!({ "ok": true })).ok();
355 } else {
356 writeln!(out, "integrity ok").ok();
357 }
358 0
359 }
360 Err(e) => report_err(&CliError::Host(e)),
361 },
362 _ => unreachable!("execute_ro only receives read-only commands"),
363 }
364}
365
366fn execute(
370 db: &Database,
371 cmd: &Command,
372 json: bool,
373 now: u64,
374 out: &mut impl Write,
375) -> Result<u8, CliError> {
376 match cmd {
377 Command::Remember {
378 text,
379 entity,
380 tags,
381 links,
382 meta,
383 valid_from,
384 vector,
385 } => {
386 let outcome = do_remember(
387 db,
388 now,
389 text,
390 entity,
391 tags,
392 links,
393 meta,
394 *valid_from,
395 vector,
396 None,
397 )?;
398 render_remember(&outcome, json, out);
399 Ok(0)
400 }
401 Command::Revise {
402 id,
403 text,
404 entity,
405 tags,
406 links,
407 meta,
408 valid_from,
409 vector,
410 } => {
411 let outcome = do_remember(
412 db,
413 now,
414 text,
415 entity,
416 tags,
417 links,
418 meta,
419 *valid_from,
420 vector,
421 Some(FactId(*id)),
422 )?;
423 render_remember(&outcome, json, out);
424 Ok(0)
425 }
426 Command::Recall { .. } => {
427 let res = with_recall_query(cmd, now, None, |q| db.recall(q))?;
428 render_recall(&res, json, out);
429 Ok(0)
430 }
431 Command::Forget { id } => {
432 let fresh = db.forget(now, FactId(*id))?;
433 if json {
434 writeln!(out, "{}", json!({ "id": id, "forgotten": fresh })).ok();
435 } else if fresh {
436 writeln!(out, "forgot fact {id}").ok();
437 } else {
438 writeln!(out, "fact {id} was already gone").ok();
439 }
440 Ok(0)
441 }
442 Command::Link {
443 src,
444 rel,
445 dst,
446 provenance,
447 } => {
448 db.link(LinkInput {
449 now,
450 src,
451 rel,
452 dst,
453 provenance: provenance.map(FactId),
454 })?;
455 if json {
456 writeln!(out, "{}", json!({ "src": src, "rel": rel, "dst": dst })).ok();
457 } else {
458 writeln!(out, "linked {src} -{rel}-> {dst}").ok();
459 }
460 Ok(0)
461 }
462 Command::Unlink { src, rel, dst } => {
463 let fresh = db.unlink(UnlinkInput { now, src, rel, dst })?;
464 if json {
465 writeln!(
466 out,
467 "{}",
468 json!({ "src": src, "rel": rel, "dst": dst, "unlinked": fresh })
469 )
470 .ok();
471 } else if fresh {
472 writeln!(out, "unlinked {src} -{rel}-> {dst}").ok();
473 } else {
474 writeln!(out, "edge {src} -{rel}-> {dst} was already absent").ok();
475 }
476 Ok(0)
477 }
478 Command::Show { id } => Ok(render_show(db.get(FactId(*id)), *id, json, out)),
479 Command::Stats => {
480 render_stats(&db.stats(), json, out);
481 Ok(0)
482 }
483 Command::Export => {
484 db.export_each(|f| write_export_line(out, &f));
485 db.export_edges_each(|src, rel, dst, fact| write_export_edge(out, src, rel, dst, fact));
486 Ok(0)
487 }
488 Command::Maintain { mode } => {
489 let report = db.maintain_with_options(now, maintenance_options(*mode))?;
490 if json {
491 writeln!(
492 out,
493 "{}",
494 json!({
495 "purged": report.purged,
496 "bytes_before": report.bytes_before,
497 "bytes_after": report.bytes_after,
498 "no_op": report.no_op,
499 "tombstones_before": report.tombstones_before,
500 "facts_before": report.facts_before,
501 "facts_after": report.facts_after,
502 "vectors_before": report.vectors_before,
503 "vectors_after": report.vectors_after,
504 "hnsw_indexed_before": report.hnsw_indexed_before,
505 "hnsw_indexed_after": report.hnsw_indexed_after,
506 "structural_compacted": report.structural_compacted,
507 "bm25_compacted": report.bm25_compacted,
508 "bm25_reindexed": report.bm25_reindexed,
509 "hnsw_rebuilt": report.hnsw_rebuilt,
510 "hnsw_remapped": report.hnsw_remapped,
511 "hnsw_inserted": report.hnsw_inserted,
512 "edges_compacted": report.edges_compacted,
513 "edges_before": report.edges_before,
514 "edge_versions_before": report.edge_versions_before,
515 })
516 )
517 .ok();
518 } else {
519 writeln!(
520 out,
521 "maintained: purged {}, {} -> {} bytes, hnsw +{}, bm25 {}{}{}",
522 report.purged,
523 report.bytes_before,
524 report.bytes_after,
525 report.hnsw_inserted,
526 if report.bm25_reindexed {
527 "reindexed"
528 } else if report.bm25_compacted {
529 "compacted"
530 } else {
531 "unchanged"
532 },
533 if report.edges_compacted {
534 ", edges repacked"
535 } else {
536 ""
537 },
538 if report.no_op { " (no-op)" } else { "" }
539 )
540 .ok();
541 }
542 Ok(0)
543 }
544 Command::Checkpoint => {
545 db.checkpoint(now)?;
546 if json {
547 writeln!(out, "{}", json!({ "ok": true })).ok();
548 } else {
549 writeln!(out, "checkpointed: journal flushed to snapshot").ok();
550 }
551 Ok(0)
552 }
553 Command::Verify => {
554 db.verify()?;
557 if json {
558 writeln!(out, "{}", json!({ "ok": true })).ok();
559 } else {
560 writeln!(out, "integrity ok").ok();
561 }
562 Ok(0)
563 }
564 Command::Scrub
566 | Command::Recover { .. }
567 | Command::Repl { .. }
568 | Command::Import { .. }
569 | Command::Workspace { .. }
570 | Command::Help { .. } => {
571 unreachable!("this command is dispatched before execute")
572 }
573 }
574}
575
576fn do_recover(src: &Path, dst: &Path, settings: &Settings, json: bool, out: &mut impl Write) -> u8 {
580 match Database::recover(src, dst, settings.config.clone(), now_ms()) {
581 Ok(r) => {
582 if json {
583 writeln!(
584 out,
585 "{}",
586 json!({
587 "kept": r.kept,
588 "dropped_text": r.dropped_text,
589 "dropped_vector": r.dropped_vector,
590 "dropped_metadata": r.dropped_metadata,
591 "dst": dst.display().to_string(),
592 })
593 )
594 .ok();
595 } else {
596 writeln!(
597 out,
598 "recovered to {}: kept {}, dropped {} text + {} vector + {} metadata",
599 dst.display(),
600 r.kept,
601 r.dropped_text,
602 r.dropped_vector,
603 r.dropped_metadata
604 )
605 .ok();
606 }
607 0
608 }
609 Err(HostError::Locked { path }) => report_locked(&path),
610 Err(e) => report_err(&CliError::Host(e)),
611 }
612}
613
614fn do_scrub(path: &Path, settings: &Settings, json: bool, out: &mut impl Write) -> u8 {
619 let ro = match Database::open_readonly(path, settings.config.clone()) {
620 Ok(ro) => ro,
621 Err(HostError::Locked { path }) => return report_locked(&path),
622 Err(e) => return report_err(&CliError::Host(e)),
623 };
624 let scrub = match ro.scrub() {
625 Ok(s) => s,
626 Err(e) => return report_err(&CliError::Host(e)),
627 };
628 let mut done = 0u64;
629 let mut total = 0u64;
630 for step in scrub {
631 match step {
632 Ok(p) => {
633 done = p.done_bytes;
634 total = p.total_bytes;
635 }
636 Err(e) => return report_err(&CliError::Host(e)),
637 }
638 }
639 if json {
640 writeln!(out, "{}", json!({ "ok": true, "bytes": done })).ok();
641 } else {
642 writeln!(out, "scrub ok: {done}/{total} bytes verified").ok();
643 }
644 0
645}
646
647#[derive(Parser)]
650#[command(
651 no_binary_name = true,
652 name = "plugmem",
653 disable_help_subcommand = true
654)]
655struct ReplLine {
656 #[command(subcommand)]
657 command: Command,
658}
659
660fn split_line(line: &str) -> Vec<String> {
664 let mut tokens = Vec::new();
665 let mut cur = String::new();
666 let mut quote: Option<char> = None;
667 let mut has = false;
668 for c in line.chars() {
669 match quote {
670 Some(q) => {
671 if c == q {
672 quote = None;
673 } else {
674 cur.push(c);
675 }
676 }
677 None if c == '"' || c == '\'' => {
678 quote = Some(c);
679 has = true;
680 }
681 None if c.is_whitespace() => {
682 if has {
683 tokens.push(std::mem::take(&mut cur));
684 has = false;
685 }
686 }
687 None => {
688 cur.push(c);
689 has = true;
690 }
691 }
692 }
693 if has {
694 tokens.push(cur);
695 }
696 tokens
697}
698
699fn run_repl(
705 path: &Path,
706 settings: Settings,
707 json: bool,
708 input: impl BufRead,
709 out: &mut impl Write,
710) -> u8 {
711 let db = match settings.open(path) {
712 Ok(db) => db,
713 Err(HostError::Locked { path }) => return report_locked(&path),
714 Err(e) => return report_err(&CliError::Host(e)),
715 };
716 eprintln!("plugmem repl — one open handle, host speed. `help` for verbs, `exit` to quit.");
717 eprint!("plugmem> ");
718 for line in input.lines() {
719 let Ok(line) = line else { break };
720 let line = line.trim();
721 if line.is_empty() {
722 eprint!("plugmem> ");
723 continue;
724 }
725 if line == "exit" || line == "quit" {
726 break;
727 } else if line == "help" {
728 writeln!(
729 out,
730 "verbs: remember recall revise forget link unlink show stats maintain checkpoint \
731 verify export import (scrub/recover stay one-shot) exit"
732 )
733 .ok();
734 } else {
735 run_repl_line(&db, line, json, out);
736 }
737 eprint!("plugmem> ");
738 }
739 eprintln!();
740 match db.checkpoint(now_ms()) {
742 Ok(()) => 0,
743 Err(e) => report_err(&CliError::Host(e)),
744 }
745}
746
747fn run_repl_line(db: &Database, line: &str, json: bool, out: &mut impl Write) {
750 let cmd = match ReplLine::try_parse_from(split_line(line)) {
751 Ok(r) => r.command,
752 Err(e) => {
754 let _ = writeln!(out, "{e}");
755 return;
756 }
757 };
758 match &cmd {
759 Command::Repl { .. } => {
760 let _ = writeln!(out, "already in a repl session");
761 }
762 Command::Scrub | Command::Recover { .. } => {
763 let _ = writeln!(
764 out,
765 "scrub/recover are one-shot commands; run them outside the repl"
766 );
767 }
768 _ => {
769 if let Err(e) = execute(db, &cmd, json, now_ms(), out) {
770 write_err(out, &e);
771 }
772 }
773 }
774}
775
776fn run_repl_ro(
789 path: &Path,
790 mut settings: Settings,
791 json: bool,
792 input: impl BufRead,
793 out: &mut impl Write,
794) -> u8 {
795 let mut ro = match Database::open_readonly(path, settings.config.clone()) {
796 Ok(ro) => ro,
797 Err(HostError::Locked { path }) => return report_locked(&path),
798 Err(e) => return report_err(&CliError::Host(e)),
801 };
802 eprintln!(
803 "plugmem repl --read-only — observing generation {} of another process's writer. \
804 `help` for verbs, `refresh`/`generation` for cross-process freshness, `exit` to quit.",
805 ro.generation()
806 );
807 eprint!("plugmem(ro)> ");
808 for line in input.lines() {
809 let Ok(line) = line else { break };
810 let line = line.trim();
811 if line.is_empty() {
812 eprint!("plugmem(ro)> ");
813 continue;
814 }
815 match line {
816 "exit" | "quit" => break,
817 "help" => {
818 writeln!(
819 out,
820 "read verbs: recall show stats export verify \
821 freshness: generation refresh exit \
822 (writes and scrub/recover are refused in a read-only session)"
823 )
824 .ok();
825 }
826 "generation" => {
829 let g = ro.generation();
830 if json {
831 writeln!(out, "{}", json!({ "generation": g })).ok();
832 } else {
833 writeln!(out, "generation {g}").ok();
834 }
835 }
836 "refresh" => match ro.refresh() {
837 Ok(advanced) => {
838 let g = ro.generation();
839 if json {
840 writeln!(out, "{}", json!({ "advanced": advanced, "generation": g })).ok();
841 } else if advanced {
842 writeln!(out, "refreshed → generation {g}").ok();
843 } else {
844 writeln!(out, "already current → generation {g}").ok();
845 }
846 }
847 Err(e) => write_err(out, &CliError::Host(e)),
848 },
849 _ => run_repl_ro_line(&ro, &mut settings, line, json, out),
850 }
851 eprint!("plugmem(ro)> ");
852 }
853 eprintln!();
854 0
856}
857
858fn run_repl_ro_line(
861 ro: &ReadOnlyDatabase,
862 settings: &mut Settings,
863 line: &str,
864 json: bool,
865 out: &mut impl Write,
866) {
867 let cmd = match ReplLine::try_parse_from(split_line(line)) {
868 Ok(r) => r.command,
869 Err(e) => {
870 let _ = writeln!(out, "{e}");
871 return;
872 }
873 };
874 let readable = matches!(
875 &cmd,
876 Command::Show { .. }
877 | Command::Stats
878 | Command::Export
879 | Command::Verify
880 | Command::Recall { .. }
881 );
882 if !readable {
883 let _ = writeln!(
884 out,
885 "read-only session: only recall/show/stats/export/verify run \
886 (plus refresh/generation); writes and one-shot commands need a writer handle"
887 );
888 return;
889 }
890 let recall_vector = match embed_recall_query(settings, &cmd) {
893 Ok(v) => v,
894 Err(e) => {
895 write_err(out, &e);
896 return;
897 }
898 };
899 let _ = execute_ro(ro, &cmd, recall_vector.as_deref(), json, out);
900}
901
902fn embed_recall_query(
910 settings: &mut Settings,
911 cmd: &Command,
912) -> Result<Option<Vec<f32>>, CliError> {
913 let Command::Recall {
914 query: Some(text),
915 vector,
916 ..
917 } = cmd
918 else {
919 return Ok(None);
920 };
921 if !vector.is_empty() {
924 return Ok(None);
925 }
926 let Some(embedder) = settings.embedder.as_ref() else {
927 return Ok(None);
928 };
929 let mut vectors = embedder.embed(&[text.as_str()]).map_err(CliError::Host)?;
930 Ok(vectors.pop())
931}
932
933fn with_recall_query<R>(
938 cmd: &Command,
939 now: u64,
940 override_vector: Option<&[f32]>,
941 f: impl FnOnce(RecallQuery<'_>) -> R,
942) -> R {
943 let Command::Recall {
944 query,
945 tags,
946 entities,
947 as_of,
948 range,
949 k,
950 closed,
951 token_budget,
952 ef,
953 graph_depth,
954 vector,
955 } = cmd
956 else {
957 unreachable!("with_recall_query called on a non-recall command");
958 };
959 let tag_refs: Vec<&str> = tags.iter().map(String::as_str).collect();
960 let ent_refs: Vec<&str> = entities.iter().map(String::as_str).collect();
961 let range_pair = range.as_ref().map(|v| (v[0], v[1]));
962 let explicit = (!vector.is_empty()).then_some(vector.as_slice());
967 let q = RecallQuery {
968 now,
969 text: query.as_deref(),
970 vector: explicit.or(override_vector),
971 tags: &tag_refs,
972 entities: &ent_refs,
973 as_of: *as_of,
974 range: range_pair,
975 k: *k,
976 token_budget: *token_budget,
977 include_closed: *closed,
978 ef: *ef,
979 graph_depth: *graph_depth,
980 };
981 f(q)
982}
983
984fn render_recall(res: &RecallResult, json: bool, out: &mut impl Write) {
987 if json {
988 let facts: Vec<_> = res
989 .facts
990 .iter()
991 .map(|f| {
992 json!({
993 "id": f.id.0,
994 "score": f.score,
995 "sources": f.sources,
996 "recorded_at": f.recorded_at,
997 "valid_from": f.valid_from,
998 "valid_to": open_or(f.valid_to),
999 })
1000 })
1001 .collect();
1002 let edges: Vec<_> = res
1007 .edges
1008 .iter()
1009 .map(|e| {
1010 json!({
1011 "src": e.src.0,
1012 "rel": e.rel.0,
1013 "dst": e.dst.0,
1014 "provenance": (e.provenance != FactId::NONE).then_some(e.provenance.0),
1015 })
1016 })
1017 .collect();
1018 writeln!(
1019 out,
1020 "{}",
1021 json!({
1022 "facts": facts,
1023 "edges": edges,
1024 "rendered": res.rendered,
1025 "truncated": res.truncated,
1026 })
1027 )
1028 .ok();
1029 } else if res.rendered.is_empty() {
1030 writeln!(out, "(nothing recalled)").ok();
1031 } else {
1032 writeln!(out, "{}", res.rendered).ok();
1033 }
1034}
1035
1036fn render_show(
1038 fact: Option<plugmem_host::FactSnapshot>,
1039 id: u32,
1040 json: bool,
1041 out: &mut impl Write,
1042) -> u8 {
1043 let Some(fact) = fact else {
1044 if json {
1045 writeln!(out, "{}", json!({ "id": id, "found": false })).ok();
1046 } else {
1047 writeln!(out, "fact {id} not found").ok();
1048 }
1049 return 1;
1050 };
1051 let r = &fact.record;
1052 if json {
1053 writeln!(
1054 out,
1055 "{}",
1056 json!({
1057 "id": r.id.0,
1058 "text": fact.text,
1059 "recorded_at": r.recorded_at,
1060 "valid_from": r.valid_from,
1061 "valid_to": open_or(r.valid_to),
1062 "closed": r.is_closed(),
1063 "tombstone": r.is_tombstone(),
1064 "revises": (r.revises != FactId::NONE).then_some(r.revises.0),
1065 "metadata": fact.metadata,
1066 })
1067 )
1068 .ok();
1069 } else {
1070 writeln!(out, "fact {}", r.id.0).ok();
1071 writeln!(out, " text {}", fact.text).ok();
1072 writeln!(out, " recorded_at {}", r.recorded_at).ok();
1073 write!(out, " valid [{}, ", r.valid_from).ok();
1074 match r.valid_to {
1075 VALID_TO_OPEN => writeln!(out, "open)").ok(),
1076 to => writeln!(out, "{to})").ok(),
1077 };
1078 if r.revises != FactId::NONE {
1079 writeln!(out, " revises fact {}", r.revises.0).ok();
1080 }
1081 if !fact.metadata.is_empty() {
1082 let rendered = fact
1083 .metadata
1084 .iter()
1085 .map(|(k, v)| format!("{k}={v}"))
1086 .collect::<Vec<_>>()
1087 .join(", ");
1088 writeln!(out, " metadata {rendered}").ok();
1089 }
1090 if r.is_tombstone() {
1091 writeln!(out, " state tombstoned").ok();
1092 }
1093 }
1094 0
1095}
1096
1097fn maintenance_options(mode: MaintainMode) -> MaintenanceOptions {
1102 match MaintenanceMode::from(mode) {
1103 MaintenanceMode::Auto => MaintenanceOptions::auto(),
1104 MaintenanceMode::Full => MaintenanceOptions::full(),
1105 mode => MaintenanceOptions {
1106 mode,
1107 ..MaintenanceOptions::auto()
1108 },
1109 }
1110}
1111
1112fn render_stats(s: &Stats, json: bool, out: &mut impl Write) {
1114 if json {
1115 writeln!(
1116 out,
1117 "{}",
1118 json!({
1119 "facts": s.facts,
1120 "entities": s.entities,
1121 "terms": s.terms,
1122 "edges": s.edges,
1123 "edge_versions": s.edge_versions,
1124 "vectors": s.vectors,
1125 "hnsw_indexed": s.hnsw_indexed,
1126 "next_fact": s.next_fact,
1127 "next_entity": s.next_entity,
1128 "next_edge": s.next_edge,
1129 "pool_bytes": s.pool_bytes,
1130 "shards": {
1131 "facts": s.shards.facts,
1132 "entities": s.shards.entities,
1133 "edges": s.shards.edges,
1134 "temporal": s.shards.temporal,
1135 "postings": s.shards.postings,
1136 },
1137 })
1138 )
1139 .ok();
1140 } else {
1141 writeln!(out, "facts {}", s.facts).ok();
1142 writeln!(out, "entities {}", s.entities).ok();
1143 writeln!(out, "terms {}", s.terms).ok();
1144 writeln!(out, "edges {}", s.edges).ok();
1145 writeln!(out, "edge_vers {}", s.edge_versions).ok();
1146 writeln!(out, "vectors {}", s.vectors).ok();
1147 writeln!(out, "hnsw_idx {}", s.hnsw_indexed).ok();
1148 writeln!(out, "next_fact {}", s.next_fact).ok();
1149 writeln!(out, "next_edge {}", s.next_edge).ok();
1150 writeln!(out, "pool_bytes {}", s.pool_bytes).ok();
1151 writeln!(
1154 out,
1155 "shards facts {} entities {} edges {} temporal {} postings {}",
1156 s.shards.facts, s.shards.entities, s.shards.edges, s.shards.temporal, s.shards.postings,
1157 )
1158 .ok();
1159 }
1160}
1161
1162fn write_export_line(out: &mut impl Write, f: &ExportedFact) {
1169 writeln!(
1170 out,
1171 "{}",
1172 json!({
1173 "kind": "fact",
1174 "id": f.id,
1175 "text": f.text,
1176 "entity": f.entity,
1177 "tags": f.tags,
1178 "metadata": f.metadata,
1179 "recorded_at": f.recorded_at,
1180 "valid_from": f.valid_from,
1181 })
1182 )
1183 .ok();
1184}
1185
1186fn write_export_edge(out: &mut impl Write, src: &str, rel: &str, dst: &str, provenance: FactId) {
1190 writeln!(
1191 out,
1192 "{}",
1193 json!({
1194 "kind": "edge",
1195 "src": src,
1196 "rel": rel,
1197 "dst": dst,
1198 "provenance": (provenance != FactId::NONE).then_some(provenance.0),
1199 })
1200 )
1201 .ok();
1202}
1203
1204#[cfg(test)]
1207fn render_export(facts: &[ExportedFact], _json: bool, out: &mut impl Write) {
1208 for f in facts {
1209 write_export_line(out, f);
1210 }
1211}
1212
1213fn do_import(
1220 db: &Database,
1221 now: u64,
1222 file: &std::path::Path,
1223 batch_size: usize,
1224 _out: &mut impl Write,
1225) -> Result<ImportReport, CliError> {
1226 let f = std::fs::File::open(file)
1227 .map_err(|e| CliError::Usage(format!("reading {}: {e}", file.display())))?;
1228 let reader = io::BufReader::new(f);
1229 let mut count = 0usize;
1230 let mut batch: Vec<ParsedFact> = Vec::with_capacity(batch_size);
1231 let mut remap: BTreeMap<u32, FactId> = BTreeMap::new();
1235 let mut edges = 0usize;
1236
1237 for (i, line) in reader.lines().enumerate() {
1238 let line = line.map_err(|e| CliError::Usage(format!("line {}: {e}", i + 1)))?;
1239 let line = line.trim();
1240 if line.is_empty() {
1241 continue;
1242 }
1243 match parse_import_line(line, i + 1)? {
1244 ImportLine::Fact(fact) => {
1245 batch.push(fact);
1246 if batch.len() >= batch_size {
1247 count += flush_import_batch(db, now, &batch, &mut remap)?;
1248 batch.clear();
1249 }
1250 }
1251 ImportLine::Edge(edge) => {
1252 if !batch.is_empty() {
1257 count += flush_import_batch(db, now, &batch, &mut remap)?;
1258 batch.clear();
1259 }
1260 db.link(LinkInput {
1261 now,
1262 src: &edge.src,
1263 rel: &edge.rel,
1264 dst: &edge.dst,
1265 provenance: edge.provenance.and_then(|old| remap.get(&old).copied()),
1269 })?;
1270 edges += 1;
1271 }
1272 }
1273 }
1274 count += flush_import_batch(db, now, &batch, &mut remap)?;
1275 Ok(ImportReport {
1276 facts: count,
1277 edges,
1278 })
1279}
1280
1281#[derive(Debug, PartialEq, Eq)]
1285struct ImportReport {
1286 facts: usize,
1287 edges: usize,
1288}
1289
1290enum ImportLine {
1292 Fact(ParsedFact),
1293 Edge(ParsedEdge),
1294}
1295
1296struct ParsedEdge {
1299 src: String,
1300 rel: String,
1301 dst: String,
1302 provenance: Option<u32>,
1303}
1304
1305struct ParsedFact {
1309 id: Option<u32>,
1310 text: String,
1311 entity: Option<String>,
1312 tags: Vec<String>,
1313 metadata: Vec<(String, String)>,
1314 valid_from: Option<u64>,
1315}
1316
1317fn parse_import_line(line: &str, lineno: usize) -> Result<ImportLine, CliError> {
1322 let v: serde_json::Value =
1323 serde_json::from_str(line).map_err(|e| CliError::Usage(format!("line {lineno}: {e}")))?;
1324 match v["kind"].as_str() {
1325 None | Some("fact") => parse_import_fact(&v, lineno).map(ImportLine::Fact),
1326 Some("edge") => parse_import_edge(&v, lineno).map(ImportLine::Edge),
1327 Some(other) => Err(CliError::Usage(format!(
1328 "line {lineno}: unknown kind \"{other}\" (expected \"fact\" or \"edge\")"
1329 ))),
1330 }
1331}
1332
1333fn parse_import_edge(v: &serde_json::Value, lineno: usize) -> Result<ParsedEdge, CliError> {
1336 let field = |key: &str| -> Result<String, CliError> {
1337 v[key]
1338 .as_str()
1339 .map(String::from)
1340 .ok_or_else(|| CliError::Usage(format!("line {lineno}: edge missing string \"{key}\"")))
1341 };
1342 Ok(ParsedEdge {
1343 src: field("src")?,
1344 rel: field("rel")?,
1345 dst: field("dst")?,
1346 provenance: v["provenance"].as_u64().and_then(|n| u32::try_from(n).ok()),
1347 })
1348}
1349
1350fn parse_import_fact(v: &serde_json::Value, lineno: usize) -> Result<ParsedFact, CliError> {
1352 let text = v["text"]
1353 .as_str()
1354 .ok_or_else(|| CliError::Usage(format!("line {lineno}: missing string \"text\"")))?
1355 .to_string();
1356 let entity = v["entity"].as_str().map(String::from);
1357 let tags = v["tags"]
1358 .as_array()
1359 .map(|a| {
1360 a.iter()
1361 .filter_map(|t| t.as_str().map(String::from))
1362 .collect()
1363 })
1364 .unwrap_or_default();
1365 let metadata = v["metadata"]
1368 .as_object()
1369 .map(|m| {
1370 m.iter()
1371 .filter_map(|(k, val)| val.as_str().map(|s| (k.clone(), s.to_string())))
1372 .collect::<BTreeMap<_, _>>()
1373 .into_iter()
1374 .collect()
1375 })
1376 .unwrap_or_default();
1377 let valid_from = v["valid_from"].as_u64();
1378 Ok(ParsedFact {
1379 id: v["id"].as_u64().and_then(|n| u32::try_from(n).ok()),
1380 text,
1381 entity,
1382 tags,
1383 metadata,
1384 valid_from,
1385 })
1386}
1387
1388fn flush_import_batch(
1391 db: &Database,
1392 now: u64,
1393 batch: &[ParsedFact],
1394 remap: &mut BTreeMap<u32, FactId>,
1395) -> Result<usize, CliError> {
1396 if batch.is_empty() {
1397 return Ok(0);
1398 }
1399 let tag_refs: Vec<Vec<&str>> = batch
1402 .iter()
1403 .map(|p| p.tags.iter().map(String::as_str).collect())
1404 .collect();
1405 let meta_refs: Vec<Vec<(&str, &str)>> = batch
1406 .iter()
1407 .map(|p| {
1408 p.metadata
1409 .iter()
1410 .map(|(k, v)| (k.as_str(), v.as_str()))
1411 .collect()
1412 })
1413 .collect();
1414 let inputs: Vec<RememberInput> = batch
1415 .iter()
1416 .zip(&tag_refs)
1417 .zip(&meta_refs)
1418 .map(|((p, tags), meta)| RememberInput {
1419 entity: p.entity.as_deref(),
1420 tags,
1421 metadata: (!meta.is_empty()).then_some(meta.as_slice()),
1422 valid_from: p.valid_from,
1423 ..RememberInput::text(now, &p.text)
1424 })
1425 .collect();
1426 let outcomes = db.remember_many(inputs)?;
1429 for (parsed, outcome) in batch.iter().zip(&outcomes) {
1430 if let Some(old) = parsed.id {
1431 remap.insert(old, outcome.id);
1432 }
1433 }
1434 Ok(batch.len())
1435}
1436
1437#[allow(clippy::too_many_arguments)]
1439fn do_remember(
1440 db: &Database,
1441 now: u64,
1442 text: &str,
1443 entity: &Option<String>,
1444 tags: &[String],
1445 links: &[String],
1446 meta: &[String],
1447 valid_from: Option<u64>,
1448 vector: &[f32],
1449 revise: Option<FactId>,
1450) -> Result<RememberOutcome, CliError> {
1451 let tag_refs: Vec<&str> = tags.iter().map(String::as_str).collect();
1452 let link_pairs = parse_links(links)?;
1453 let link_refs: Vec<(&str, &str)> = link_pairs
1454 .iter()
1455 .map(|(r, e)| (r.as_str(), e.as_str()))
1456 .collect();
1457 let meta_map = parse_meta(meta)?;
1461 let meta_refs: Vec<(&str, &str)> = meta_map
1462 .iter()
1463 .map(|(k, v)| (k.as_str(), v.as_str()))
1464 .collect();
1465 let input = RememberInput {
1466 entity: entity.as_deref(),
1467 tags: &tag_refs,
1468 links: &link_refs,
1469 metadata: (!meta_refs.is_empty()).then_some(meta_refs.as_slice()),
1470 valid_from,
1471 vector: (!vector.is_empty()).then_some(vector),
1475 ..RememberInput::text(now, text)
1476 };
1477 match revise {
1478 Some(target) => Ok(db.revise(target, input)?),
1479 None => Ok(db.remember(input)?),
1480 }
1481}
1482
1483fn parse_meta(meta: &[String]) -> Result<BTreeMap<String, String>, CliError> {
1486 let mut map = BTreeMap::new();
1487 for s in meta {
1488 let (k, v) = s
1489 .split_once('=')
1490 .filter(|(k, _)| !k.is_empty())
1491 .ok_or_else(|| CliError::Usage(format!("bad --meta `{s}` — expected KEY=VALUE")))?;
1492 map.insert(k.to_string(), v.to_string());
1493 }
1494 Ok(map)
1495}
1496
1497fn parse_links(links: &[String]) -> Result<Vec<(String, String)>, CliError> {
1499 links
1500 .iter()
1501 .map(|s| {
1502 s.split_once(':')
1503 .filter(|(r, e)| !r.is_empty() && !e.is_empty())
1504 .map(|(r, e)| (r.to_string(), e.to_string()))
1505 .ok_or_else(|| CliError::Usage(format!("bad --link `{s}` — expected REL:ENTITY")))
1506 })
1507 .collect()
1508}
1509
1510fn render_remember(outcome: &RememberOutcome, json: bool, out: &mut impl Write) {
1512 if json {
1513 let similar: Vec<_> = outcome
1514 .similar
1515 .iter()
1516 .map(|s| json!({ "id": s.id.0, "score": s.score, "reason": format!("{:?}", s.reason) }))
1517 .collect();
1518 writeln!(
1519 out,
1520 "{}",
1521 json!({
1522 "id": outcome.id.0,
1523 "entity": outcome.entity.map(|e| e.0),
1524 "similar": similar,
1525 })
1526 )
1527 .ok();
1528 } else {
1529 writeln!(out, "remembered fact {}", outcome.id.0).ok();
1530 for s in &outcome.similar {
1531 writeln!(
1532 out,
1533 " ~ similar to fact {} ({:?}, {:.2})",
1534 s.id.0, s.reason, s.score
1535 )
1536 .ok();
1537 }
1538 }
1539}
1540
1541fn open_or(valid_to: u64) -> Option<u64> {
1543 (valid_to != VALID_TO_OPEN).then_some(valid_to)
1544}
1545
1546#[cfg(test)]
1547mod tests {
1548 use plugmem_host::Config;
1549
1550 use super::*;
1551
1552 struct StubEmbedder;
1554 impl plugmem_host::Embedder for StubEmbedder {
1555 fn dim(&self) -> usize {
1556 3
1557 }
1558 fn embed(&self, texts: &[&str]) -> Result<Vec<Vec<f32>>, HostError> {
1559 Ok(texts.iter().map(|_| vec![0.1, 0.2, 0.3]).collect())
1560 }
1561 }
1562
1563 fn recall_cmd(query: Option<&str>) -> Command {
1564 Command::Recall {
1565 query: query.map(str::to_owned),
1566 tags: vec![],
1567 entities: vec![],
1568 as_of: None,
1569 range: None,
1570 k: 0,
1571 closed: false,
1572 token_budget: None,
1573 ef: None,
1574 graph_depth: None,
1575 vector: Vec::new(),
1576 }
1577 }
1578
1579 fn settings_with(embedder: Option<Box<dyn plugmem_host::Embedder>>) -> Settings {
1580 Settings {
1581 database_path: None,
1582 config: Config::default(),
1583 embedder,
1584 snapshot_every_ops: None,
1585 snapshot_journal_bytes: None,
1586 maintain_every_forgets: None,
1587 fsync: None,
1588 workspace: plugmem_host::WorkspaceSettings {
1589 dir: None,
1590 limits: plugmem_host::WorkspaceLimits::default(),
1591 },
1592 warnings: Vec::new(),
1593 }
1594 }
1595
1596 #[test]
1597 fn embed_recall_query_embeds_recall_text_only_when_an_embedder_is_set() {
1598 let mut with = settings_with(Some(Box::new(StubEmbedder)));
1600 assert_eq!(
1601 embed_recall_query(&mut with, &recall_cmd(Some("tokio"))).unwrap(),
1602 Some(vec![0.1, 0.2, 0.3])
1603 );
1604
1605 let mut without = settings_with(None);
1607 assert_eq!(
1608 embed_recall_query(&mut without, &recall_cmd(Some("tokio"))).unwrap(),
1609 None
1610 );
1611
1612 let mut with_empty = settings_with(Some(Box::new(StubEmbedder)));
1614 assert_eq!(
1615 embed_recall_query(&mut with_empty, &recall_cmd(None)).unwrap(),
1616 None
1617 );
1618
1619 let mut with_stats = settings_with(Some(Box::new(StubEmbedder)));
1621 assert_eq!(
1622 embed_recall_query(&mut with_stats, &Command::Stats).unwrap(),
1623 None
1624 );
1625 }
1626
1627 #[test]
1628 fn split_line_honors_quotes_and_whitespace() {
1629 assert_eq!(split_line("remember hello"), ["remember", "hello"]);
1630 assert_eq!(
1631 split_line(r#"remember "two words" --tag x"#),
1632 ["remember", "two words", "--tag", "x"]
1633 );
1634 assert_eq!(split_line(" recall 'a b' "), ["recall", "a b"]);
1635 assert_eq!(split_line(""), Vec::<String>::new());
1636 assert_eq!(split_line(r#"remember """#), ["remember", ""]);
1638 }
1639
1640 #[test]
1641 fn repl_runs_over_one_handle_and_checkpoints_on_exit() {
1642 let (db, tmp) = TempDb::open();
1643 let path = tmp.0.join("m.plugmem");
1644 drop(db); let settings = settings_with(None);
1647 let script = b"remember \"hello tokio world\"\nrecall tokio\nrevise 0 \"goodbye tokio\"\nbadcmd\nexit\n";
1649 let mut out = Vec::new();
1650 let code = run_repl(&path, settings, false, &script[..], &mut out);
1651 let text = String::from_utf8(out).unwrap();
1652
1653 assert_eq!(code, 0);
1654 assert!(text.contains("remembered fact 0"), "{text}");
1655 assert!(text.contains("tokio"), "{text}");
1656 assert!(text.contains("unrecognized subcommand"), "{text}");
1658
1659 let ro = Database::open_readonly(&path, Config::default()).unwrap();
1663 assert_eq!(ro.stats().facts, 2, "original + successor after the revise");
1664 }
1665
1666 #[test]
1667 fn read_only_repl_observes_a_writer_reports_freshness_and_refuses_writes() {
1668 let (db, tmp) = TempDb::open();
1669 let path = tmp.0.join("m.plugmem");
1670 let mut sink = Vec::new();
1673 execute(
1674 &db,
1675 &remember("seed fact tokio", None, &[]),
1676 false,
1677 1_000,
1678 &mut sink,
1679 )
1680 .unwrap();
1681 db.checkpoint(1_001).unwrap();
1682
1683 let settings = settings_with(None);
1684 let script = b"generation\nstats\nrefresh\nremember \"nope\"\nexit\n";
1686 let mut out = Vec::new();
1687 let code = run_repl_ro(&path, settings, false, &script[..], &mut out);
1688 let text = String::from_utf8(out).unwrap();
1689
1690 assert_eq!(code, 0);
1691 assert!(text.contains("generation 1"), "generation verb: {text}");
1692 assert!(text.contains("fact"), "stats ran: {text}");
1693 assert!(
1696 text.contains("already current → generation 1"),
1697 "refresh no-op: {text}"
1698 );
1699 assert!(text.contains("read-only session"), "write refused: {text}");
1701
1702 assert_eq!(db.stats().facts, 1);
1705 }
1706
1707 #[test]
1708 fn read_only_repl_refresh_advances_after_the_writer_checkpoints() {
1709 let (db, tmp) = TempDb::open();
1710 let path = tmp.0.join("m.plugmem");
1711 let mut sink = Vec::new();
1712 execute(&db, &remember("first", None, &[]), false, 1_000, &mut sink).unwrap();
1713 db.checkpoint(1_001).unwrap();
1714
1715 struct HookOnFirstRead<'a> {
1719 script: std::io::Cursor<&'a [u8]>,
1720 db: &'a Database,
1721 fired: bool,
1722 }
1723 impl std::io::Read for HookOnFirstRead<'_> {
1724 fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
1725 if !self.fired {
1726 self.fired = true;
1727 let mut s = Vec::new();
1730 execute(
1731 self.db,
1732 &remember("second", None, &[]),
1733 false,
1734 2_000,
1735 &mut s,
1736 )
1737 .unwrap();
1738 self.db.checkpoint(2_001).unwrap();
1739 }
1740 self.script.read(buf)
1741 }
1742 }
1743 let reader = std::io::BufReader::new(HookOnFirstRead {
1744 script: std::io::Cursor::new(b"refresh\nstats\nexit\n" as &[u8]),
1745 db: &db,
1746 fired: false,
1747 });
1748
1749 let mut out = Vec::new();
1750 let code = run_repl_ro(&path, settings_with(None), false, reader, &mut out);
1751 let text = String::from_utf8(out).unwrap();
1752
1753 assert_eq!(code, 0);
1754 assert!(text.contains("refreshed → generation 2"), "advance: {text}");
1756 assert!(text.contains("fact"), "stats after refresh: {text}");
1758 assert_eq!(db.stats().facts, 2);
1759 }
1760
1761 #[test]
1762 fn read_only_repl_freshness_verbs_emit_json() {
1763 let (db, tmp) = TempDb::open();
1764 let path = tmp.0.join("m.plugmem");
1765 let mut sink = Vec::new();
1766 execute(&db, &remember("j", None, &[]), false, 1_000, &mut sink).unwrap();
1767 db.checkpoint(1_001).unwrap();
1768
1769 let script = b"generation\nrefresh\nexit\n";
1770 let mut out = Vec::new();
1771 let code = run_repl_ro(&path, settings_with(None), true, &script[..], &mut out);
1772 let text = String::from_utf8(out).unwrap();
1773
1774 assert_eq!(code, 0);
1775 assert!(
1776 text.contains(r#""generation":1"#),
1777 "generation json: {text}"
1778 );
1779 assert!(text.contains(r#""advanced":false"#), "refresh json: {text}");
1780 }
1781
1782 struct TempDb(PathBuf);
1784 impl TempDb {
1785 fn open() -> (Database, Self) {
1786 let dir = std::env::temp_dir().join(format!(
1787 "plugmem-cli-{}-{}",
1788 std::process::id(),
1789 now_ms_unique()
1790 ));
1791 std::fs::create_dir_all(&dir).unwrap();
1792 let path = dir.join("m.plugmem");
1793 let (db, _) = Database::open(&path, Config::default()).unwrap();
1794 (db, TempDb(dir))
1795 }
1796 }
1797 impl Drop for TempDb {
1798 fn drop(&mut self) {
1799 let _ = std::fs::remove_dir_all(&self.0);
1800 }
1801 }
1802
1803 fn now_ms_unique() -> String {
1806 use std::sync::atomic::{AtomicU64, Ordering};
1807 static N: AtomicU64 = AtomicU64::new(0);
1808 format!("{}-{}", now_ms(), N.fetch_add(1, Ordering::Relaxed))
1809 }
1810
1811 fn run_cmd(db: &Database, cmd: &Command, json: bool, now: u64) -> (u8, String) {
1812 let mut buf = Vec::new();
1813 let code = execute(db, cmd, json, now, &mut buf).expect("execute");
1814 (code, String::from_utf8(buf).unwrap())
1815 }
1816
1817 fn remember(text: &str, entity: Option<&str>, tags: &[&str]) -> Command {
1818 Command::Remember {
1819 text: text.into(),
1820 entity: entity.map(Into::into),
1821 tags: tags.iter().map(|t| (*t).into()).collect(),
1822 links: Vec::new(),
1823 meta: Vec::new(),
1824 valid_from: None,
1825 vector: Vec::new(),
1826 }
1827 }
1828
1829 fn remember_with_meta(text: &str, meta: &[&str]) -> Command {
1830 Command::Remember {
1831 text: text.into(),
1832 entity: None,
1833 tags: Vec::new(),
1834 links: Vec::new(),
1835 meta: meta.iter().map(|m| (*m).into()).collect(),
1836 valid_from: None,
1837 vector: Vec::new(),
1838 }
1839 }
1840
1841 #[test]
1842 fn meta_flag_renders_sorted_in_show_and_export_and_rejects_bad_input() {
1843 let (db, _t) = TempDb::open();
1844 let cmd = remember_with_meta("a scan", &["uri=s3://b/x", "page=2", "page=3"]);
1846 assert_eq!(run_cmd(&db, &cmd, false, 1_000).0, 0);
1847
1848 let (_, human) = run_cmd(&db, &Command::Show { id: 0 }, false, 2_000);
1850 assert!(
1851 human.contains("metadata page=3, uri=s3://b/x"),
1852 "{human}"
1853 );
1854 let (_, jshow) = run_cmd(&db, &Command::Show { id: 0 }, true, 2_000);
1856 let v: serde_json::Value = serde_json::from_str(&jshow).unwrap();
1857 assert_eq!(v["metadata"]["page"], "3");
1858 assert_eq!(v["metadata"]["uri"], "s3://b/x");
1859
1860 let (_, exp) = run_cmd(&db, &Command::Export, false, 2_000);
1862 let line: serde_json::Value = serde_json::from_str(exp.lines().next().unwrap()).unwrap();
1863 assert_eq!(line["metadata"]["uri"], "s3://b/x");
1864
1865 assert!(matches!(
1867 parse_meta(&["noequals".to_string()]),
1868 Err(CliError::Usage(_))
1869 ));
1870 assert!(parse_meta(&["=noKey".to_string()]).is_err());
1871 }
1872
1873 #[test]
1874 fn remember_then_recall_human_and_json() {
1875 let (db, _t) = TempDb::open();
1876 let (code, out) = run_cmd(
1877 &db,
1878 &remember("prefers tokio", Some("user"), &["pref"]),
1879 false,
1880 1_000,
1881 );
1882 assert_eq!(code, 0);
1883 assert!(out.starts_with("remembered fact 0"), "{out}");
1884
1885 let recall = Command::Recall {
1887 query: Some("tokio".into()),
1888 tags: Vec::new(),
1889 entities: Vec::new(),
1890 as_of: None,
1891 range: None,
1892 k: 0,
1893 closed: false,
1894 token_budget: None,
1895 ef: None,
1896 graph_depth: None,
1897 vector: Vec::new(),
1898 };
1899 let (code, out) = run_cmd(&db, &recall, false, 2_000);
1900 assert_eq!(code, 0);
1901 assert!(out.contains("tokio"), "{out}");
1902
1903 let (code, out) = run_cmd(&db, &recall, true, 2_000);
1905 assert_eq!(code, 0);
1906 let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
1907 assert!(!v["facts"].as_array().unwrap().is_empty(), "{out}");
1908 }
1909
1910 #[test]
1911 fn recall_empty_is_ok_with_a_note() {
1912 let (db, _t) = TempDb::open();
1913 let recall = Command::Recall {
1914 query: Some("nothing here".into()),
1915 tags: Vec::new(),
1916 entities: Vec::new(),
1917 as_of: None,
1918 range: None,
1919 k: 0,
1920 closed: false,
1921 token_budget: None,
1922 ef: None,
1923 graph_depth: None,
1924 vector: Vec::new(),
1925 };
1926 let (code, out) = run_cmd(&db, &recall, false, 1_000);
1927 assert_eq!(code, 0);
1928 assert!(out.contains("nothing recalled"), "{out}");
1929 }
1930
1931 #[test]
1932 fn revise_closes_the_predecessor_and_conflict_is_surfaced() {
1933 let (db, _t) = TempDb::open();
1934 run_cmd(
1935 &db,
1936 &remember("lives in Moscow", Some("user"), &[]),
1937 false,
1938 1_000,
1939 );
1940 let (_, out) = run_cmd(
1942 &db,
1943 &remember("lives in Moscow now", Some("user"), &[]),
1944 false,
1945 1_500,
1946 );
1947 assert!(out.contains("similar to fact"), "{out}");
1948
1949 let revise = Command::Revise {
1950 id: 0,
1951 text: "lives in Berlin".into(),
1952 entity: Some("user".into()),
1953 tags: Vec::new(),
1954 links: Vec::new(),
1955 meta: Vec::new(),
1956 valid_from: None,
1957 vector: Vec::new(),
1958 };
1959 let (code, out) = run_cmd(&db, &revise, false, 2_000);
1960 assert_eq!(code, 0);
1961 assert!(out.starts_with("remembered fact"), "{out}");
1962 }
1963
1964 #[test]
1965 fn show_found_and_missing() {
1966 let (db, _t) = TempDb::open();
1967 run_cmd(&db, &remember("a note", None, &[]), false, 1_000);
1968
1969 let (code, out) = run_cmd(&db, &Command::Show { id: 0 }, false, 2_000);
1970 assert_eq!(code, 0);
1971 assert!(
1972 out.contains("a note") && out.contains("recorded_at 1000"),
1973 "{out}"
1974 );
1975
1976 let (code, out) = run_cmd(&db, &Command::Show { id: 999 }, false, 2_000);
1977 assert_eq!(code, 1, "missing id is a soft miss");
1978 assert!(out.contains("not found"), "{out}");
1979
1980 let (_, out) = run_cmd(&db, &Command::Show { id: 0 }, true, 2_000);
1982 let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
1983 assert_eq!(v["text"], "a note");
1984 assert_eq!(v["valid_to"], serde_json::Value::Null); }
1986
1987 #[test]
1988 fn forget_then_maintain_purges() {
1989 let (db, _t) = TempDb::open();
1990 run_cmd(&db, &remember("temp", None, &[]), false, 1_000);
1991
1992 let (code, out) = run_cmd(&db, &Command::Forget { id: 0 }, false, 2_000);
1993 assert_eq!(code, 0);
1994 assert!(out.contains("forgot fact 0"), "{out}");
1995 let (_, out) = run_cmd(&db, &Command::Forget { id: 0 }, false, 2_100);
1997 assert!(out.contains("already gone"), "{out}");
1998
1999 let (code, out) = run_cmd(
2000 &db,
2001 &Command::Maintain {
2002 mode: MaintainMode::Auto,
2003 },
2004 false,
2005 3_000,
2006 );
2007 assert_eq!(code, 0);
2008 assert!(out.contains("purged 1"), "{out}");
2009 }
2010
2011 #[test]
2012 fn link_and_stats_and_json() {
2013 let (db, _t) = TempDb::open();
2014 run_cmd(
2015 &db,
2016 &remember("uses tokio", Some("plugmem"), &[]),
2017 false,
2018 1_000,
2019 );
2020 let link = Command::Link {
2021 src: "plugmem".into(),
2022 rel: "depends_on".into(),
2023 dst: "tokio".into(),
2024 provenance: None,
2025 };
2026 let (code, out) = run_cmd(&db, &link, false, 2_000);
2027 assert_eq!(code, 0);
2028 assert!(out.contains("plugmem -depends_on-> tokio"), "{out}");
2029 let unlink = Command::Unlink {
2030 src: "plugmem".into(),
2031 rel: "depends_on".into(),
2032 dst: "tokio".into(),
2033 };
2034 let (code, out) = run_cmd(&db, &unlink, false, 2_500);
2035 assert_eq!(code, 0);
2036 assert!(
2037 out.contains("unlinked plugmem -depends_on-> tokio"),
2038 "{out}"
2039 );
2040
2041 let (code, out) = run_cmd(&db, &Command::Stats, true, 3_000);
2042 assert_eq!(code, 0);
2043 let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2044 assert_eq!(v["facts"], 1);
2045 assert_eq!(v["edges"], 0);
2046 assert_eq!(v["edge_versions"], 1);
2047 }
2048
2049 #[test]
2050 fn bad_link_is_a_usage_error() {
2051 let (db, _t) = TempDb::open();
2052 let cmd = Command::Remember {
2053 text: "x".into(),
2054 entity: Some("user".into()),
2055 tags: Vec::new(),
2056 links: vec!["not-a-pair".into()],
2057 meta: Vec::new(),
2058 valid_from: None,
2059 vector: Vec::new(),
2060 };
2061 let mut buf = Vec::new();
2062 let err = execute(&db, &cmd, false, 1_000, &mut buf).unwrap_err();
2063 assert!(matches!(err, CliError::Usage(_)));
2064 }
2065
2066 #[test]
2067 fn as_of_time_travel_via_recall() {
2068 let (db, _t) = TempDb::open();
2069 run_cmd(
2070 &db,
2071 &remember("lives in Moscow", Some("user"), &[]),
2072 false,
2073 1_000,
2074 );
2075 let revise = Command::Revise {
2076 id: 0,
2077 text: "lives in Berlin".into(),
2078 entity: Some("user".into()),
2079 tags: Vec::new(),
2080 links: Vec::new(),
2081 meta: Vec::new(),
2082 valid_from: None,
2083 vector: Vec::new(),
2084 };
2085 run_cmd(&db, &revise, false, 2_000);
2086
2087 let as_of = Command::Recall {
2088 query: Some("lives".into()),
2089 tags: Vec::new(),
2090 entities: vec!["user".into()],
2091 as_of: Some(1_500),
2092 range: None,
2093 k: 0,
2094 closed: false,
2095 token_budget: None,
2096 ef: None,
2097 graph_depth: None,
2098 vector: Vec::new(),
2099 };
2100 let (_, out) = run_cmd(&db, &as_of, false, 3_000);
2101 assert!(out.contains("Moscow"), "as-of 1500 → Moscow: {out}");
2102 }
2103
2104 #[test]
2105 fn every_command_has_a_json_shape() {
2106 let (db, _t) = TempDb::open();
2107 let (_, out) = run_cmd(
2109 &db,
2110 &remember("uses tokio", Some("plugmem"), &["pref"]),
2111 true,
2112 1_000,
2113 );
2114 let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2115 assert_eq!(v["id"], 0);
2116 assert!(v["similar"].is_array());
2117
2118 let revise = Command::Revise {
2120 id: 0,
2121 text: "uses tokio now".into(),
2122 entity: Some("plugmem".into()),
2123 tags: Vec::new(),
2124 links: Vec::new(),
2125 meta: Vec::new(),
2126 valid_from: None,
2127 vector: Vec::new(),
2128 };
2129 let (_, out) = run_cmd(&db, &revise, true, 1_500);
2130 assert!(serde_json::from_str::<serde_json::Value>(out.trim()).is_ok());
2131
2132 let link = Command::Link {
2134 src: "plugmem".into(),
2135 rel: "depends_on".into(),
2136 dst: "tokio".into(),
2137 provenance: None,
2138 };
2139 let (_, out) = run_cmd(&db, &link, true, 2_000);
2140 let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2141 assert_eq!(v["rel"], "depends_on");
2142 let unlink = Command::Unlink {
2143 src: "plugmem".into(),
2144 rel: "depends_on".into(),
2145 dst: "tokio".into(),
2146 };
2147 let (_, out) = run_cmd(&db, &unlink, true, 2_100);
2148 let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2149 assert_eq!(v["unlinked"], true);
2150
2151 let (_, out) = run_cmd(&db, &Command::Forget { id: 1 }, true, 2_500);
2153 let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2154 assert_eq!(v["forgotten"], true);
2155 let (_, out) = run_cmd(
2156 &db,
2157 &Command::Maintain {
2158 mode: MaintainMode::Auto,
2159 },
2160 true,
2161 3_000,
2162 );
2163 let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2164 assert!(v["purged"].as_u64().unwrap() >= 1);
2165
2166 let (code, out) = run_cmd(&db, &Command::Show { id: 999 }, true, 3_500);
2168 assert_eq!(code, 1);
2169 let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2170 assert_eq!(v["found"], false);
2171
2172 let recall = Command::Recall {
2174 query: None,
2175 tags: Vec::new(),
2176 entities: vec!["plugmem".into()],
2177 as_of: None,
2178 range: Some(vec![0, 10_000]),
2179 k: 4,
2180 closed: true,
2181 token_budget: None,
2182 ef: None,
2183 graph_depth: None,
2184 vector: Vec::new(),
2185 };
2186 let (_, out) = run_cmd(&db, &recall, true, 4_000);
2187 assert!(serde_json::from_str::<serde_json::Value>(out.trim()).is_ok());
2188 }
2189
2190 #[test]
2191 fn graph_depth_is_a_per_call_flag_over_the_configured_default() {
2192 let (db, _t) = TempDb::open();
2195 for (i, (entity, next)) in [("a", "b"), ("b", "c"), ("c", "d"), ("d", "e")]
2196 .iter()
2197 .enumerate()
2198 {
2199 let cmd = Command::Remember {
2200 text: format!("fact on {entity}"),
2201 entity: Some((*entity).into()),
2202 tags: Vec::new(),
2203 links: vec![format!("leads_to:{next}")],
2204 meta: Vec::new(),
2205 valid_from: None,
2206 vector: Vec::new(),
2207 };
2208 run_cmd(&db, &cmd, false, 1_000 + i as u64);
2209 }
2210
2211 let reached = |depth: Option<u32>| {
2212 let recall = Command::Recall {
2213 query: None,
2214 tags: Vec::new(),
2215 entities: vec!["a".into()],
2216 as_of: None,
2217 range: None,
2218 k: 64,
2219 closed: false,
2220 token_budget: Some(4096),
2221 ef: None,
2222 graph_depth: depth,
2223 vector: Vec::new(),
2224 };
2225 let (_, out) = run_cmd(&db, &recall, false, 5_000);
2226 out.lines().filter(|l| l.starts_with("- [f")).count()
2227 };
2228
2229 assert_eq!(reached(None), 3, "the configured default is 2 hops");
2230 assert_eq!(reached(Some(0)), 1, "no expansion: the anchor's own fact");
2231 assert_eq!(reached(Some(1)), 2);
2232 assert_eq!(reached(Some(3)), 4);
2233 assert_eq!(reached(Some(99)), 4);
2236 assert_eq!(reached(Some(u32::MAX)), 4);
2237 }
2238
2239 #[test]
2240 fn stats_human_lists_the_counters() {
2241 let (db, _t) = TempDb::open();
2242 run_cmd(&db, &remember("a", None, &[]), false, 1_000);
2243 let (code, out) = run_cmd(&db, &Command::Stats, false, 2_000);
2244 assert_eq!(code, 0);
2245 assert!(out.contains("facts") && out.contains("pool_bytes"), "{out}");
2246 }
2247
2248 #[test]
2249 fn verify_command_renders_human_and_json() {
2250 let (db, _t) = TempDb::open();
2251 run_cmd(&db, &remember("clean", None, &[]), false, 1_000);
2252
2253 let (code, out) = run_cmd(&db, &Command::Verify, false, 2_000);
2254 assert_eq!(code, 0);
2255 assert_eq!(out.trim(), "integrity ok");
2256
2257 let (code, out) = run_cmd(&db, &Command::Verify, true, 2_100);
2258 assert_eq!(code, 0);
2259 let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2260 assert_eq!(v["ok"], true);
2261 }
2262
2263 #[test]
2264 fn show_json_of_a_revised_predecessor_is_closed() {
2265 let (db, _t) = TempDb::open();
2266 run_cmd(&db, &remember("v1", Some("e"), &[]), false, 1_000);
2267 let revise = Command::Revise {
2268 id: 0,
2269 text: "v2".into(),
2270 entity: Some("e".into()),
2271 tags: Vec::new(),
2272 links: Vec::new(),
2273 meta: Vec::new(),
2274 valid_from: None,
2275 vector: Vec::new(),
2276 };
2277 run_cmd(&db, &revise, false, 2_000);
2278 let (_, out) = run_cmd(&db, &Command::Show { id: 1 }, false, 3_000);
2280 assert!(out.contains("revises fact 0"), "{out}");
2281 }
2282
2283 #[test]
2284 fn resolve_db_path_prefers_the_flag() {
2285 let p = "/tmp/explicit.plugmem";
2286 assert_eq!(resolve_db_path(Some(p), None, None), PathBuf::from(p));
2287 let configured = std::path::Path::new("/tmp/configured.plugmem");
2288 assert_eq!(
2289 resolve_db_path(None, Some(configured), None),
2290 PathBuf::from(configured)
2291 );
2292 let _ = resolve_db_path(None, None, None);
2295 }
2296
2297 #[test]
2298 fn a_bare_name_is_a_memory_only_when_a_workspace_is_configured() {
2299 let root = PathBuf::from("/srv/bot");
2300
2301 assert_eq!(
2304 resolve_db_path(Some("work"), None, None),
2305 PathBuf::from("work")
2306 );
2307
2308 assert_eq!(
2310 resolve_db_path(Some("work"), None, Some(&root)),
2311 PathBuf::from("/srv/bot/db/work.plugmem")
2312 );
2313 for path in ["./work", "work.plugmem", "/srv/other.plugmem", "../up"] {
2316 assert_eq!(
2317 resolve_db_path(Some(path), None, Some(&root)),
2318 PathBuf::from(path),
2319 "{path}"
2320 );
2321 }
2322
2323 let configured = std::path::Path::new("work");
2325 assert_eq!(
2326 resolve_db_path(None, Some(configured), Some(&root)),
2327 PathBuf::from("work")
2328 );
2329 }
2330
2331 #[test]
2332 fn settings_help_runs_without_opening_a_database() {
2333 let cli = Cli::try_parse_from(["plugmem-cli", "help", "settings"]).unwrap();
2334 let mut output = Vec::new();
2335 assert_eq!(run_parsed(cli, &mut output), 0);
2336 let output = String::from_utf8(output).unwrap();
2337 assert!(output.contains("plugmem settings"));
2338 assert!(output.contains("[database]"));
2339 assert!(output.contains("path (path string"));
2340
2341 let cli = Cli::try_parse_from(["plugmem-cli", "--json", "help", "settings"]).unwrap();
2342 let mut output = Vec::new();
2343 assert_eq!(run_parsed(cli, &mut output), 0);
2344 let output: serde_json::Value = serde_json::from_slice(&output).unwrap();
2345 assert_eq!(output["topic"], "settings");
2346 assert!(output["config_path_precedence"].is_array());
2347 assert!(output["settings"].as_array().unwrap().len() > 10);
2348 }
2349
2350 #[test]
2351 fn run_parsed_opens_runs_and_reports() {
2352 let dir = std::env::temp_dir().join(format!(
2353 "plugmem-run-{}-{}",
2354 std::process::id(),
2355 now_ms_unique()
2356 ));
2357 std::fs::create_dir_all(&dir).unwrap();
2358 let path = dir.join("m.plugmem");
2359 let cli = Cli {
2360 db: Some(path.display().to_string()),
2361 workspace: None,
2362 config: None,
2363 json: false,
2364 command: Command::Stats,
2365 };
2366 let mut buf = Vec::new();
2367 let code = run_parsed(cli, &mut buf);
2368 assert_eq!(code, 0);
2369 assert!(String::from_utf8(buf).unwrap().contains("facts"));
2370 let _ = std::fs::remove_dir_all(&dir);
2371 }
2372
2373 #[test]
2374 fn recover_and_scrub_render_json_and_human_shapes() {
2375 let (db, tmp) = TempDb::open();
2376 let path = tmp.0.join("m.plugmem");
2377 run_cmd(&db, &remember("recoverable fact", None, &[]), false, 1_000);
2378 run_cmd(&db, &Command::Checkpoint, false, 2_000);
2379 drop(db);
2380
2381 let settings = settings_with(None);
2382 let mut out = Vec::new();
2383 assert_eq!(do_scrub(&path, &settings, true, &mut out), 0);
2384 let scrub: serde_json::Value = serde_json::from_slice(&out).unwrap();
2385 assert_eq!(scrub["ok"], true);
2386 assert!(scrub["bytes"].as_u64().unwrap() > 0);
2387 let mut out = Vec::new();
2388 assert_eq!(do_scrub(&path, &settings, false, &mut out), 0);
2389 let out = String::from_utf8(out).unwrap();
2390 assert!(out.contains("scrub ok:"), "{out}");
2391
2392 let json_dst = tmp.0.join("copy-json.plugmem");
2393 let mut out = Vec::new();
2394 assert_eq!(do_recover(&path, &json_dst, &settings, true, &mut out), 0);
2395 let recover: serde_json::Value = serde_json::from_slice(&out).unwrap();
2396 assert_eq!(recover["kept"], 1);
2397 assert_eq!(recover["dropped_text"], 0);
2398 assert_eq!(recover["dst"], json_dst.display().to_string());
2399
2400 let human_dst = tmp.0.join("copy-human.plugmem");
2401 let mut out = Vec::new();
2402 assert_eq!(do_recover(&path, &human_dst, &settings, false, &mut out), 0);
2403 let out = String::from_utf8(out).unwrap();
2404 assert!(out.contains("recovered to"), "{out}");
2405 assert!(out.contains("kept 1"), "{out}");
2406 }
2407
2408 #[test]
2409 fn readonly_dispatcher_renders_every_read_shape() {
2410 let (db, tmp) = TempDb::open();
2411 let path = tmp.0.join("m.plugmem");
2412 run_cmd(
2413 &db,
2414 &remember("readonly tokio fact", Some("plugmem"), &["pref"]),
2415 false,
2416 1_000,
2417 );
2418 run_cmd(&db, &Command::Checkpoint, false, 2_000);
2419 let ro = Database::open_readonly(&path, Config::default()).unwrap();
2420
2421 let mut out = Vec::new();
2422 assert_eq!(execute_ro(&ro, &Command::Stats, None, true, &mut out), 0);
2423 let stats: serde_json::Value = serde_json::from_slice(&out).unwrap();
2424 assert_eq!(stats["facts"], 1);
2425
2426 let mut out = Vec::new();
2427 assert_eq!(
2428 execute_ro(&ro, &Command::Show { id: 0 }, None, false, &mut out),
2429 0
2430 );
2431 let text = String::from_utf8(out).unwrap();
2432 assert!(text.contains("readonly tokio fact"), "{text}");
2433
2434 let mut out = Vec::new();
2435 assert_eq!(execute_ro(&ro, &Command::Export, None, false, &mut out), 0);
2436 let exported: serde_json::Value =
2437 serde_json::from_str(String::from_utf8(out).unwrap().lines().next().unwrap()).unwrap();
2438 assert_eq!(exported["text"], "readonly tokio fact");
2439
2440 let mut out = Vec::new();
2441 let recall = Command::Recall {
2442 query: Some("tokio".into()),
2443 tags: vec!["pref".into()],
2444 entities: vec!["plugmem".into()],
2445 as_of: None,
2446 range: None,
2447 k: 1,
2448 closed: false,
2449 token_budget: None,
2450 ef: None,
2451 graph_depth: None,
2452 vector: Vec::new(),
2453 };
2454 assert_eq!(execute_ro(&ro, &recall, None, false, &mut out), 0);
2455 let text = String::from_utf8(out).unwrap();
2456 assert!(text.contains("tokio"), "{text}");
2457
2458 let mut out = Vec::new();
2459 assert_eq!(execute_ro(&ro, &Command::Verify, None, true, &mut out), 0);
2460 let verify: serde_json::Value = serde_json::from_slice(&out).unwrap();
2461 assert_eq!(verify["ok"], true);
2462 }
2463
2464 #[test]
2465 fn run_parsed_on_a_locked_database_returns_one() {
2466 let (_held, dir) = {
2467 let dir = std::env::temp_dir().join(format!(
2468 "plugmem-lock-{}-{}",
2469 std::process::id(),
2470 now_ms_unique()
2471 ));
2472 std::fs::create_dir_all(&dir).unwrap();
2473 let path = dir.join("m.plugmem");
2474 (Database::open(&path, Config::default()).unwrap(), dir)
2475 };
2476 let cli = Cli {
2477 db: Some(dir.join("m.plugmem").display().to_string()),
2478 workspace: None,
2479 config: None,
2480 json: false,
2481 command: Command::Stats,
2482 };
2483 let mut buf = Vec::new();
2484 assert_eq!(run_parsed(cli, &mut buf), 1);
2485 let _ = std::fs::remove_dir_all(&dir);
2486 }
2487
2488 #[test]
2489 fn run_parsed_propagates_a_usage_error_as_two() {
2490 let dir = std::env::temp_dir().join(format!(
2491 "plugmem-usage-{}-{}",
2492 std::process::id(),
2493 now_ms_unique()
2494 ));
2495 std::fs::create_dir_all(&dir).unwrap();
2496 let cli = Cli {
2497 db: Some(dir.join("m.plugmem").display().to_string()),
2498 workspace: None,
2499 config: None,
2500 json: false,
2501 command: Command::Remember {
2502 text: "x".into(),
2503 entity: None,
2504 tags: Vec::new(),
2505 links: vec!["bad".into()],
2506 meta: Vec::new(),
2507 valid_from: None,
2508 vector: Vec::new(),
2509 },
2510 };
2511 let mut buf = Vec::new();
2512 assert_eq!(run_parsed(cli, &mut buf), 2);
2513 let _ = std::fs::remove_dir_all(&dir);
2514 }
2515
2516 struct Scratch(PathBuf);
2518 impl Scratch {
2519 fn new(tag: &str) -> Self {
2520 let dir = std::env::temp_dir().join(format!(
2521 "plugmem-cli-{tag}-{}-{}",
2522 std::process::id(),
2523 now_ms_unique()
2524 ));
2525 std::fs::create_dir_all(&dir).unwrap();
2526 Scratch(dir)
2527 }
2528 }
2529 impl Drop for Scratch {
2530 fn drop(&mut self) {
2531 let _ = std::fs::remove_dir_all(&self.0);
2532 }
2533 }
2534
2535 #[test]
2536 fn export_import_roundtrip_preserves_open_facts() {
2537 let (a, _ta) = TempDb::open();
2542 run_cmd(
2543 &a,
2544 &Command::Remember {
2545 text: "prefers tokio".into(),
2546 entity: Some("user".into()),
2547 tags: vec!["pref".into(), "lang".into()],
2548 links: Vec::new(),
2549 meta: vec!["uri=s3://b/x".into(), "src=chat".into()],
2550 valid_from: Some(500),
2551 vector: Vec::new(),
2552 },
2553 false,
2554 1_000,
2555 );
2556 run_cmd(
2557 &a,
2558 &remember("lives in Moscow", Some("user"), &[]),
2559 false,
2560 1_100,
2561 ); run_cmd(
2563 &a,
2564 &Command::Revise {
2565 id: 1,
2566 text: "lives in Berlin".into(),
2567 entity: Some("user".into()),
2568 tags: vec!["geo".into()],
2569 links: Vec::new(),
2570 meta: Vec::new(),
2571 valid_from: None,
2572 vector: Vec::new(),
2573 },
2574 false,
2575 1_200,
2576 ); run_cmd(&a, &remember("junk", None, &[]), false, 1_300); run_cmd(&a, &Command::Forget { id: 3 }, false, 1_400); run_cmd(
2580 &a,
2581 &remember("uses rust", Some("plugmem"), &["lang"]),
2582 false,
2583 1_500,
2584 ); let mut dump = Vec::new();
2588 render_export(&a.export(), false, &mut dump);
2589 let scratch = Scratch::new("roundtrip");
2590 let file = scratch.0.join("dump.jsonl");
2591 std::fs::write(&file, &dump).unwrap();
2592
2593 let (b, _tb) = TempDb::open();
2595 let n = do_import(&b, 9_000, &file, 128, &mut Vec::new()).unwrap();
2596
2597 let key = |f: &ExportedFact| {
2599 let mut tags = f.tags.clone();
2600 tags.sort();
2601 (f.text.clone(), f.entity.clone(), tags, f.valid_from)
2602 };
2603 let mut ak: Vec<_> = a.export().iter().map(key).collect();
2604 let mut bk: Vec<_> = b.export().iter().map(key).collect();
2605 ak.sort();
2606 bk.sort();
2607 assert_eq!(n.facts, ak.len());
2608 assert_eq!(
2609 ak, bk,
2610 "roundtrip must preserve text/entity/tags/valid_from"
2611 );
2612
2613 let b_open = b.export();
2616 assert!(b_open.iter().any(|f| f.text == "prefers tokio"
2617 && f.valid_from == 500
2618 && f.entity.as_deref() == Some("user")
2619 && f.tags == vec!["pref".to_string(), "lang".to_string()]
2620 && f.metadata.get("uri").map(String::as_str) == Some("s3://b/x")
2621 && f.metadata.get("src").map(String::as_str) == Some("chat")));
2622 assert!(b_open.iter().any(|f| f.text == "lives in Berlin"));
2623 assert!(b_open.iter().any(|f| f.text == "uses rust"));
2624 assert!(!b_open.iter().any(|f| f.text.contains("Moscow")));
2625 assert!(!b_open.iter().any(|f| f.text == "junk"));
2626 }
2627
2628 #[test]
2629 fn export_command_emits_jsonl_regardless_of_json_flag() {
2630 let (db, _t) = TempDb::open();
2631 run_cmd(&db, &remember("a fact", Some("e"), &["t"]), false, 1_000);
2632 for json in [false, true] {
2633 let (code, out) = run_cmd(&db, &Command::Export, json, 2_000);
2634 assert_eq!(code, 0);
2635 let v: serde_json::Value = serde_json::from_str(out.trim()).unwrap();
2636 assert_eq!(v["text"], "a fact");
2637 assert_eq!(v["entity"], "e");
2638 assert_eq!(v["tags"][0], "t");
2639 }
2640 }
2641
2642 #[test]
2643 fn import_command_counts_and_rejects_bad_lines() {
2644 let (db, _t) = TempDb::open();
2645 let scratch = Scratch::new("import");
2646 let good = scratch.0.join("in.jsonl");
2647 std::fs::write(
2648 &good,
2649 "{\"text\":\"from jsonl\",\"entity\":\"user\",\"tags\":[\"x\"],\"valid_from\":42}\n\n{\"text\":\"second\"}\n",
2650 )
2651 .unwrap();
2652 let n = do_import(&db, 9_000, &good, 1, &mut Vec::new()).unwrap();
2654 assert_eq!(n.facts, 2, "both facts imported, blank line skipped");
2655
2656 let bad = scratch.0.join("bad.jsonl");
2657 std::fs::write(&bad, "not json at all\n").unwrap();
2658 let err = do_import(&db, 9_000, &bad, 128, &mut Vec::new()).unwrap_err();
2659 assert!(matches!(err, CliError::Usage(_)));
2660 }
2661
2662 #[test]
2663 fn import_batch_size_does_not_change_the_result() {
2664 let scratch = Scratch::new("import-batch");
2667 let file = scratch.0.join("facts.jsonl");
2668 let mut jsonl = String::new();
2669 for i in 0..5 {
2670 jsonl.push_str(&format!("{{\"text\":\"fact number {i}\"}}\n"));
2671 }
2672 std::fs::write(&file, &jsonl).unwrap();
2673
2674 let (a, _ta) = TempDb::open();
2675 let (b, _tb) = TempDb::open();
2676 let na = do_import(&a, 9_000, &file, 1, &mut Vec::new()).unwrap();
2677 let nb = do_import(&b, 9_000, &file, 100, &mut Vec::new()).unwrap();
2678
2679 assert_eq!(na.facts, 5);
2680 assert_eq!(nb.facts, 5);
2681 let texts = |db: &Database| {
2682 let mut t: Vec<_> = db.export().into_iter().map(|f| f.text).collect();
2683 t.sort();
2684 t
2685 };
2686 assert_eq!(texts(&a), texts(&b), "batch size must not change the facts");
2687 }
2688
2689 #[test]
2690 fn config_table_feeds_settings_and_the_cli_batch_size() {
2691 let scratch = Scratch::new("settings");
2695 let cfgfile = scratch.0.join("config.toml");
2696 std::fs::write(
2697 &cfgfile,
2698 "[engine]\ndim = 512\n[embedder]\nkind = \"none\"\n\
2699 [maintenance]\nsnapshot_every_ops = 64\nbatch_size = 200\n",
2700 )
2701 .unwrap();
2702 let table = plugmem_host::read_config(Some(&cfgfile)).unwrap();
2703 let s = Settings::from_table(table.as_ref()).unwrap();
2704 assert_eq!(s.config.dim, 512);
2705 assert!(s.embedder.is_none());
2706 assert_eq!(s.snapshot_every_ops, Some(64));
2707 assert_eq!(read_batch_size(table.as_ref()), Some(200));
2708
2709 assert!(plugmem_host::read_config(Some(&scratch.0.join("nope.toml"))).is_err());
2711 }
2712
2713 #[test]
2714 fn checkpoint_command_flushes_the_journal_and_enables_the_readonly_path() {
2715 let scratch = Scratch::new("checkpoint-cmd");
2716 let path = scratch.0.join("m.plugmem");
2717
2718 let remember = Cli {
2720 db: Some(path.display().to_string()),
2721 workspace: None,
2722 config: None,
2723 json: false,
2724 command: Command::Remember {
2725 text: "hello tokio".into(),
2726 entity: None,
2727 tags: Vec::new(),
2728 links: Vec::new(),
2729 meta: Vec::new(),
2730 valid_from: None,
2731 vector: Vec::new(),
2732 },
2733 };
2734 assert_eq!(run_parsed(remember, &mut Vec::new()), 0);
2735
2736 let checkpoint = |json| Cli {
2738 db: Some(path.display().to_string()),
2739 workspace: None,
2740 config: None,
2741 json,
2742 command: Command::Checkpoint,
2743 };
2744 let mut buf = Vec::new();
2745 assert_eq!(run_parsed(checkpoint(false), &mut buf), 0);
2746 assert!(String::from_utf8(buf).unwrap().contains("checkpointed"));
2747
2748 let mut buf = Vec::new();
2750 assert_eq!(run_parsed(checkpoint(true), &mut buf), 0);
2751 let v: serde_json::Value =
2752 serde_json::from_str(String::from_utf8(buf).unwrap().trim()).unwrap();
2753 assert_eq!(v["ok"], true);
2754
2755 let scrub = Cli {
2758 db: Some(path.display().to_string()),
2759 workspace: None,
2760 config: None,
2761 json: false,
2762 command: Command::Scrub,
2763 };
2764 let mut buf = Vec::new();
2765 assert_eq!(run_parsed(scrub, &mut buf), 0);
2766 assert!(String::from_utf8(buf).unwrap().contains("scrub ok"));
2767 }
2768
2769 #[test]
2770 fn run_parsed_uses_the_readonly_path_after_a_checkpoint() {
2771 let scratch = Scratch::new("ro-route");
2772 let path = scratch.0.join("m.plugmem");
2773 {
2774 let (db, _) = Database::open(&path, Config::default()).unwrap();
2775 db.remember(RememberInput::text(1_000, "hello tokio"))
2776 .unwrap();
2777 db.checkpoint(2_000).unwrap(); }
2779 let cli = Cli {
2781 db: Some(path.display().to_string()),
2782 workspace: None,
2783 config: None,
2784 json: false,
2785 command: Command::Stats,
2786 };
2787 let mut buf = Vec::new();
2788 assert_eq!(run_parsed(cli, &mut buf), 0);
2789 assert!(String::from_utf8(buf).unwrap().contains("facts"));
2790
2791 let cli = Cli {
2793 db: Some(path.display().to_string()),
2794 workspace: None,
2795 config: None,
2796 json: false,
2797 command: Command::Recall {
2798 query: Some("tokio".into()),
2799 tags: Vec::new(),
2800 entities: Vec::new(),
2801 as_of: None,
2802 range: None,
2803 k: 0,
2804 closed: false,
2805 token_budget: None,
2806 ef: None,
2807 graph_depth: None,
2808 vector: Vec::new(),
2809 },
2810 };
2811 let mut buf = Vec::new();
2812 assert_eq!(run_parsed(cli, &mut buf), 0);
2813 assert!(String::from_utf8(buf).unwrap().contains("tokio"));
2814 }
2815}