topodb_json/
compose.rs

1//! Composed write planning shared by TopoDB's front ends: the `remember`
2//! verb (store + find-or-create + link + supersede as ONE atomic batch) and
3//! the entity/memory lookups it is built from. Unlike the rest of this
4//! crate, functions here READ from a `Db` — but never write: every function
5//! returns planned `Op`s (or a lookup result) and the caller submits.
6
7use std::collections::{BTreeMap, BTreeSet, HashMap};
8
9use serde_json::Value;
10use topodb::{
11    Db, EdgeId, NodeId, NodeRecord, Op, PropValue, Props, Scope, ScopeSet, TimeAxis, TopoError,
12};
13
14use crate::{
15    merge_required_prop, normalize_edge_type, scopes_to_scope_set, ALIAS_EDGE_TYPE, ALIAS_LABEL,
16    ALIAS_NAME_PROP, ENTITY_LABEL, ENTITY_NAME_PROP, MEMORY_CONTENT_HASH_PROP, MEMORY_CONTENT_PROP,
17    MEMORY_FORGOTTEN_AT_PROP, MEMORY_LABEL, MEMORY_SUPERSEDED_AT_PROP,
18};
19
20/// Edge type `remember` uses when the caller doesn't name one.
21pub const DEFAULT_REMEMBER_EDGE_TYPE: &str = "about";
22
23/// Planning failure: `Invalid` is a caller-fixable input problem (surface the
24/// message verbatim); `Engine` is a database failure unrelated to the input.
25#[derive(Debug)]
26pub enum ComposeError {
27    Invalid(String),
28    Engine(TopoError),
29}
30
31impl From<TopoError> for ComposeError {
32    fn from(e: TopoError) -> Self {
33        ComposeError::Engine(e)
34    }
35}
36
37// --- moved verbatim from topodb-mcp/src/server.rs (self.db -> db) ---
38// Keep the original doc comments from server.rs on each of these when moving.
39
40/// In-call dedup key for `remember`'s entity names: whitespace-collapsed,
41/// lowercased — mirroring the engine's prop-index normalization
42/// (`prop_index::normalize_str`, which is pub(crate) and thus can't be
43/// called from here). Drift between the two only weakens IN-CALL dedup
44/// (["Drew", "drew"] in one call); cross-call dedup always goes through the
45/// engine's own normalized index via find_existing_entity.
46pub fn entity_dedup_key(name: &str) -> String {
47    name.split_whitespace()
48        .collect::<Vec<_>>()
49        .join(" ")
50        .to_lowercase()
51}
52
53/// Normalize memory content for dedup: trim and collapse internal whitespace.
54/// Deliberately NOT lowercased — casing can carry meaning in a stored fact.
55pub fn normalize_content(content: &str) -> String {
56    content.split_whitespace().collect::<Vec<_>>().join(" ")
57}
58
59/// Stable FNV-1a 64-bit hash of normalized content, hex-encoded. PERSISTED
60/// (equality-indexed as `content_hash`) — the algorithm must never change.
61/// Collisions are harmless: dedup always verifies exact normalized content.
62pub fn content_hash(content: &str) -> String {
63    let normalized = normalize_content(content);
64    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
65    for b in normalized.as_bytes() {
66        h ^= *b as u64;
67        h = h.wrapping_mul(0x0000_0100_0000_01b3);
68    }
69    format!("{h:016x}")
70}
71
72/// Builds a new Memory node's props: caller `extra` is validated (the
73/// system-maintained keys below are rejected, and `content` collides via
74/// merge_required_prop), then the whitespace-normalized content hash is
75/// stamped. The ONE constructor for every front end's new-memory write
76/// (plan_remember, MCP create_memory, CLI create-memory) — so the reserved
77/// set cannot drift between surfaces.
78pub fn memory_props(content: &str, extra: Option<&Value>) -> Result<Props, String> {
79    if let Some(Value::Object(map)) = extra {
80        for reserved in [
81            MEMORY_CONTENT_HASH_PROP,
82            MEMORY_SUPERSEDED_AT_PROP,
83            MEMORY_FORGOTTEN_AT_PROP,
84        ] {
85            if map.contains_key(reserved) {
86                return Err(format!(
87                    "props must not include {reserved:?}: it is maintained by the engine write path"
88                ));
89            }
90        }
91        if map.contains_key(crate::MEMORY_KIND_PROP) {
92            return Err(format!(
93                "props must not include {:?}: set it via remember's kind parameter \
94                 (episodic | semantic | procedural | decision)",
95                crate::MEMORY_KIND_PROP
96            ));
97        }
98    }
99    let mut props = merge_required_prop(
100        MEMORY_CONTENT_PROP,
101        PropValue::Str(content.to_string()),
102        extra,
103    )?;
104    props.insert(
105        MEMORY_CONTENT_HASH_PROP.into(),
106        PropValue::Str(content_hash(content)),
107    );
108    Ok(props)
109}
110
111/// Canonical entities for `name`: direct (Entity, name) matches plus
112/// (Alias, name) matches followed through alias_of. Deduped by id,
113/// oldest first.
114///
115/// Returns the raw `TopoError` (not `ErrorData`) rather than swallowing
116/// it: the two existing call sites disagree on what an undeclared
117/// (Entity, name) index should mean. `find_by_prop` must still surface it
118/// as a caller error — that is the exact contract tests pin down (an
119/// undeclared-index probe on a custom spec must error, not silently return
120/// empty, or a clobbered spec reopen would go undetected). `create_entity`
121/// instead treats it as "can't dedup on this spec" and degrades to
122/// create-always. Only the (Alias, name) probe's `Rejected` is
123/// unconditionally swallowed here — a spec that predates the Alias index
124/// (or a custom spec that never declared it) simply has no aliases to
125/// resolve, which is never a caller error.
126pub fn resolve_entities_by_name(
127    db: &Db,
128    scopes: &ScopeSet,
129    name: &str,
130) -> Result<Vec<NodeRecord>, TopoError> {
131    let value = PropValue::Str(name.to_string());
132    let mut out = db.nodes_by_prop_normalized(scopes, ENTITY_LABEL, ENTITY_NAME_PROP, &value)?;
133    let aliases = match db.nodes_by_prop_normalized(scopes, ALIAS_LABEL, ALIAS_NAME_PROP, &value) {
134        Ok(hits) => hits,
135        Err(TopoError::Rejected(_)) => Vec::new(),
136        Err(e) => return Err(e),
137    };
138    for alias in aliases {
139        for edge in db.edges_from(
140            scopes,
141            alias.id,
142            None,
143            Some(ALIAS_EDGE_TYPE),
144            true,
145            TimeAxis::Valid,
146        )? {
147            if let Some(canonical) = db.node(scopes, edge.to) {
148                if canonical.label == ENTITY_LABEL {
149                    out.push(canonical);
150                }
151            }
152        }
153    }
154    out.sort_by_key(|n| n.id);
155    out.dedup_by_key(|n| n.id);
156    Ok(out)
157}
158
159/// `lookup` is the caller's collision surface (MCP: default read scopes +
160/// write scope + shared; CLI: write scope + shared). `Ok(None)` means
161/// "create it" — covering both no-visible-match and a custom spec without
162/// the (Entity, name) equality index (`Rejected`), which degrades to
163/// create-always rather than failing the write.
164pub fn find_existing_entity(
165    db: &Db,
166    lookup: &ScopeSet,
167    name: &str,
168) -> Result<Option<NodeRecord>, TopoError> {
169    match resolve_entities_by_name(db, lookup, name) {
170        Ok(hits) => Ok(hits.into_iter().min_by_key(|n| n.id)),
171        Err(TopoError::Rejected(_)) => Ok(None),
172        Err(e) => Err(e),
173    }
174}
175
176/// The id of a Memory in `write_scope` whose normalized content equals
177/// `content`. Hash-bucket lookup, then exact normalized-content verify on
178/// every candidate; oldest id wins. Superseded or forgotten memories …
179/// are excluded — re-learning a retired fact is a NEW fact, and the
180/// tombstone's `as_of` history stays intact.
181pub fn existing_memory(
182    db: &Db,
183    write_scope: Scope,
184    content: &str,
185) -> Result<Option<NodeId>, TopoError> {
186    let hash = content_hash(content);
187    let want = normalize_content(content);
188    let scope_set = scopes_to_scope_set(&[write_scope]);
189    let candidates = db.nodes_by_prop(
190        &scope_set,
191        MEMORY_LABEL,
192        MEMORY_CONTENT_HASH_PROP,
193        &PropValue::Str(hash),
194    )?;
195    Ok(candidates
196        .into_iter()
197        .filter(|n| {
198            crate::MEMORY_TOMBSTONE_PROPS
199                .iter()
200                .all(|p| !n.props.contains_key(*p))
201                && matches!(n.props.get(MEMORY_CONTENT_PROP), Some(PropValue::Str(c)) if normalize_content(c) == want)
202        })
203        .min_by_key(|n| n.id)
204        .map(|n| n.id))
205}
206
207/// Ops marking `ids` superseded (stamp + close open out-edges) plus the ids
208/// actually marked. Moved from server.rs `supersede_ops`; `now_ms` is a
209/// parameter so tests are deterministic. Error strings must stay identical.
210/// Public building block for layers (e.g. topodb-obsidian) that supersede
211/// without going through `plan_remember`.
212pub fn plan_supersede(
213    db: &Db,
214    write_scope: Scope,
215    ids: &[String],
216    now_ms: i64,
217) -> Result<(Vec<Op>, Vec<String>), ComposeError> {
218    let mut ops = Vec::new();
219    let mut marked = Vec::new();
220    if ids.is_empty() {
221        return Ok((ops, marked));
222    }
223    let scope_set = scopes_to_scope_set(&[write_scope]);
224    let mut seen = BTreeSet::new();
225    for raw in ids {
226        let id: NodeId = raw
227            .parse()
228            .map_err(|e| ComposeError::Invalid(format!("invalid node id {raw:?}: {e}")))?;
229        if !seen.insert(id) {
230            continue;
231        }
232        let node = db.node(&scope_set, id).ok_or_else(|| {
233            ComposeError::Invalid(format!(
234                "supersedes id {raw} is not a node in the write scope"
235            ))
236        })?;
237        if node.label != MEMORY_LABEL {
238            return Err(ComposeError::Invalid(format!(
239                "supersedes id {raw} is a {}, not a Memory",
240                node.label
241            )));
242        }
243        if node.props.contains_key(MEMORY_SUPERSEDED_AT_PROP) {
244            continue;
245        }
246        let mut props: BTreeMap<String, Option<PropValue>> = BTreeMap::new();
247        props.insert(
248            MEMORY_SUPERSEDED_AT_PROP.into(),
249            Some(PropValue::Int(now_ms)),
250        );
251        ops.push(Op::SetNodeProps { id, props });
252        for e in db.edges_from(&scope_set, id, None, None, true, TimeAxis::Valid)? {
253            ops.push(Op::CloseEdge {
254                id: e.id,
255                valid_to: None,
256                superseded_at: None,
257            });
258        }
259        marked.push(id.to_string());
260    }
261    Ok((ops, marked))
262}
263
264/// Ops marking `ids` forgotten (stamp `forgotten_at` + close open out-edges)
265/// plus the forgotten ids. The disconnect motion is `plan_supersede`'s, but
266/// the contract is STRICTER: supersede is a bulk mark that skips
267/// already-retired ids; `forget` is an explicit judgment, so EVERY id must
268/// be a live Memory in the write scope and any violation rejects the whole
269/// call before ops build. `now_ms` is a parameter so tests are deterministic.
270pub fn plan_forget(
271    db: &Db,
272    write_scope: Scope,
273    ids: &[String],
274    now_ms: i64,
275) -> Result<(Vec<Op>, Vec<String>), ComposeError> {
276    if ids.is_empty() {
277        return Err(ComposeError::Invalid(
278            "forget requires at least one memory id".into(),
279        ));
280    }
281    let scope_set = scopes_to_scope_set(&[write_scope]);
282    let mut ops = Vec::new();
283    let mut forgotten = Vec::new();
284    let mut seen = BTreeSet::new();
285    for raw in ids {
286        let id: NodeId = raw
287            .parse()
288            .map_err(|e| ComposeError::Invalid(format!("invalid node id {raw:?}: {e}")))?;
289        if !seen.insert(id) {
290            continue;
291        }
292        let node = db.node(&scope_set, id).ok_or_else(|| {
293            ComposeError::Invalid(format!("forget id {raw} is not a node in the write scope"))
294        })?;
295        if node.label != MEMORY_LABEL {
296            return Err(ComposeError::Invalid(format!(
297                "forget id {raw} is a {}, not a Memory",
298                node.label
299            )));
300        }
301        if node.props.contains_key(MEMORY_FORGOTTEN_AT_PROP) {
302            return Err(ComposeError::Invalid(format!(
303                "forget id {raw} is already forgotten"
304            )));
305        }
306        if node.props.contains_key(MEMORY_SUPERSEDED_AT_PROP) {
307            return Err(ComposeError::Invalid(format!(
308                "forget id {raw} is already superseded — it has already left recall"
309            )));
310        }
311        let mut props: BTreeMap<String, Option<PropValue>> = BTreeMap::new();
312        props.insert(
313            MEMORY_FORGOTTEN_AT_PROP.into(),
314            Some(PropValue::Int(now_ms)),
315        );
316        ops.push(Op::SetNodeProps { id, props });
317        for e in db.edges_from(&scope_set, id, None, None, true, TimeAxis::Valid)? {
318            ops.push(Op::CloseEdge {
319                id: e.id,
320                valid_to: None,
321                superseded_at: None,
322            });
323        }
324        forgotten.push(id.to_string());
325    }
326    Ok((ops, forgotten))
327}
328
329// --- the composed verb ---
330
331pub struct RememberRequest {
332    pub content: String,
333    pub entities: Vec<String>,
334    pub edge_type: Option<String>,
335    pub supersedes: Vec<String>,
336    /// Extra memory metadata as a JSON object (same contract as
337    /// `merge_required_prop`'s `extra`).
338    pub props: Option<Value>,
339    /// Declared taxonomy kind for a NEW memory: `episodic`, `semantic`,
340    /// `procedural`, or `decision`. `None` = unstamped (reads as `semantic`).
341    /// Ignored on a dedup hit — the existing node's stored kind wins.
342    pub kind: Option<String>,
343}
344
345impl RememberRequest {
346    /// Input-only validation (no db access): the edge type normalizes and
347    /// at least one non-blank entity name is present. `plan_remember` runs
348    /// this itself; front ends call it FIRST when they must report input
349    /// errors ahead of scope/db errors (the pre-refactor precedence).
350    /// Returns the normalized edge type.
351    pub fn validate(&self) -> Result<String, String> {
352        let ty = normalize_edge_type(
353            self.edge_type
354                .as_deref()
355                .unwrap_or(DEFAULT_REMEMBER_EDGE_TYPE),
356        )?;
357        if self.entities.is_empty() {
358            return Err(
359                "entities must contain at least one name — use create_memory for a deliberately unlinked note".into(),
360            );
361        }
362        if self.entities.iter().any(|n| n.trim().is_empty()) {
363            return Err("entity names must be non-empty".into());
364        }
365        if let Some(kind) = &self.kind {
366            crate::validate_memory_kind(kind)?;
367        }
368        Ok(ty)
369    }
370}
371
372#[derive(Debug, Clone, PartialEq, Eq)]
373pub struct PlannedEntity {
374    pub name: String,
375    pub id: NodeId,
376    pub created: bool,
377}
378
379pub struct RememberPlan {
380    /// ONE atomic batch; possibly empty (pure no-op). The caller submits.
381    pub ops: Vec<Op>,
382    pub memory_id: NodeId,
383    /// True iff content was found in an existing node in the write scope.
384    /// Note: if that node is in `superseded`, deduplicated is still false —
385    /// superseding a node you would otherwise dedup to means "replace it".
386    pub deduplicated: bool,
387    /// The content, iff a new Memory node is planned (callers with an
388    /// embedder append `SetEmbedding` ops keyed on this).
389    pub new_memory: Option<String>,
390    /// (id, name) of every planned Entity create, for the same purpose.
391    pub new_entities: Vec<(NodeId, String)>,
392    pub entities: Vec<PlannedEntity>,
393    pub edge_ids: Vec<String>,
394    pub superseded: Vec<String>,
395}
396
397pub fn plan_remember(
398    db: &Db,
399    write_scope: Scope,
400    lookup: &ScopeSet,
401    now_ms: i64,
402    req: &RememberRequest,
403) -> Result<RememberPlan, ComposeError> {
404    let ty = req.validate().map_err(ComposeError::Invalid)?;
405
406    // Validate reserved keys BEFORE the dedup check (so reserved keys are always rejected).
407    let mut memory_props_result =
408        memory_props(&req.content, req.props.as_ref()).map_err(ComposeError::Invalid)?;
409    if let Some(kind) = &req.kind {
410        memory_props_result.insert(crate::MEMORY_KIND_PROP.into(), PropValue::Str(kind.clone()));
411    }
412
413    // Parse the supersedes list early to detect self-supersede.
414    // If the dedup hit is in the supersedes list, treat it as a fresh node creation.
415    // Error strings match plan_supersede's parse errors exactly.
416    let mut supersedes_ids = BTreeSet::new();
417    for raw in &req.supersedes {
418        let id: NodeId = raw
419            .parse()
420            .map_err(|e| ComposeError::Invalid(format!("invalid node id {raw:?}: {e}")))?;
421        supersedes_ids.insert(id);
422    }
423
424    let existing = existing_memory(db, write_scope, &req.content)?;
425    // If we found a dedup match but it's in the supersedes list, treat as NOT deduplicated.
426    let deduplicated = existing.is_some() && !supersedes_ids.contains(&existing.unwrap());
427    let memory_id = if deduplicated {
428        existing.unwrap()
429    } else {
430        NodeId::new()
431    };
432    let (supersede_ops, superseded) = plan_supersede(db, write_scope, &req.supersedes, now_ms)?;
433
434    struct Resolved {
435        name: String,
436        id: NodeId,
437        created: bool,
438        op: Option<Op>,
439    }
440    let mut seen = BTreeSet::new();
441    let mut resolved: Vec<Resolved> = Vec::new();
442    for name in req
443        .entities
444        .iter()
445        .filter(|n| seen.insert(entity_dedup_key(n)))
446    {
447        match find_existing_entity(db, lookup, name)? {
448            Some(node) => resolved.push(Resolved {
449                name: name.clone(),
450                id: node.id,
451                created: false,
452                op: None,
453            }),
454            None => {
455                let id = NodeId::new();
456                let props =
457                    merge_required_prop(ENTITY_NAME_PROP, PropValue::Str(name.clone()), None)
458                        .map_err(ComposeError::Invalid)?;
459                // UpsertNode, not CreateNode: the plan-time `find_existing_entity`
460                // above found nothing, but a CONCURRENT writer may create the same
461                // entity between now and apply. The applier resolves this upsert
462                // against committed state atomically — if the entity now exists it
463                // collapses onto it and remaps this id (and the edge below) — so
464                // concurrent writers never fragment the graph into duplicate
465                // entities. `apply_upsert_remap` (post-submit) rewrites this
466                // entity's reported id/created if the collapse happened.
467                resolved.push(Resolved {
468                    name: name.clone(),
469                    id,
470                    created: true,
471                    op: Some(Op::UpsertNode {
472                        id,
473                        scope: write_scope,
474                        label: ENTITY_LABEL.into(),
475                        key_prop: ENTITY_NAME_PROP.into(),
476                        props,
477                    }),
478                });
479            }
480        }
481    }
482
483    let mut ops: Vec<Op> = Vec::new();
484    let mut new_memory = None;
485    if !deduplicated {
486        ops.push(Op::CreateNode {
487            id: memory_id,
488            scope: write_scope,
489            label: MEMORY_LABEL.into(),
490            props: memory_props_result,
491        });
492        new_memory = Some(req.content.clone());
493    }
494
495    // Two names resolving to one node (e.g. via alias) collapse to one edge.
496    let mut seen_ids = BTreeSet::new();
497    resolved.retain(|r| seen_ids.insert(r.id));
498
499    // On a dedup hit, entities already linked keep their edge.
500    let already_linked: HashMap<NodeId, EdgeId> = if deduplicated {
501        let scope_set = scopes_to_scope_set(&[write_scope]);
502        db.edges_from(
503            &scope_set,
504            memory_id,
505            None,
506            Some(ty.as_str()),
507            true,
508            TimeAxis::Valid,
509        )?
510        .into_iter()
511        .map(|e| (e.to, e.id))
512        .collect()
513    } else {
514        HashMap::new()
515    };
516
517    let mut entities = Vec::with_capacity(resolved.len());
518    let mut edge_ids = Vec::with_capacity(resolved.len());
519    let mut new_entities = Vec::new();
520    for r in resolved {
521        if let Some(op) = r.op {
522            new_entities.push((r.id, r.name.clone()));
523            ops.push(op);
524        }
525        let edge_id = match already_linked.get(&r.id) {
526            Some(existing_edge) => existing_edge.to_string(),
527            None => {
528                let id = EdgeId::new();
529                ops.push(Op::CreateEdge {
530                    id,
531                    scope: write_scope,
532                    ty: ty.clone().into(),
533                    from: memory_id,
534                    to: r.id,
535                    props: Props::new(),
536                    valid_from: None,
537                    recorded_at: None,
538                });
539                id.to_string()
540            }
541        };
542        edge_ids.push(edge_id);
543        entities.push(PlannedEntity {
544            name: r.name,
545            id: r.id,
546            created: r.created,
547        });
548    }
549    ops.extend(supersede_ops);
550    Ok(RememberPlan {
551        ops,
552        memory_id,
553        deduplicated,
554        new_memory,
555        new_entities,
556        entities,
557        edge_ids,
558        superseded,
559    })
560}
561
562/// Correct planned entities against an applier's upsert remap.
563///
564/// A `plan_remember` entity that had to be CREATED was emitted as an
565/// `UpsertNode`; if a concurrent writer already created that entity, the applier
566/// collapsed onto the surviving node and reported `(planned_id -> surviving_id)`
567/// in `AppliedBatch::remap`. This rewrites such an entity's reported `id` to the
568/// surviving node and flips `created` to false (this call did not create it).
569/// Entities that were not remapped (this writer won the race, or the entity
570/// pre-existed) are untouched. No-op when `remap` is empty (the common case).
571pub fn apply_upsert_remap(entities: &mut [PlannedEntity], remap: &[(NodeId, NodeId)]) {
572    if remap.is_empty() {
573        return;
574    }
575    let map: HashMap<NodeId, NodeId> = remap.iter().copied().collect();
576    for e in entities.iter_mut() {
577        if let Some(&surviving) = map.get(&e.id) {
578            e.id = surviving;
579            e.created = false;
580        }
581    }
582}
583
584#[cfg(test)]
585mod tests {
586    use super::*;
587    use topodb::Op;
588
589    #[test]
590    fn apply_upsert_remap_rewrites_collapsed_entities_only() {
591        let planned_a = NodeId::new();
592        let surviving = NodeId::new();
593        let planned_b = NodeId::new();
594        let mut entities = vec![
595            PlannedEntity {
596                name: "A".into(),
597                id: planned_a,
598                created: true,
599            },
600            PlannedEntity {
601                name: "B".into(),
602                id: planned_b,
603                created: true,
604            },
605        ];
606        // Only A collapsed onto an existing node; B won its race.
607        apply_upsert_remap(&mut entities, &[(planned_a, surviving)]);
608        assert_eq!(
609            entities[0].id, surviving,
610            "collapsed entity takes surviving id"
611        );
612        assert!(
613            !entities[0].created,
614            "collapsed entity is not created by this call"
615        );
616        assert_eq!(entities[1].id, planned_b, "un-remapped entity is untouched");
617        assert!(entities[1].created, "un-remapped entity keeps created=true");
618        // Empty remap is a no-op.
619        let before = entities.clone();
620        apply_upsert_remap(&mut entities, &[]);
621        assert_eq!(entities, before);
622    }
623
624    #[test]
625    fn plan_supersede_is_public_and_stamps() {
626        let dir = tempfile::tempdir().unwrap();
627        let db = Db::open(dir.path().join("t.redb")).unwrap();
628        let id = NodeId::new();
629        db.submit(vec![Op::CreateNode {
630            id,
631            scope: Scope::Shared,
632            label: MEMORY_LABEL.into(),
633            props: memory_props("old fact", None).unwrap(),
634        }])
635        .unwrap();
636        let (ops, marked) = plan_supersede(&db, Scope::Shared, &[id.to_string()], 42).unwrap();
637        assert_eq!(marked, vec![id.to_string()]);
638        assert!(!ops.is_empty());
639    }
640}
topodb_json/compose.rs

topodb_json/
compose.rs
