car-server-core 0.47.0

Transport-neutral library for the CAR daemon JSON-RPC dispatcher (used by car-server and tokhn-daemon)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
//! Durable memory for the assistant: `remember` and `recall`.
//!
//! Backed by CAR's own graph memory ([`car_memgine::MemgineEngine`]) rather than
//! a bespoke store — this is the capability the assistant exists to showcase.
//! Remembered facts are persisted to `~/.car/memory/assistant.json` and
//! re-ingested on startup, so the assistant remembers across sessions and
//! process restarts.
//!
//! Recall runs the memgine's **full** retrieval path (`build_context`), not the
//! Fast one. Fast mode skips exactly the parts that make this a graph rather
//! than a list: skill lookup, PPR-based fact scoring, inline repair of stale
//! dependents, and known-unknowns extraction. The assistant was on Fast, so the
//! flagship agent advertised graph memory while doing keyword recall over flat
//! notes — the differentiator was bypassed on the one path users actually hit.
//!
//! Full is affordable here specifically because this engine is built with
//! `MemgineEngine::new(None)` and never has inference attached: the one
//! genuinely expensive step Fast avoids is the embedding flush, and that is
//! guarded by `self.inference.is_some()`, so it no-ops either way. What Full
//! adds is in-memory graph work — no I/O, no model call. If inference is ever
//! wired to this engine, re-measure before assuming that still holds.

use std::path::PathBuf;
use std::sync::{Arc, Mutex};

use async_trait::async_trait;
use car_engine::ToolExecutor;
use car_eventlog::Event;
use car_memgine::{
    MemgineEngine, ProactiveMaintenanceRequest, ProactiveMemoryDecision, ProactiveMemoryRequest,
    ProactiveMemorySave,
};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};

#[derive(Clone, Serialize, Deserialize)]
struct Note {
    subject: String,
    body: String,
    /// How the fact is used at recall time. `#[serde(default)]` so notes written
    /// before the field existed load as plain facts rather than failing the
    /// whole store's deserialization.
    #[serde(default)]
    kind: NoteKind,
}

/// The model-facing memory taxonomy. Each variant routes to genuinely different
/// downstream behavior in memgine — this is the behavioral hint the tool schema
/// carries instead of prose in the system prompt telling the model what is
/// "worth keeping".
#[derive(Clone, Copy, Default, PartialEq, Eq, Debug, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum NoteKind {
    /// Retrieved when relevant to the query (the Facts layer).
    #[default]
    Fact,
    /// A standing instruction. Saved as a memgine *constraint*, so it lands in
    /// the Active Constraints layer that context assembly includes in every
    /// session regardless of query.
    Preference,
    /// Procedural evidence — what was attempted and whether it worked.
    Procedure,
}

impl NoteKind {
    fn parse(raw: Option<&str>) -> Result<Self, String> {
        match raw.map(str::trim) {
            None | Some("") | Some("fact") => Ok(Self::Fact),
            Some("preference") => Ok(Self::Preference),
            Some("procedure") => Ok(Self::Procedure),
            Some(other) => Err(format!(
                "remember: unknown kind '{other}' (expected fact, preference, or procedure)"
            )),
        }
    }

    /// The wire spelling, matching the serde `rename_all = "lowercase"` and the
    /// vocabulary [`Self::parse`] accepts. Used for the sync payload.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Fact => "fact",
            Self::Preference => "preference",
            Self::Procedure => "procedure",
        }
    }

    /// Lenient read-side parse for a value that came off the sync wire.
    ///
    /// Unlike [`Self::parse`], an unrecognized kind is NOT an error: it means a
    /// peer running a newer build used a variant this device does not know
    /// about, and degrading that to a plain fact is strictly better than
    /// dropping the fact or failing the recall. A missing kind is a fact
    /// written before the wire carried one.
    pub fn from_wire(raw: Option<&str>) -> Self {
        Self::parse(raw).unwrap_or_default()
    }
}

/// One fact as it crosses the sync wire.
///
/// `kind` rides along because it is not decoration — a `Preference` is saved as
/// a memgine *constraint* and therefore lands in the always-included Active
/// Constraints layer, while a `Fact` surfaces only when a recall query matches
/// it. Reconstructing every peer-synced note as a plain fact (which is what
/// happened before car#665) silently downgraded a standing rule to something
/// the assistant would only sometimes see, on the second device and not the
/// first, with nothing anywhere reporting an error.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SyncedFact {
    pub subject: String,
    pub body: String,
    pub kind: NoteKind,
}

/// A sink that mirrors remembered facts into CAR's synced oplog
/// (`Surface::Knowledge`) so the assistant's memory can converge across a user's
/// devices. Supplied ONLY in the daemon-attached `car do --serve` process, which
/// reaches the daemon's single oplog owner over WS; `None` for one-shot `car do`,
/// so the standalone CLI never becomes a second writer to the lock-guarded oplog.
/// The write is best-effort: a sync failure must never fail the local remember.
#[async_trait]
pub trait MemorySync: Send + Sync {
    /// Append a remembered fact to the knowledge oplog. The fact is
    /// content-addressed by its `{subject, body, kind}` payload (no explicit id),
    /// so on the grow-only Knowledge tier each distinct value is its own
    /// immutable entity: an edit is a NEW op (both survive the fold), and an
    /// identical re-remember dedups. The read side ([`Self::pull_knowledge`])
    /// converges a subject to its newest-by-HLC body. `subject` is normalized
    /// (trimmed) to match the local store's entity boundary.
    ///
    /// **`kind` is part of the content address, and that is a deliberate
    /// migration decision (car#665).** Adding it changes the address of a fact
    /// that was already synced under the old `{subject, body}` payload, so such
    /// a fact re-remembered after the upgrade folds as a *second* entity rather
    /// than deduping against the first. That is harmless here because
    /// [`Self::pull_knowledge`] reduces newest-per-subject before anything sees
    /// the facts: both entities share a subject, the kinded one carries the
    /// later HLC, and the reduce keeps it. So the duplicate never reaches a
    /// reader — it only costs one extra op on a tier that is grow-only by
    /// design (an edit already appends). The alternative — inferring a kind on
    /// the receiving side from a locally-known preference with the same subject
    /// — would leave the wire lossy and only work when the same rule had been
    /// set on both devices, which is exactly the case that needs no fixing.
    async fn append_knowledge(
        &self,
        subject: &str,
        body: &str,
        kind: NoteKind,
    ) -> Result<(), String>;

    /// Pull peer-synced knowledge facts — the read side — already reduced to
    /// newest-per-subject. Best-effort: an `Err` means "no peer facts available
    /// right now", never a failure; `recall` degrades to local memory. The impl
    /// pumps the oplog then reads the folded Knowledge facts, reducing
    /// newest-per-subject by LAST-in-ascending-hlc order.
    async fn pull_knowledge(&self) -> Result<Vec<SyncedFact>, String>;
}

/// `remember` / `recall`, backed by a persistent memgine graph.
pub struct MemoryTools {
    inner: Mutex<Inner>,
    path: PathBuf,
    /// Optional mirror into the synced knowledge oplog (daemon serve only).
    sync: Option<Arc<dyn MemorySync>>,
}

struct Inner {
    engine: MemgineEngine,
    notes: Vec<Note>,
}

impl MemoryTools {
    /// Open (or create) the assistant's memory at `path`, re-ingesting any
    /// previously remembered facts so recall works immediately.
    pub fn open(path: PathBuf) -> Self {
        let notes: Vec<Note> = std::fs::read_to_string(&path)
            .ok()
            .and_then(|s| serde_json::from_str(&s).ok())
            .unwrap_or_default();
        let mut engine = MemgineEngine::new(None);
        for (i, n) in notes.iter().enumerate() {
            ingest(&mut engine, i, n);
        }
        Self {
            inner: Mutex::new(Inner { engine, notes }),
            path,
            sync: None,
        }
    }

    /// Attach a synced-oplog mirror so remembered facts converge across the
    /// user's devices. Set only on the daemon-attached serve path; leaving it
    /// unset keeps `remember` a purely local write (one-shot `car do`).
    pub fn with_sync(mut self, sync: Option<Arc<dyn MemorySync>>) -> Self {
        self.sync = sync;
        self
    }

    /// The two model-facing tool schemas.
    pub fn tool_defs() -> Vec<Value> {
        vec![
            json!({
                "name": "remember",
                "description": "Save a durable fact about the user or task so you can recall it in \
                                future sessions. Use for things worth persisting, not transient \
                                details. `kind` decides how the fact is used later, so pick it \
                                deliberately.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "subject": { "type": "string", "description": "Short label for the fact (e.g. 'project name')." },
                        "body": { "type": "string", "description": "The fact to remember." },
                        "kind": {
                            "type": "string",
                            "enum": ["fact", "preference", "procedure"],
                            "description": "'fact' (default): something true about the user, \
                                            project, or environment; retrieved when relevant to \
                                            the query. 'preference': a standing instruction about \
                                            how the user wants you to work — surfaced in EVERY \
                                            future session, so reserve it for rules you should \
                                            never violate. 'procedure': how a task was done and \
                                            whether it worked, for reuse next time."
                        }
                    },
                    "required": ["subject", "body"]
                },
                "mutating": true,
                "tier": "full_access"
            }),
            json!({
                "name": "recall",
                "description": "Retrieve previously remembered facts relevant to a query. Use at the \
                                start of a task to recall what you know about the user or project.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": { "type": "string", "description": "What to recall." }
                    },
                    "required": ["query"]
                }
            }),
        ]
    }

    /// Returns the tool result plus the validated [`NoteKind`], so the caller's
    /// sync mirror sends the kind this call actually stored rather than
    /// re-parsing the raw params and risking a second, divergent answer.
    fn remember(&self, params: &Value) -> Result<(Value, NoteKind), String> {
        // Trim the subject up front so the local entity boundary matches the
        // synced one (the mirror sends the same normalized subject).
        let subject = normalize_subject(
            params
                .get("subject")
                .and_then(Value::as_str)
                .ok_or("remember requires a 'subject' string")?,
        );
        let body = params
            .get("body")
            .and_then(Value::as_str)
            .ok_or("remember requires a 'body' string")?;
        let kind = NoteKind::parse(params.get("kind").and_then(Value::as_str))?;
        let mut g = self.inner.lock().map_err(|_| "memory lock poisoned")?;
        let note = Note {
            subject: subject.to_string(),
            body: body.to_string(),
            kind,
        };
        // Supersede an existing fact with the same subject (case-insensitive) so
        // an updated fact replaces the stale one instead of accumulating. A
        // re-remember may also RECLASSIFY (a fact promoted to a preference), so
        // the kind is superseded with the body.
        match g
            .notes
            .iter_mut()
            .find(|n| n.subject.eq_ignore_ascii_case(subject))
        {
            Some(existing) => {
                existing.body = body.to_string();
                existing.kind = kind;
            }
            None => g.notes.push(note),
        }
        // Re-ingest the whole set so the memgine graph mirrors the current notes
        // (cheap for a personal store; correctness over cleverness).
        let mut engine = MemgineEngine::new(None);
        for (i, n) in g.notes.iter().enumerate() {
            ingest(&mut engine, i, n);
        }
        g.engine = engine;
        // Persist the whole set (small; correctness over cleverness).
        if let Some(parent) = self.path.parent() {
            let _ = std::fs::create_dir_all(parent);
        }
        let serialized =
            serde_json::to_string_pretty(&g.notes).map_err(|e| format!("serialize: {e}"))?;
        std::fs::write(&self.path, serialized).map_err(|e| format!("persist memory: {e}"))?;
        Ok((
            json!({ "remembered": subject, "total_facts": g.notes.len() }),
            kind,
        ))
    }

    /// Best-effort mirror of a just-remembered fact into the synced knowledge
    /// oplog. Only active when a [`MemorySync`] is attached (daemon serve). A
    /// sync failure is logged and swallowed — the fact is already saved locally,
    /// and convergence is best-effort, so it must never fail the tool call.
    async fn mirror_to_sync(&self, params: &Value, kind: NoteKind) {
        let Some(sync) = &self.sync else {
            return;
        };
        let (Some(subject), Some(body)) = (
            params.get("subject").and_then(Value::as_str),
            params.get("body").and_then(Value::as_str),
        ) else {
            return;
        };
        // Same normalized subject as the local store; the fact is content-addressed
        // by {subject, body, kind} on the sync side (no explicit id). The kind is
        // the one `remember` already validated, not a re-parse of the raw params.
        if let Err(e) = sync
            .append_knowledge(normalize_subject(subject), body, kind)
            .await
        {
            tracing::debug!(
                target: "assistant.memory",
                "knowledge sync append failed (non-fatal): {e}"
            );
        }
    }

    /// Best-effort pull of peer-synced facts. Returns empty on no sink or any
    /// error — recall must never fail because sync is unavailable.
    async fn pull_peer_knowledge(&self) -> Vec<SyncedFact> {
        let Some(sync) = &self.sync else {
            return Vec::new();
        };
        match sync.pull_knowledge().await {
            Ok(facts) => facts,
            Err(e) => {
                tracing::debug!(
                    target: "assistant.memory",
                    "peer knowledge pull failed (non-fatal): {e}"
                );
                Vec::new()
            }
        }
    }

    async fn recall(&self, params: &Value) -> Result<Value, String> {
        let query = params
            .get("query")
            .and_then(Value::as_str)
            .ok_or("recall requires a 'query' string")?;
        // Best-effort peer pull FIRST — no lock held across the await. Peers are
        // already reduced to newest-per-subject by the sink.
        let peers = self.pull_peer_knowledge().await;
        let mut g = self.inner.lock().map_err(|_| "memory lock poisoned")?;
        if g.notes.is_empty() && peers.is_empty() {
            return Ok(json!({ "query": query, "context": "", "note": "no facts remembered yet" }));
        }
        // Option B: local notes are authoritative for the subjects THIS device
        // holds; peer facts fill only the gaps. A peer's newer edit to a locally
        // held subject is intentionally NOT applied — local has no HLC to compare
        // against, so this is conservative local authority, not cross-device
        // newest-wins. Same case-insensitive trimmed key as the local supersede.
        let local_keys: std::collections::HashSet<String> =
            g.notes.iter().map(|n| subject_key(&n.subject)).collect();
        let gaps: Vec<&SyncedFact> = peers
            .iter()
            .filter(|fact| !local_keys.contains(&subject_key(&fact.subject)))
            .collect();
        if gaps.is_empty() {
            // Nothing new from peers — the maintained local engine has the answer.
            let context = g.engine.build_context(query);
            return Ok(json!({ "query": query, "context": context }));
        }
        // Build a fresh view = local notes + peer-gap facts for this recall (the
        // peer facts live in the oplog and are re-pulled each recall, never
        // written into the local notes file).
        let mut engine = MemgineEngine::new(None);
        let mut idx = 0;
        for n in &g.notes {
            ingest(&mut engine, idx, n);
            idx += 1;
        }
        for fact in gaps {
            ingest(
                &mut engine,
                idx,
                &Note {
                    subject: fact.subject.clone(),
                    body: fact.body.clone(),
                    // The peer's own kind, carried over the wire (car#665). A
                    // preference must stay a preference here: reconstructing it
                    // as a plain fact demoted the constraint to something
                    // recall-only, so a standing rule quietly stopped applying
                    // on this device and kept applying on the one that set it.
                    kind: fact.kind,
                },
            );
            idx += 1;
        }
        let context = engine.build_context(query);
        Ok(json!({ "query": query, "context": context }))
    }

    /// Run the paper-style proactive memory pass for the assistant loop.
    ///
    /// This is deliberately host-side and deterministic: update compact memory
    /// from the runtime trajectory, then decide whether one remembered fact is
    /// strong enough to interrupt the next model turn. The model still has the
    /// explicit `recall` tool, but it no longer has to remember to call it before
    /// CAR can surface high-value memory.
    pub async fn proactive_intervention(
        &self,
        query: &str,
        recent: Vec<String>,
        events: &[Event],
    ) -> Result<
        (
            car_memgine::ProactiveMaintenanceReport,
            ProactiveMemoryDecision,
        ),
        String,
    > {
        let peers = self.pull_peer_knowledge().await;
        let mut g = self.inner.lock().map_err(|_| "memory lock poisoned")?;

        let local_keys: std::collections::HashSet<String> =
            g.notes.iter().map(|n| subject_key(&n.subject)).collect();
        let gaps: Vec<&SyncedFact> = peers
            .iter()
            .filter(|fact| !local_keys.contains(&subject_key(&fact.subject)))
            .collect();

        let maintenance_request = ProactiveMaintenanceRequest {
            max_recent: 32,
            tenant_id: None,
        };
        let mut request = ProactiveMemoryRequest {
            query: query.to_string(),
            recent,
            ..Default::default()
        };

        if gaps.is_empty() {
            let maintenance = g
                .engine
                .maintain_proactive_memory_from_events(events, &maintenance_request);
            request.trigger.merge(maintenance.trigger.clone());
            let decision = g.engine.proactive_intervention(&request);
            return Ok((maintenance, decision));
        }

        let mut engine = MemgineEngine::new(None);
        let mut idx = 0;
        for n in &g.notes {
            ingest(&mut engine, idx, n);
            idx += 1;
        }
        for fact in gaps {
            ingest(
                &mut engine,
                idx,
                &Note {
                    subject: fact.subject.clone(),
                    body: fact.body.clone(),
                    // The peer's own kind, carried over the wire (car#665). A
                    // preference must stay a preference here: reconstructing it
                    // as a plain fact demoted the constraint to something
                    // recall-only, so a standing rule quietly stopped applying
                    // on this device and kept applying on the one that set it.
                    kind: fact.kind,
                },
            );
            idx += 1;
        }
        let maintenance =
            engine.maintain_proactive_memory_from_events(events, &maintenance_request);
        request.trigger.merge(maintenance.trigger.clone());
        let decision = engine.proactive_intervention(&request);
        Ok((maintenance, decision))
    }
}

/// The shared subject normalization for the local store and the synced mirror,
/// so both draw the fact's entity boundary at the same place (trimmed).
fn normalize_subject(subject: &str) -> &str {
    subject.trim()
}

/// The case-insensitive, trimmed key for a subject — the same entity boundary
/// the local supersede uses (`normalize_subject` + case-insensitive match). The
/// peer-merge gate and the sink's newest-per-subject reduce must use this exact
/// key, or `Pet`/`pet` split into two entities.
fn subject_key(subject: &str) -> String {
    normalize_subject(subject).to_ascii_lowercase()
}

/// Ingest one note as a memgine fact. Deterministic id per index so a reload
/// re-creates the same graph.
fn ingest(engine: &mut MemgineEngine, idx: usize, n: &Note) {
    let save = ProactiveMemorySave {
        id: Some(format!("assistant-note-{idx}")),
        subject: n.subject.clone(),
        body: n.body.clone(),
        tags: vec!["assistant_memory".to_string()],
        confidence: Some("high".to_string()),
        tenant_id: None,
        // A preference is a standing rule: mark it a constraint so context
        // assembly surfaces it in the always-included Active Constraints layer
        // rather than only when the query happens to match.
        is_constraint: n.kind == NoteKind::Preference,
    };
    match n.kind {
        NoteKind::Procedure => engine.save_proactive_procedural(save),
        NoteKind::Fact | NoteKind::Preference => engine.save_proactive_knowledge(save),
    };
}

#[async_trait]
impl ToolExecutor for MemoryTools {
    async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
        match tool {
            "remember" => {
                let (out, kind) = self.remember(params)?;
                // Mirror into the synced oplog after the local write succeeds.
                self.mirror_to_sync(params, kind).await;
                Ok(out)
            }
            "recall" => self.recall(params).await,
            other => Err(format!("unknown tool: '{other}'")),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn remember_declares_mutating_and_full_access() {
        let defs = MemoryTools::tool_defs();
        let remember = defs
            .iter()
            .find(|def| def["name"] == "remember")
            .expect("remember tool def");
        assert_eq!(remember["mutating"], true);
        assert_eq!(remember["tier"], "full_access");
    }

    #[test]
    fn remember_advertises_the_memory_taxonomy() {
        let defs = MemoryTools::tool_defs();
        let remember = defs.iter().find(|def| def["name"] == "remember").unwrap();
        let kinds = remember["parameters"]["properties"]["kind"]["enum"]
            .as_array()
            .expect("kind is an enum — the schema is what teaches the model");
        for expected in ["fact", "preference", "procedure"] {
            assert!(kinds.iter().any(|k| k == expected), "missing {expected}");
        }
        // Not required: an unspecified kind is a plain fact.
        let required = remember["parameters"]["required"].as_array().unwrap();
        assert!(!required.iter().any(|r| r == "kind"));
    }

    #[tokio::test]
    async fn preference_becomes_an_always_included_constraint() {
        let dir = tempfile::tempdir().unwrap();
        let mem = MemoryTools::open(dir.path().join("assistant.json"));
        mem.execute(
            "remember",
            &json!({
                "subject": "code review",
                "body": "Never merge without a green CI run.",
                "kind": "preference"
            }),
        )
        .await
        .unwrap();
        mem.execute(
            "remember",
            &json!({ "subject": "project name", "body": "The project is called Zephyr." }),
        )
        .await
        .unwrap();

        // A query unrelated to either fact: the preference still surfaces (it is
        // a constraint), which is the behavioral difference the enum encodes.
        let out = mem
            .execute("recall", &json!({ "query": "what time is the standup" }))
            .await
            .unwrap();
        let context = out["context"].as_str().unwrap();
        assert!(
            context.contains("Active Constraints") && context.contains("green CI run"),
            "preference must land in the always-included constraints layer: {context}"
        );
    }

    #[tokio::test]
    async fn unknown_kind_is_rejected_rather_than_silently_stored() {
        let dir = tempfile::tempdir().unwrap();
        let mem = MemoryTools::open(dir.path().join("assistant.json"));
        let err = mem
            .execute(
                "remember",
                &json!({ "subject": "s", "body": "b", "kind": "identity" }),
            )
            .await
            .unwrap_err();
        assert!(err.contains("unknown kind 'identity'"), "{err}");
    }

    #[tokio::test]
    async fn remembers_across_reopen() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("assistant.json");

        {
            let mem = MemoryTools::open(path.clone());
            mem.execute(
                "remember",
                &json!({ "subject": "project name", "body": "The project is called Zephyr." }),
            )
            .await
            .unwrap();
        }

        // Re-open (simulating a new process/session) and recall.
        let mem = MemoryTools::open(path.clone());
        let out = mem
            .execute("recall", &json!({ "query": "what is my project called?" }))
            .await
            .unwrap();
        let ctx = out["context"].as_str().unwrap();
        assert!(
            ctx.contains("Zephyr"),
            "recall should surface the fact: {ctx}"
        );
    }

    #[tokio::test]
    async fn remember_supersedes_same_subject() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("m.json");
        let mem = MemoryTools::open(path.clone());
        mem.execute("remember", &json!({ "subject": "editor", "body": "vim" }))
            .await
            .unwrap();
        let out = mem
            .execute("remember", &json!({ "subject": "Editor", "body": "emacs" }))
            .await
            .unwrap();
        // Same subject (case-insensitive) → replaced, not accumulated.
        assert_eq!(out["total_facts"], 1);
        let notes: Vec<Note> =
            serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
        assert_eq!(notes.len(), 1);
        assert_eq!(notes[0].body, "emacs");
    }

    #[tokio::test]
    async fn recall_on_empty_is_graceful() {
        let dir = tempfile::tempdir().unwrap();
        let mem = MemoryTools::open(dir.path().join("m.json"));
        let out = mem
            .execute("recall", &json!({ "query": "anything" }))
            .await
            .unwrap();
        assert_eq!(out["context"], "");
    }

    /// Recall must run the FULL memgine path, not Fast.
    ///
    /// The observable difference on the assistant's real data shape is **fact
    /// ordering**. The four-layer model is relevance-*ascending* — most relevant
    /// last, so it sits closest to the model's recency attention — and that
    /// ranking is PPR-scored, which Fast skips entirely. On Fast the assistant
    /// emitted facts in creation order: recall was not relevance-ranked at all,
    /// while the docs advertised spreading activation.
    ///
    /// So this asserts the matching fact lands LAST despite being created FIRST.
    /// Under Fast it comes back first and this fails, which is the regression
    /// worth catching: switching one call back to `build_context_fast` would
    /// silently downgrade the flagship agent to keyword recall over flat notes.
    ///
    /// (Known-unknowns extraction, the other Full-only step, reads *conversation*
    /// nodes; assistant notes ingest as facts, so it does not fire here.)
    #[tokio::test]
    async fn recall_ranks_by_relevance_not_creation_order() {
        let dir = tempfile::tempdir().unwrap();
        let mem = MemoryTools::open(dir.path().join("m.json"));

        // The match is remembered FIRST, so creation order and relevance order
        // disagree — otherwise the assertion would pass either way.
        for (subject, body) in [
            ("deployment target", "Unclear which region we deploy to."),
            ("db", "Postgres 16 in prod."),
            ("owner", "Matt owns the release process."),
        ] {
            mem.execute("remember", &json!({ "subject": subject, "body": body }))
                .await
                .unwrap();
        }

        let out = mem
            .execute("recall", &json!({ "query": "deployment target" }))
            .await
            .unwrap();
        let ctx = out["context"].as_str().unwrap_or_default();

        let facts: Vec<&str> = ctx
            .lines()
            .skip_while(|l| !l.starts_with("## Current Facts"))
            .filter(|l| l.starts_with("- "))
            .collect();
        assert!(facts.len() >= 3, "expected all facts back, got: {ctx}");
        assert!(
            facts.last().unwrap().contains("deployment target"),
            "the query-matching fact must rank last (relevance-ascending); \
             got creation order, which means Fast mode is running.\nfacts: {facts:#?}"
        );
    }

    struct RecordingSync(Mutex<Vec<SyncedFact>>);
    #[async_trait]
    impl MemorySync for RecordingSync {
        async fn append_knowledge(
            &self,
            subject: &str,
            body: &str,
            kind: NoteKind,
        ) -> Result<(), String> {
            self.0.lock().unwrap().push(SyncedFact {
                subject: subject.into(),
                body: body.into(),
                kind,
            });
            Ok(())
        }
        async fn pull_knowledge(&self) -> Result<Vec<SyncedFact>, String> {
            Ok(Vec::new())
        }
    }

    /// A sink that returns a fixed set of peer facts on pull (append is a no-op).
    struct PeerSync(Vec<SyncedFact>);
    #[async_trait]
    impl MemorySync for PeerSync {
        async fn append_knowledge(&self, _: &str, _: &str, _: NoteKind) -> Result<(), String> {
            Ok(())
        }
        async fn pull_knowledge(&self) -> Result<Vec<SyncedFact>, String> {
            Ok(self.0.clone())
        }
    }

    /// A peer fact, for the `PeerSync` fixtures.
    fn peer_fact(subject: &str, body: &str, kind: NoteKind) -> SyncedFact {
        SyncedFact {
            subject: subject.into(),
            body: body.into(),
            kind,
        }
    }

    #[tokio::test]
    async fn remember_mirrors_normalized_subject_and_body_to_sync() {
        let dir = tempfile::tempdir().unwrap();
        let rec = Arc::new(RecordingSync(Mutex::new(Vec::new())));
        let mem = MemoryTools::open(dir.path().join("m.json")).with_sync(Some(rec.clone()));
        // Untrimmed subject → the mirror sees the same normalized subject the
        // local store uses, so both draw the entity boundary in the same place.
        mem.execute(
            "remember",
            &json!({ "subject": "  Editor  ", "body": "vim" }),
        )
        .await
        .unwrap();
        let got = rec.0.lock().unwrap().clone();
        assert_eq!(got, vec![peer_fact("Editor", "vim", NoteKind::Fact)]);
    }

    #[tokio::test]
    async fn remember_without_sync_stays_local_only() {
        // One-shot `car do`: no sink attached, remember still works.
        let dir = tempfile::tempdir().unwrap();
        let mem = MemoryTools::open(dir.path().join("m.json"));
        let out = mem
            .execute("remember", &json!({ "subject": "x", "body": "y" }))
            .await
            .unwrap();
        assert_eq!(out["remembered"], "x");
    }

    struct FailingSync;
    #[async_trait]
    impl MemorySync for FailingSync {
        async fn append_knowledge(&self, _: &str, _: &str, _: NoteKind) -> Result<(), String> {
            Err("oplog unreachable".into())
        }
        async fn pull_knowledge(&self) -> Result<Vec<SyncedFact>, String> {
            Err("oplog unreachable".into())
        }
    }

    #[tokio::test]
    async fn recall_surfaces_a_peer_fact_not_held_locally() {
        // No local notes; the fact exists only via a peer's sync.
        let dir = tempfile::tempdir().unwrap();
        let peer = Arc::new(PeerSync(vec![peer_fact(
            "project",
            "the project is Zephyr",
            NoteKind::Fact,
        )]));
        let mem = MemoryTools::open(dir.path().join("m.json")).with_sync(Some(peer));
        let out = mem
            .execute("recall", &json!({ "query": "what is my project?" }))
            .await
            .unwrap();
        assert!(
            out["context"].as_str().unwrap().contains("Zephyr"),
            "{}",
            out["context"]
        );
    }

    /// car#665. A `preference` is saved as a memgine *constraint*, which context
    /// assembly includes in every session regardless of the query; a `fact`
    /// surfaces only when the query matches it. Reconstructing every peer-synced
    /// note as `NoteKind::default()` collapsed that distinction, so a standing
    /// rule set on device A quietly stopped being a standing rule on device B —
    /// with no error anywhere.
    ///
    /// The assertion is on the *layer* the note lands in, mirroring the local
    /// `preference_becomes_an_always_included_constraint`. Mere presence in the
    /// context does not discriminate: with a small store `build_context`
    /// lists an unmatched fact under Current Facts anyway. Active Constraints is
    /// what makes a rule unconditional, so that is what must be asserted.
    #[tokio::test]
    async fn a_peer_preference_stays_a_standing_rule_and_a_peer_fact_does_not() {
        let rule = "Never merge without a green CI run.";
        let unrelated = "what time is the standup";

        let recall_with = |kind| async move {
            let dir = tempfile::tempdir().unwrap();
            let peer = Arc::new(PeerSync(vec![peer_fact("code review", rule, kind)]));
            let mem = MemoryTools::open(dir.path().join("m.json")).with_sync(Some(peer));
            let out = mem
                .execute("recall", &json!({ "query": unrelated }))
                .await
                .unwrap();
            out["context"].as_str().unwrap().to_string()
        };

        let as_preference = recall_with(NoteKind::Preference).await;
        assert!(
            as_preference.contains("Active Constraints") && as_preference.contains(rule),
            "a synced preference must land in the always-included constraints \
             layer, exactly as it does on the device that set it: {as_preference}"
        );

        // The control: the same note synced as a plain fact — which is what the
        // receiving device used to reconstruct for *every* peer note — must not
        // reach that layer. Without this the assertion above would also pass if
        // the layer heading were always emitted, proving nothing.
        let as_fact = recall_with(NoteKind::Fact).await;
        assert!(
            !as_fact.contains("Active Constraints"),
            "control: a plain fact is not a standing rule — if it reaches the \
             constraints layer the test cannot tell the two kinds apart: {as_fact}"
        );
    }

    #[tokio::test]
    async fn remember_mirrors_the_kind_so_a_preference_crosses_the_wire() {
        let dir = tempfile::tempdir().unwrap();
        let rec = Arc::new(RecordingSync(Mutex::new(Vec::new())));
        let mem = MemoryTools::open(dir.path().join("m.json")).with_sync(Some(rec.clone()));
        mem.execute(
            "remember",
            &json!({
                "subject": "code review",
                "body": "Never merge without a green CI run.",
                "kind": "preference"
            }),
        )
        .await
        .unwrap();
        assert_eq!(
            rec.0.lock().unwrap().clone(),
            vec![peer_fact(
                "code review",
                "Never merge without a green CI run.",
                NoteKind::Preference
            )],
            "the mirror must send the kind `remember` validated, not a default"
        );
    }

    #[tokio::test]
    async fn recall_keeps_local_over_a_peer_edit_of_the_same_subject() {
        // Local set editor=vim; a peer later set editor=emacs. Option B: local is
        // authoritative for its own subjects, so the peer edit is SUPPRESSED
        // (conservative local authority, not cross-device newest-wins). Pinned so
        // nobody "fixes" it by accident.
        let dir = tempfile::tempdir().unwrap();
        let peer = Arc::new(PeerSync(vec![peer_fact("editor", "emacs", NoteKind::Fact)]));
        let mem = MemoryTools::open(dir.path().join("m.json")).with_sync(Some(peer));
        mem.execute("remember", &json!({ "subject": "editor", "body": "vim" }))
            .await
            .unwrap();
        let out = mem
            .execute("recall", &json!({ "query": "which editor?" }))
            .await
            .unwrap();
        let ctx = out["context"].as_str().unwrap().to_string();
        assert!(ctx.contains("vim"), "{ctx}");
        assert!(
            !ctx.contains("emacs"),
            "peer edit must not override local under Option B: {ctx}"
        );
    }

    #[tokio::test]
    async fn recall_peer_gate_is_case_insensitive() {
        // "Pet" (local) and "pet" (peer) are the same entity — local wins.
        let dir = tempfile::tempdir().unwrap();
        let peer = Arc::new(PeerSync(vec![peer_fact(
            "pet",
            "a peer dog",
            NoteKind::Fact,
        )]));
        let mem = MemoryTools::open(dir.path().join("m.json")).with_sync(Some(peer));
        mem.execute(
            "remember",
            &json!({ "subject": "Pet", "body": "my cat Mittens" }),
        )
        .await
        .unwrap();
        let out = mem
            .execute("recall", &json!({ "query": "what pet?" }))
            .await
            .unwrap();
        let ctx = out["context"].as_str().unwrap().to_string();
        assert!(ctx.contains("Mittens"), "{ctx}");
        assert!(
            !ctx.contains("peer dog"),
            "a case-variant peer subject must be gated: {ctx}"
        );
    }

    #[tokio::test]
    async fn recall_pull_failure_degrades_to_local() {
        // A pull error must never fail recall — it answers from local memory.
        let dir = tempfile::tempdir().unwrap();
        let mem =
            MemoryTools::open(dir.path().join("m.json")).with_sync(Some(Arc::new(FailingSync)));
        mem.execute("remember", &json!({ "subject": "city", "body": "Denver" }))
            .await
            .unwrap();
        let out = mem
            .execute("recall", &json!({ "query": "which city?" }))
            .await
            .unwrap();
        assert!(out["context"].as_str().unwrap().contains("Denver"));
    }

    #[tokio::test]
    async fn remember_survives_sync_failure() {
        // A sync hiccup must never fail the local remember — the fact is saved.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("m.json");
        let mem = MemoryTools::open(path.clone()).with_sync(Some(Arc::new(FailingSync)));
        let out = mem
            .execute("remember", &json!({ "subject": "a", "body": "b" }))
            .await
            .unwrap();
        assert_eq!(out["remembered"], "a");
        // And it persisted locally despite the sync error.
        let notes: Vec<Note> =
            serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
        assert_eq!(notes.len(), 1);
        assert_eq!(notes[0].body, "b");
    }

    // Fold two same-subject facts through the REAL car-sync fold to prove the
    // write format converges correctly on the grow-only Knowledge tier. The
    // earlier stable-per-subject key collapsed these to one op and kept the
    // EARLIEST (dropping the update); content-addressing keeps both, and the
    // read model (newest-by-HLC per subject) recovers the edit.
    #[test]
    fn synced_knowledge_survives_edits_and_converges_to_newest_body() {
        use car_sync::fold::fold;
        use car_sync::oplog::{DeviceLog, Scope, Surface};

        let mut log = DeviceLog::new("device-a");
        let vim = log.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({ "subject": "editor", "body": "vim" }),
        );
        let emacs = log.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({ "subject": "editor", "body": "emacs" }),
        );

        let state = fold(&[vim, emacs]);
        let entries = state.log_entries("knowledge");
        // Both edits survive as distinct immutable entities (no update dropped).
        assert_eq!(entries.len(), 2);
        // The read model converges a subject to its newest-by-HLC body.
        let newest = entries
            .iter()
            .filter(|r| r.payload["subject"] == "editor")
            .max_by(|a, b| a.hlc.cmp(&b.hlc))
            .unwrap();
        assert_eq!(newest.payload["body"], "emacs");
    }

    #[test]
    fn identical_reremember_dedups_in_the_synced_fold() {
        use car_sync::fold::fold;
        use car_sync::oplog::{DeviceLog, Scope, Surface};

        let mut log = DeviceLog::new("device-a");
        let a = log.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({ "subject": "editor", "body": "vim" }),
        );
        let b = log.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({ "subject": "editor", "body": "vim" }),
        );
        // Same {subject, body} → same content-addressed key → one entity, no spew.
        let state = fold(&[a, b]);
        assert_eq!(state.log_entries("knowledge").len(), 1);
    }
}