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::{Db, EdgeId, NodeId, NodeRecord, Op, PropValue, Props, Scope, ScopeSet, TopoError};
11
12use crate::{
13    merge_required_prop, normalize_edge_type, scopes_to_scope_set, ALIAS_EDGE_TYPE, ALIAS_LABEL,
14    ALIAS_NAME_PROP, ENTITY_LABEL, ENTITY_NAME_PROP, MEMORY_CONTENT_HASH_PROP, MEMORY_CONTENT_PROP,
15    MEMORY_LABEL, MEMORY_SUPERSEDED_AT_PROP,
16};
17
18/// Edge type `remember` uses when the caller doesn't name one.
19pub const DEFAULT_REMEMBER_EDGE_TYPE: &str = "about";
20
21/// Planning failure: `Invalid` is a caller-fixable input problem (surface the
22/// message verbatim); `Engine` is a database failure unrelated to the input.
23#[derive(Debug)]
24pub enum ComposeError {
25    Invalid(String),
26    Engine(TopoError),
27}
28
29impl From<TopoError> for ComposeError {
30    fn from(e: TopoError) -> Self {
31        ComposeError::Engine(e)
32    }
33}
34
35// --- moved verbatim from topodb-mcp/src/server.rs (self.db -> db) ---
36// Keep the original doc comments from server.rs on each of these when moving.
37
38/// In-call dedup key for `remember`'s entity names: whitespace-collapsed,
39/// lowercased — mirroring the engine's prop-index normalization
40/// (`prop_index::normalize_str`, which is pub(crate) and thus can't be
41/// called from here). Drift between the two only weakens IN-CALL dedup
42/// (["Drew", "drew"] in one call); cross-call dedup always goes through the
43/// engine's own normalized index via find_existing_entity.
44pub fn entity_dedup_key(name: &str) -> String {
45    name.split_whitespace()
46        .collect::<Vec<_>>()
47        .join(" ")
48        .to_lowercase()
49}
50
51/// Normalize memory content for dedup: trim and collapse internal whitespace.
52/// Deliberately NOT lowercased — casing can carry meaning in a stored fact.
53pub fn normalize_content(content: &str) -> String {
54    content.split_whitespace().collect::<Vec<_>>().join(" ")
55}
56
57/// Stable FNV-1a 64-bit hash of normalized content, hex-encoded. PERSISTED
58/// (equality-indexed as `content_hash`) — the algorithm must never change.
59/// Collisions are harmless: dedup always verifies exact normalized content.
60pub fn content_hash(content: &str) -> String {
61    let normalized = normalize_content(content);
62    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
63    for b in normalized.as_bytes() {
64        h ^= *b as u64;
65        h = h.wrapping_mul(0x0000_0100_0000_01b3);
66    }
67    format!("{h:016x}")
68}
69
70/// Builds a new Memory node's props: caller `extra` is validated (the
71/// system-maintained keys below are rejected, and `content` collides via
72/// merge_required_prop), then the whitespace-normalized content hash is
73/// stamped. The ONE constructor for every front end's new-memory write
74/// (plan_remember, MCP create_memory, CLI create-memory) — so the reserved
75/// set cannot drift between surfaces.
76pub fn memory_props(content: &str, extra: Option<&Value>) -> Result<Props, String> {
77    if let Some(Value::Object(map)) = extra {
78        for reserved in [MEMORY_CONTENT_HASH_PROP, MEMORY_SUPERSEDED_AT_PROP] {
79            if map.contains_key(reserved) {
80                return Err(format!(
81                    "props must not include {reserved:?}: it is maintained by the engine write path"
82                ));
83            }
84        }
85    }
86    let mut props = merge_required_prop(
87        MEMORY_CONTENT_PROP,
88        PropValue::Str(content.to_string()),
89        extra,
90    )?;
91    props.insert(
92        MEMORY_CONTENT_HASH_PROP.into(),
93        PropValue::Str(content_hash(content)),
94    );
95    Ok(props)
96}
97
98/// Canonical entities for `name`: direct (Entity, name) matches plus
99/// (Alias, name) matches followed through alias_of. Deduped by id,
100/// oldest first.
101///
102/// Returns the raw `TopoError` (not `ErrorData`) rather than swallowing
103/// it: the two existing call sites disagree on what an undeclared
104/// (Entity, name) index should mean. `find_by_prop` must still surface it
105/// as a caller error — that is the exact contract tests pin down (an
106/// undeclared-index probe on a custom spec must error, not silently return
107/// empty, or a clobbered spec reopen would go undetected). `create_entity`
108/// instead treats it as "can't dedup on this spec" and degrades to
109/// create-always. Only the (Alias, name) probe's `Rejected` is
110/// unconditionally swallowed here — a spec that predates the Alias index
111/// (or a custom spec that never declared it) simply has no aliases to
112/// resolve, which is never a caller error.
113pub fn resolve_entities_by_name(
114    db: &Db,
115    scopes: &ScopeSet,
116    name: &str,
117) -> Result<Vec<NodeRecord>, TopoError> {
118    let value = PropValue::Str(name.to_string());
119    let mut out = db.nodes_by_prop_normalized(scopes, ENTITY_LABEL, ENTITY_NAME_PROP, &value)?;
120    let aliases = match db.nodes_by_prop_normalized(scopes, ALIAS_LABEL, ALIAS_NAME_PROP, &value) {
121        Ok(hits) => hits,
122        Err(TopoError::Rejected(_)) => Vec::new(),
123        Err(e) => return Err(e),
124    };
125    for alias in aliases {
126        for edge in db.edges_from(scopes, alias.id, None, Some(ALIAS_EDGE_TYPE), true)? {
127            if let Some(canonical) = db.node(scopes, edge.to) {
128                if canonical.label == ENTITY_LABEL {
129                    out.push(canonical);
130                }
131            }
132        }
133    }
134    out.sort_by_key(|n| n.id);
135    out.dedup_by_key(|n| n.id);
136    Ok(out)
137}
138
139/// `lookup` is the caller's collision surface (MCP: default read scopes +
140/// write scope + shared; CLI: write scope + shared). `Ok(None)` means
141/// "create it" — covering both no-visible-match and a custom spec without
142/// the (Entity, name) equality index (`Rejected`), which degrades to
143/// create-always rather than failing the write.
144pub fn find_existing_entity(
145    db: &Db,
146    lookup: &ScopeSet,
147    name: &str,
148) -> Result<Option<NodeRecord>, TopoError> {
149    match resolve_entities_by_name(db, lookup, name) {
150        Ok(hits) => Ok(hits.into_iter().min_by_key(|n| n.id)),
151        Err(TopoError::Rejected(_)) => Ok(None),
152        Err(e) => Err(e),
153    }
154}
155
156/// The id of a Memory in `write_scope` whose normalized content equals
157/// `content`. Hash-bucket lookup, then exact normalized-content verify on
158/// every candidate; oldest id wins. Superseded memories (those with a
159/// `superseded_at` timestamp) are excluded — re-learning a retired fact is
160/// a NEW fact, and the tombstone's `as_of` history stays intact.
161pub fn existing_memory(
162    db: &Db,
163    write_scope: Scope,
164    content: &str,
165) -> Result<Option<NodeId>, TopoError> {
166    let hash = content_hash(content);
167    let want = normalize_content(content);
168    let scope_set = scopes_to_scope_set(&[write_scope]);
169    let candidates = db.nodes_by_prop(
170        &scope_set,
171        MEMORY_LABEL,
172        MEMORY_CONTENT_HASH_PROP,
173        &PropValue::Str(hash),
174    )?;
175    Ok(candidates
176        .into_iter()
177        .filter(|n| {
178            !n.props.contains_key(MEMORY_SUPERSEDED_AT_PROP)
179                && matches!(n.props.get(MEMORY_CONTENT_PROP), Some(PropValue::Str(c)) if normalize_content(c) == want)
180        })
181        .min_by_key(|n| n.id)
182        .map(|n| n.id))
183}
184
185/// Ops marking `ids` superseded (stamp + close open out-edges) plus the ids
186/// actually marked. Moved from server.rs `supersede_ops`; `now_ms` is a
187/// parameter so tests are deterministic. Error strings must stay identical.
188fn plan_supersede(
189    db: &Db,
190    write_scope: Scope,
191    ids: &[String],
192    now_ms: i64,
193) -> Result<(Vec<Op>, Vec<String>), ComposeError> {
194    let mut ops = Vec::new();
195    let mut marked = Vec::new();
196    if ids.is_empty() {
197        return Ok((ops, marked));
198    }
199    let scope_set = scopes_to_scope_set(&[write_scope]);
200    let mut seen = BTreeSet::new();
201    for raw in ids {
202        let id: NodeId = raw
203            .parse()
204            .map_err(|e| ComposeError::Invalid(format!("invalid node id {raw:?}: {e}")))?;
205        if !seen.insert(id) {
206            continue;
207        }
208        let node = db.node(&scope_set, id).ok_or_else(|| {
209            ComposeError::Invalid(format!(
210                "supersedes id {raw} is not a node in the write scope"
211            ))
212        })?;
213        if node.label != MEMORY_LABEL {
214            return Err(ComposeError::Invalid(format!(
215                "supersedes id {raw} is a {}, not a Memory",
216                node.label
217            )));
218        }
219        if node.props.contains_key(MEMORY_SUPERSEDED_AT_PROP) {
220            continue;
221        }
222        let mut props: BTreeMap<String, Option<PropValue>> = BTreeMap::new();
223        props.insert(
224            MEMORY_SUPERSEDED_AT_PROP.into(),
225            Some(PropValue::Int(now_ms)),
226        );
227        ops.push(Op::SetNodeProps { id, props });
228        for e in db.edges_from(&scope_set, id, None, None, true)? {
229            ops.push(Op::CloseEdge {
230                id: e.id,
231                valid_to: None,
232            });
233        }
234        marked.push(id.to_string());
235    }
236    Ok((ops, marked))
237}
238
239// --- the composed verb ---
240
241pub struct RememberRequest {
242    pub content: String,
243    pub entities: Vec<String>,
244    pub edge_type: Option<String>,
245    pub supersedes: Vec<String>,
246    /// Extra memory metadata as a JSON object (same contract as
247    /// `merge_required_prop`'s `extra`).
248    pub props: Option<Value>,
249}
250
251impl RememberRequest {
252    /// Input-only validation (no db access): the edge type normalizes and
253    /// at least one non-blank entity name is present. `plan_remember` runs
254    /// this itself; front ends call it FIRST when they must report input
255    /// errors ahead of scope/db errors (the pre-refactor precedence).
256    /// Returns the normalized edge type.
257    pub fn validate(&self) -> Result<String, String> {
258        let ty = normalize_edge_type(
259            self.edge_type
260                .as_deref()
261                .unwrap_or(DEFAULT_REMEMBER_EDGE_TYPE),
262        )?;
263        if self.entities.is_empty() {
264            return Err(
265                "entities must contain at least one name — use create_memory for a deliberately unlinked note".into(),
266            );
267        }
268        if self.entities.iter().any(|n| n.trim().is_empty()) {
269            return Err("entity names must be non-empty".into());
270        }
271        Ok(ty)
272    }
273}
274
275pub struct PlannedEntity {
276    pub name: String,
277    pub id: NodeId,
278    pub created: bool,
279}
280
281pub struct RememberPlan {
282    /// ONE atomic batch; possibly empty (pure no-op). The caller submits.
283    pub ops: Vec<Op>,
284    pub memory_id: NodeId,
285    /// True iff content was found in an existing node in the write scope.
286    /// Note: if that node is in `superseded`, deduplicated is still false —
287    /// superseding a node you would otherwise dedup to means "replace it".
288    pub deduplicated: bool,
289    /// The content, iff a new Memory node is planned (callers with an
290    /// embedder append `SetEmbedding` ops keyed on this).
291    pub new_memory: Option<String>,
292    /// (id, name) of every planned Entity create, for the same purpose.
293    pub new_entities: Vec<(NodeId, String)>,
294    pub entities: Vec<PlannedEntity>,
295    pub edge_ids: Vec<String>,
296    pub superseded: Vec<String>,
297}
298
299pub fn plan_remember(
300    db: &Db,
301    write_scope: Scope,
302    lookup: &ScopeSet,
303    now_ms: i64,
304    req: &RememberRequest,
305) -> Result<RememberPlan, ComposeError> {
306    let ty = req.validate().map_err(ComposeError::Invalid)?;
307
308    // Validate reserved keys BEFORE the dedup check (so reserved keys are always rejected).
309    let memory_props_result =
310        memory_props(&req.content, req.props.as_ref()).map_err(ComposeError::Invalid)?;
311
312    // Parse the supersedes list early to detect self-supersede.
313    // If the dedup hit is in the supersedes list, treat it as a fresh node creation.
314    // Error strings match plan_supersede's parse errors exactly.
315    let mut supersedes_ids = BTreeSet::new();
316    for raw in &req.supersedes {
317        let id: NodeId = raw
318            .parse()
319            .map_err(|e| ComposeError::Invalid(format!("invalid node id {raw:?}: {e}")))?;
320        supersedes_ids.insert(id);
321    }
322
323    let existing = existing_memory(db, write_scope, &req.content)?;
324    // If we found a dedup match but it's in the supersedes list, treat as NOT deduplicated.
325    let deduplicated = existing.is_some() && !supersedes_ids.contains(&existing.unwrap());
326    let memory_id = if deduplicated {
327        existing.unwrap()
328    } else {
329        NodeId::new()
330    };
331    let (supersede_ops, superseded) = plan_supersede(db, write_scope, &req.supersedes, now_ms)?;
332
333    struct Resolved {
334        name: String,
335        id: NodeId,
336        created: bool,
337        op: Option<Op>,
338    }
339    let mut seen = BTreeSet::new();
340    let mut resolved: Vec<Resolved> = Vec::new();
341    for name in req
342        .entities
343        .iter()
344        .filter(|n| seen.insert(entity_dedup_key(n)))
345    {
346        match find_existing_entity(db, lookup, name)? {
347            Some(node) => resolved.push(Resolved {
348                name: name.clone(),
349                id: node.id,
350                created: false,
351                op: None,
352            }),
353            None => {
354                let id = NodeId::new();
355                let props =
356                    merge_required_prop(ENTITY_NAME_PROP, PropValue::Str(name.clone()), None)
357                        .map_err(ComposeError::Invalid)?;
358                resolved.push(Resolved {
359                    name: name.clone(),
360                    id,
361                    created: true,
362                    op: Some(Op::CreateNode {
363                        id,
364                        scope: write_scope,
365                        label: ENTITY_LABEL.into(),
366                        props,
367                    }),
368                });
369            }
370        }
371    }
372
373    let mut ops: Vec<Op> = Vec::new();
374    let mut new_memory = None;
375    if !deduplicated {
376        ops.push(Op::CreateNode {
377            id: memory_id,
378            scope: write_scope,
379            label: MEMORY_LABEL.into(),
380            props: memory_props_result,
381        });
382        new_memory = Some(req.content.clone());
383    }
384
385    // Two names resolving to one node (e.g. via alias) collapse to one edge.
386    let mut seen_ids = BTreeSet::new();
387    resolved.retain(|r| seen_ids.insert(r.id));
388
389    // On a dedup hit, entities already linked keep their edge.
390    let already_linked: HashMap<NodeId, EdgeId> = if deduplicated {
391        let scope_set = scopes_to_scope_set(&[write_scope]);
392        db.edges_from(&scope_set, memory_id, None, Some(ty.as_str()), true)?
393            .into_iter()
394            .map(|e| (e.to, e.id))
395            .collect()
396    } else {
397        HashMap::new()
398    };
399
400    let mut entities = Vec::with_capacity(resolved.len());
401    let mut edge_ids = Vec::with_capacity(resolved.len());
402    let mut new_entities = Vec::new();
403    for r in resolved {
404        if let Some(op) = r.op {
405            new_entities.push((r.id, r.name.clone()));
406            ops.push(op);
407        }
408        let edge_id = match already_linked.get(&r.id) {
409            Some(existing_edge) => existing_edge.to_string(),
410            None => {
411                let id = EdgeId::new();
412                ops.push(Op::CreateEdge {
413                    id,
414                    scope: write_scope,
415                    ty: ty.clone().into(),
416                    from: memory_id,
417                    to: r.id,
418                    props: Props::new(),
419                    valid_from: None,
420                });
421                id.to_string()
422            }
423        };
424        edge_ids.push(edge_id);
425        entities.push(PlannedEntity {
426            name: r.name,
427            id: r.id,
428            created: r.created,
429        });
430    }
431    ops.extend(supersede_ops);
432    Ok(RememberPlan {
433        ops,
434        memory_id,
435        deduplicated,
436        new_memory,
437        new_entities,
438        entities,
439        edge_ids,
440        superseded,
441    })
442}
topodb_json/compose.rs

topodb_json/
compose.rs
