memstead_base/engine/drift.rs
1//! Drift detection and per-mem change synthesis.
2//!
3//! `reload_if_stale` probes each candidate mount's `current_head()`
4//! cursor on every operation (no throttle), reloads mems whose
5//! on-disk state has advanced past the engine's cached head, and
6//! surfaces `MemReloaded` warnings for handlers that need to
7//! re-derive conclusions from a now-reloaded snapshot. `changes_since`
8//! produces
9//! the per-entity diff between a stored cursor and the backend's
10//! current state — folder mounts synthesise from the changelog, the
11//! git-branch hook walks the tree with rename detection, archive
12//! mounts return empty.
13
14use crate::backend::BackendError;
15use crate::workspace::MountStorage;
16
17use super::mutation::lookup_title_and_type;
18use super::{Engine, EngineError};
19
20impl Engine {
21 /// Reload-before-operation: before any read or write executes,
22 /// check the mem ref; if it advanced past the engine's cached
23 /// `last_known_head`, reload the affected mem(s) and return one
24 /// [`WarningHint::MemReloaded`] per reload so the caller can
25 /// surface the drift to the agent (the response *itself* already
26 /// carries fresh content — the warning explains why state
27 /// shifted).
28 ///
29 /// The ref check runs on **every** call — there is no throttle
30 /// window. A per-operation `current_head()` read is microseconds,
31 /// effectively free at LLM latencies, and a throttle that let an
32 /// operation execute against an already-moved ref would reintroduce
33 /// the exact silent-staleness this guards against. This is the
34 /// correctness floor: no operation acts on a projection that is
35 /// behind git truth.
36 ///
37 /// `mem = Some(name)` scopes the probe to one mount; `None`
38 /// scans every mount. Read handlers that target a known mem
39 /// (`memstead_entity` derives the mem from the id;
40 /// `memstead_changes_since` takes it as a param) and every mutation
41 /// (which knows its target mem) pass a name; tools that scan
42 /// multi-mem (`memstead_search` without a mem filter,
43 /// `memstead_overview`, `memstead_health`) pass `None`.
44 ///
45 /// Behaviour matrix per mount:
46 /// - cached `Some(old)` + on-disk `Some(new)`, `old != new` →
47 /// reload the mem, emit `MemReloaded`, refresh the cached
48 /// head to `new`.
49 /// - cached `None` + on-disk `Some(new)` → silently capture the
50 /// first observed head as the baseline (no warning — there's
51 /// no prior in-memory snapshot to be stale against).
52 /// - cached / on-disk match, on-disk `None` (folder, archive,
53 /// refdb hiccup), or `current_head` errors → no-op.
54 ///
55 /// Reload errors are warn-logged and the affected mem is
56 /// skipped — the caller's response is still served from the (now
57 /// stale) in-memory snapshot rather than failing the entire
58 /// request. The next operation retries.
59 ///
60 /// Cache invalidation rides on `reload_one_mem` — community
61 /// and search-index memos drop when any mem reloads.
62 pub fn reload_if_stale(&mut self, mem: Option<&str>) -> Vec<crate::ops::WarningHint> {
63 let mut warnings = Vec::new();
64 // Phase -1 — membership. The roster is probed before content: a
65 // mem that left must not be loaded or served, a mem that arrived
66 // must be there for the operation that follows (roster.rs).
67 match self.reconcile_roster() {
68 Ok(Some(change)) if !change.is_empty() => {
69 warnings.push(crate::ops::WarningHint::MemRosterChanged {
70 added: change.added,
71 removed: change.removed,
72 quarantined: change.quarantined,
73 failures: change
74 .failures
75 .iter()
76 .map(|f| format!("{}: {}", f.item, f.error))
77 .collect(),
78 });
79 }
80 Ok(_) => {}
81 Err(e) => {
82 warnings.push(crate::ops::WarningHint::MemRosterChanged {
83 added: Vec::new(),
84 removed: Vec::new(),
85 quarantined: Vec::new(),
86 failures: vec![e.to_string()],
87 });
88 }
89 }
90 warnings.extend(self.reload_if_stale_content_only(mem));
91 warnings
92 }
93
94 /// The content half of [`Self::reload_if_stale`] alone — the branch-tip
95 /// probe without the roster probe. Exposed for the cost comparison the
96 /// roster probe is held to; every operation goes through the full form.
97 pub fn reload_if_stale_content_only(
98 &mut self,
99 mem: Option<&str>,
100 ) -> Vec<crate::ops::WarningHint> {
101 let mut warnings = Vec::new();
102 // Phase 0 — the lazy-mount first-read trigger. Every operation
103 // funnels through this check, so a deferred mem the operation's
104 // scope touches loads here, before the staleness probes — a
105 // scoped operation loads exactly its mem, a workspace-scoped one
106 // loads every deferred mem so no answer is computed over a
107 // partial store. Operations scoped to eager mems never load a
108 // lazy sibling as a side effect.
109 self.ensure_mems_loaded(mem);
110
111 // Phase 1 — pick candidate mem names that match the filter.
112 // Cloned so the immutable borrow doesn't survive into the
113 // mutation phase.
114 let candidates: Vec<String> = self
115 .mounts
116 .iter()
117 .filter(|m| mem.is_none_or(|v| m.mount.mem == v))
118 .map(|m| m.mount.mem.clone())
119 .collect();
120
121 if candidates.is_empty() {
122 return warnings;
123 }
124
125 // Phase 2 — probe every candidate's current head via the
126 // backend. Errors collapse to None so a transient backend
127 // hiccup doesn't surface as a warning; the next operation
128 // retries.
129 let probes: Vec<(String, Option<String>, Option<String>)> = candidates
130 .iter()
131 .filter_map(|name| {
132 let m = self.mounts.iter().find(|m| &m.mount.mem == name)?;
133 let new_head = m.backend.current_head().ok().flatten();
134 let cached = m.last_known_head.clone();
135 Some((name.clone(), cached, new_head))
136 })
137 .collect();
138
139 // Phase 3 — act on each probe. The drift case is the only
140 // one that calls `reload_one_mem`; every other arm just
141 // (for first-observation) captures the baseline head silently.
142 for (name, cached, new_head) in probes {
143 match (cached, new_head.clone()) {
144 (Some(old), Some(new)) if old != new => {
145 match self.reload_one_mem(&name) {
146 Ok(report) => {
147 warnings.push(crate::ops::WarningHint::MemReloaded {
148 mem: name.clone(),
149 old_head: old.clone(),
150 new_head: new.clone(),
151 entities_loaded: report.added.len() + report.changed.len(),
152 });
153 if let Some(state) =
154 self.mounts.iter_mut().find(|m| m.mount.mem == name)
155 {
156 state.last_known_head = Some(new.clone());
157 }
158 // Build the structured notice now — the
159 // backend's current head equals `new` and no
160 // follow-on write in this operation has
161 // committed yet, so the `old → new` delta
162 // describes only the sibling's change. Stashed
163 // for the response layer to drain.
164 let notice = self.mem_changed_notice(&name, &old, &new);
165 self.pending_mem_changed.push(notice);
166 // Emit the same change on the mem-change
167 // event channel: subscribers (SSE forwarders
168 // foremost) previously saw only this engine's
169 // own writes — a sibling process's commit,
170 // detected here as drift, is every bit as
171 // much a change. `n_commits: 1` per the
172 // watcher precedent (events batch by
173 // detection, not by commit archaeology).
174 self.emit_mem_changed(&crate::engine::events::MemChangedEvent {
175 mem: name.clone(),
176 head: new.clone(),
177 previous: old.clone(),
178 n_commits: 1,
179 });
180 }
181 Err(e) => {
182 tracing::warn!(
183 mem = %name,
184 error = %e,
185 "drift-detected reload_one_mem failed; serving \
186 stale snapshot — will retry on the next operation"
187 );
188 }
189 }
190 }
191 _ => {
192 if let Some(state) = self.mounts.iter_mut().find(|m| m.mount.mem == name)
193 && state.last_known_head.is_none()
194 {
195 state.last_known_head = new_head;
196 }
197 }
198 }
199 }
200
201 warnings
202 }
203
204 /// Mark `mount_idx`'s on-disk head as advanced by *this* engine's
205 /// own write so the next `reload_if_stale` doesn't surface
206 /// `MEM_RELOADED` for the commit we just produced. Mutation
207 /// paths call this immediately after `backend.commit` returns —
208 /// the cached `last_known_head` jumps straight to the new SHA
209 /// without going through a reload. Because every mutation runs
210 /// `reload_if_stale` for its target mem *before* committing,
211 /// the cached `last_known_head` is current at commit time, so
212 /// this advance is over a verified parent — it can never jump
213 /// the cache past an unobserved sibling commit. Cross-session and
214 /// out-of-band advances (sibling engine, manual `git pull`) that
215 /// land before the next operation still mismatch the cached value
216 /// and fire the warning as before.
217 ///
218 /// Empty SHA is a no-op (no commit landed — e.g. duplicate-add
219 /// relate). Backends that don't track a head (folder, archive)
220 /// leave `last_known_head` at `None` and still no-op via the
221 /// drift-check's `cached: None` branch.
222 pub(crate) fn record_self_write(&mut self, mount_idx: usize, write_id: &str) {
223 if write_id.is_empty() {
224 return;
225 }
226 // Capture the pre-write head + mem name so the
227 // `MemChangedEvent` we emit reflects the transition the
228 // current commit produced. We do this before mutating
229 // `last_known_head` because that field is the previous SHA
230 // from the event's point of view.
231 let (mem, previous) = match self.mounts.get(mount_idx) {
232 Some(state) => (
233 state.mount.mem.clone(),
234 state.last_known_head.clone().unwrap_or_default(),
235 ),
236 None => return,
237 };
238 // The recorded head must equal what the backend's next
239 // `current_head()` probe will report, or every self-write
240 // would look like sibling drift on the following operation.
241 // For git-branch backends the probe returns exactly the commit
242 // SHA just produced; the folder backend's drift cursor is the
243 // changelog's last-line timestamp (a different dialect from
244 // its synthetic commit id), so probe once and prefer the
245 // backend's answer. Probe errors fall back to the commit id —
246 // drift detection stays best-effort, never blocking the write.
247 let recorded = self
248 .mounts
249 .get(mount_idx)
250 .and_then(|state| state.backend.current_head().ok().flatten())
251 .unwrap_or_else(|| write_id.to_string());
252 if let Some(state) = self.mounts.get_mut(mount_idx) {
253 state.last_known_head = Some(recorded.clone());
254 }
255 // Skip emit when no SHA actually advanced — folder backends
256 // (and archive backends) carry `last_known_head: None` and
257 // pass `write_id = ""` in some paths; the early-return at
258 // the top already catches the explicit empty case, but
259 // `previous == write_id` covers idempotent re-writes that
260 // pass through the same write path (e.g. a relate that
261 // re-applies the same edge). Skipping keeps the event stream
262 // a stream of *changes* rather than a stream of *writes*.
263 if previous == recorded {
264 return;
265 }
266 let event = crate::engine::events::MemChangedEvent {
267 mem,
268 // The corrected head, not the raw commit id: consumers feed
269 // event heads into `changes_since`, whose folder dialect is
270 // the changelog-timestamp cursor `recorded` carries.
271 head: recorded,
272 previous,
273 n_commits: 1,
274 };
275 self.emit_mem_changed(&event);
276 }
277
278 /// Drain the reload-before-operation notices accumulated since the
279 /// last drain. The response layer calls this after an operation
280 /// completes to attach the structured `mem_changed` notice. Every
281 /// handler that can trigger a reload (directly via
282 /// [`Self::reload_if_stale`] or indirectly through a mutation) must
283 /// drain, or an undrained notice leaks into the next operation's
284 /// response.
285 pub fn take_mem_changed_notices(&mut self) -> Vec<crate::ops::MemChangedNotice> {
286 std::mem::take(&mut self.pending_mem_changed)
287 }
288
289 /// Build a [`crate::ops::MemChangedNotice`] describing the
290 /// per-entity delta a reload applied to `mem` (from `from_head`
291 /// to `to_head`). Derived from [`Self::changes_since`] so it
292 /// carries rename detection on git-branch mounts; on any backend
293 /// error (e.g. an unresolvable cursor) it falls back to an empty
294 /// delta — the heads alone still tell the agent the mem moved.
295 ///
296 /// Callers pair this with [`Self::reload_if_stale`]: a returned
297 /// [`crate::ops::WarningHint::MemReloaded`] carries the
298 /// `old_head` / `new_head` to pass here. The delta matches the
299 /// transition the reload applied (`changes_since` walks the same
300 /// `from_head → current` range).
301 pub fn mem_changed_notice(
302 &self,
303 mem: &str,
304 from_head: &str,
305 to_head: &str,
306 ) -> crate::ops::MemChangedNotice {
307 let changes = self
308 .changes_since(mem, from_head, None)
309 .map(|r| r.changes)
310 .unwrap_or_default();
311 crate::ops::MemChangedNotice::from_delta(
312 mem.to_string(),
313 from_head.to_string(),
314 to_head.to_string(),
315 changes,
316 )
317 }
318
319 /// Per-entity events for `mem` between `since` and the backend's
320 /// current state.
321 ///
322 /// 1. Resolves the mount (returns [`EngineError::UnknownMem`]
323 /// on unknown mem).
324 /// 2. Validates `rename_similarity` against
325 /// `[RENAME_SIMILARITY_MIN, RENAME_SIMILARITY_MAX]`. Out-of-range
326 /// values refuse with [`EngineError::InvalidInput`] carrying
327 /// `details.allowed_range` and `details.requested`. `None` falls
328 /// back to [`crate::ops::RENAME_SIMILARITY_DEFAULT`].
329 /// 3. Dispatches on the mount's `MountStorage`:
330 /// - Folder mounts synthesize from the JSONL changelog via
331 /// [`crate::ops::folder_changes_since`].
332 /// - Git-branch mounts call the registered
333 /// [`GitBranchOps::changes_since`] hook (real tree-diff with
334 /// rename detection); missing hook = full flavour not loaded
335 /// and the report comes back empty.
336 /// - Archive mounts return an empty report.
337 /// 4. Enriches each envelope's `title` / `entity_type` from the
338 /// in-memory store (best-effort — `Removed` envelopes always
339 /// leave both `None`; missing-from-store entities also leave
340 /// them `None`).
341 /// 5. Returns [`crate::ops::ChangesReport`] with `mem`,
342 /// `since` (echoed), `head` (backend-resolved current
343 /// cursor), enriched `changes`, and any clamping warnings.
344 pub fn changes_since(
345 &self,
346 mem: &str,
347 since: &str,
348 rename_similarity: Option<f32>,
349 ) -> Result<crate::ops::ChangesReport, EngineError> {
350 let m = self.find_mount(mem)?;
351
352 // Reject out-of-range `rename_similarity` early so CLI and MCP
353 // share one refusal surface.
354 // The prior clamp+warn shape silently accepted nonsense values
355 // (e.g. 1.5 ≡ 1.0); typed refusal gives the agent a recoverable
356 // signal.
357 if let Some(v) = rename_similarity
358 && !(crate::ops::RENAME_SIMILARITY_MIN..=crate::ops::RENAME_SIMILARITY_MAX).contains(&v)
359 {
360 return Err(EngineError::RenameSimilarityOutOfRange {
361 requested: v,
362 allowed_min: crate::ops::RENAME_SIMILARITY_MIN,
363 allowed_max: crate::ops::RENAME_SIMILARITY_MAX,
364 });
365 }
366 let clamped = rename_similarity.unwrap_or(crate::ops::RENAME_SIMILARITY_DEFAULT);
367
368 let backend_changes = match &m.mount.storage {
369 MountStorage::Folder { path } => {
370 match crate::ops::folder_changes_since(path, mem, since) {
371 Ok(c) => c,
372 // Lift the backend's typed bad-`since` marker, parallel
373 // to the git arm's COMMIT_NOT_FOUND lift below: the
374 // folder cursor is a timestamp, and anything else —
375 // a mutation's `write_id` above all — refuses instead
376 // of silently replaying the whole history.
377 Err(BackendError::Other(msg)) if msg.starts_with("INVALID_TS_CURSOR:") => {
378 let since = msg
379 .strip_prefix("INVALID_TS_CURSOR:")
380 .unwrap_or_default()
381 .to_string();
382 return Err(EngineError::InvalidTimestampCursor {
383 mem: mem.to_string(),
384 since,
385 });
386 }
387 Err(e) => return Err(EngineError::Backend(e)),
388 }
389 }
390 MountStorage::GitBranch { gitdir, branch } => match self.git_branch_ops.as_ref() {
391 Some(hook) => match (hook.changes_since)(gitdir, branch, mem, since, clamped) {
392 Ok(c) => c,
393 // Lift the backend's typed bad-`since` marker to a typed
394 // engine error carrying the untruncated SHA, parallel
395 // to the UNKNOWN_REMOTE / LOCAL_DIVERGENCE prefixes.
396 Err(BackendError::Other(msg)) if msg.starts_with("COMMIT_NOT_FOUND:") => {
397 let since = msg
398 .strip_prefix("COMMIT_NOT_FOUND:")
399 .unwrap_or_default()
400 .to_string();
401 return Err(EngineError::InvalidChangesCursor {
402 mem: mem.to_string(),
403 since,
404 });
405 }
406 Err(e) => return Err(EngineError::Backend(e)),
407 },
408 None => crate::ops::BackendChanges::empty_at(since),
409 },
410 // Archive is sealed; the in-memory backend keeps a
411 // provenance log but no cursor-addressable change history,
412 // so both yield no backend-derived changes here (the live
413 // playground stream rides the engine's event broadcast, not
414 // this path).
415 MountStorage::Archive { .. } | MountStorage::InMemory => {
416 crate::ops::BackendChanges::empty_at(since)
417 }
418 };
419
420 // Enrich each id-only envelope from the engine's store.
421 // `Removed` always leaves title / entity_type None — the
422 // entity is gone by definition; the post-reload store does
423 // not have it. Other variants populate when the lookup
424 // succeeds; missing ids stay None.
425 let enriched: Vec<crate::ops::ChangeEnvelope> = backend_changes
426 .changes
427 .into_iter()
428 .map(|env| match env {
429 crate::ops::ChangeEnvelope::Added { id, .. } => {
430 let (title, entity_type) = lookup_title_and_type(&self.store, &id);
431 crate::ops::ChangeEnvelope::Added {
432 id,
433 title,
434 entity_type,
435 }
436 }
437 crate::ops::ChangeEnvelope::Updated { id, .. } => {
438 let (title, entity_type) = lookup_title_and_type(&self.store, &id);
439 crate::ops::ChangeEnvelope::Updated {
440 id,
441 title,
442 entity_type,
443 }
444 }
445 crate::ops::ChangeEnvelope::Removed { id, .. } => {
446 crate::ops::ChangeEnvelope::Removed {
447 id,
448 title: None,
449 entity_type: None,
450 }
451 }
452 crate::ops::ChangeEnvelope::Renamed { from_id, to_id, .. } => {
453 let (title, entity_type) = lookup_title_and_type(&self.store, &to_id);
454 crate::ops::ChangeEnvelope::Renamed {
455 from_id,
456 to_id,
457 title,
458 entity_type,
459 }
460 }
461 })
462 .collect();
463
464 // Out-of-range `rename_similarity` is now a hard refusal (see
465 // early-return above); the response carries no clamping warning.
466 let warnings: Vec<crate::ops::WarningHint> = Vec::new();
467
468 // The backend populates
469 // notes + memstead_ref on every git-branch call (folder + archive
470 // backends leave them empty / None). Surface them
471 // unconditionally; the MCP `include_notes` parameter becomes
472 // a renderer-side filter rather than a separate engine call.
473 let notes = if backend_changes.notes.is_empty() && backend_changes.memstead_ref.is_none() {
474 None
475 } else {
476 Some(backend_changes.notes)
477 };
478 Ok(crate::ops::ChangesReport {
479 mem: mem.to_string(),
480 since: backend_changes.since,
481 head: backend_changes.head,
482 changes: enriched,
483 warnings,
484 notes,
485 memstead_ref: backend_changes.memstead_ref,
486 })
487 }
488
489 /// Fetch updates from `remote` into the workspace's mem-repo.
490 /// Advances remote-tracking refs only; the local branch pointer
491 /// is not moved.
492 ///
493 /// `refspecs` is forwarded verbatim to `git fetch`. An empty list
494 /// uses the remote's configured defaults.
495 ///
496 /// Refusal codes: `UNKNOWN_MEM`, `UNKNOWN_REMOTE`,
497 /// `INVALID_INPUT` (folder / archive mounts).
498 ///
499 /// V1 atomicity: schema-validation quarantine for
500 /// fetched commits is not yet wired. The remote-tracking refs
501 /// advance unconditionally on a successful fetch; downstream
502 /// schema validation runs on read via the engine's existing
503 /// reload pipeline.
504 pub fn fetch(
505 &self,
506 mem: &str,
507 remote: &str,
508 refspecs: &[String],
509 ) -> Result<crate::ops::FetchOutcome, EngineError> {
510 let m = self.find_mount(mem)?;
511 match &m.mount.storage {
512 MountStorage::Folder { .. } | MountStorage::Archive { .. } | MountStorage::InMemory => {
513 Err(EngineError::InvalidInput(format!(
514 "mem '{mem}' is not git-backed — `memstead_fetch` requires a git-branch mount",
515 )))
516 }
517 MountStorage::GitBranch { gitdir, .. } => match self.git_branch_ops.as_ref() {
518 Some(hook) => (hook.fetch)(gitdir, remote, refspecs).map_err(|e| match e {
519 BackendError::Other(msg) if msg.starts_with("UNKNOWN_REMOTE:") => {
520 EngineError::UnknownRemote(
521 msg.trim_start_matches("UNKNOWN_REMOTE:").trim().to_string(),
522 )
523 }
524 other => EngineError::Backend(other),
525 }),
526 None => Err(EngineError::Backend(BackendError::Other(
527 "git-branch fetch hook not installed (full flavour not loaded)".to_string(),
528 ))),
529 },
530 }
531 }
532
533 /// Pull updates from `remote` into the named mem's branch.
534 /// Fetches into the remote-tracking ref, runs a pre-merge schema
535 /// validation pass against the prospective state, then
536 /// fast-forwards the local branch. Refuses with
537 /// `LOCAL_DIVERGENCE` for diverged local branches and with
538 /// `SCHEMA_VIOLATION_IN_FETCH` when the prospective state fails
539 /// schema validation — in both refusal cases the local branch
540 /// pointer is untouched (the underlying fetch has updated
541 /// `refs/remotes/*` but the engine has not promoted the new
542 /// state).
543 pub fn pull(
544 &mut self,
545 mem: &str,
546 remote: &str,
547 ) -> Result<crate::ops::PullOutcome, EngineError> {
548 let mount_idx = self
549 .mounts
550 .iter()
551 .position(|m| m.mount.mem == mem)
552 .ok_or_else(|| self.unknown_mem_error(mem))?;
553 let (gitdir, branch) = match &self.mounts[mount_idx].mount.storage {
554 MountStorage::Folder { .. } | MountStorage::Archive { .. } | MountStorage::InMemory => {
555 return Err(EngineError::InvalidInput(format!(
556 "mem '{mem}' is not git-backed — `memstead_pull` requires a git-branch mount",
557 )));
558 }
559 MountStorage::GitBranch { gitdir, branch } => (gitdir.clone(), branch.clone()),
560 };
561
562 // Run the fetch step alone first so we can validate the
563 // prospective state against the schema before letting the
564 // pull's fast-forward land. Errors map to the typed surface
565 // just like a standalone `memstead_fetch` call.
566 let hook = self.git_branch_ops.ok_or_else(|| {
567 EngineError::Backend(BackendError::Other(
568 "git-branch pull hook not installed (full flavour not loaded)".to_string(),
569 ))
570 })?;
571 (hook.fetch)(&gitdir, remote, &[]).map_err(|e| match e {
572 BackendError::Other(msg) if msg.starts_with("UNKNOWN_REMOTE:") => {
573 EngineError::UnknownRemote(
574 msg.trim_start_matches("UNKNOWN_REMOTE:").trim().to_string(),
575 )
576 }
577 other => EngineError::Backend(other),
578 })?;
579
580 // Pre-merge schema validation. The remote-tracking ref now
581 // points at the fetched tip; we walk it, parse every `.md`
582 // blob against the mem's pinned schema, and refuse the
583 // pull if any parse fails. The local branch pointer is still
584 // unchanged at this point — the refusal is fully atomic.
585 // The remote-tracking ref follows the mount's declared
586 // branch, never the mem name — the two differ on namespaced
587 // mounts.
588 let remote_ref = format!(
589 "refs/remotes/{remote}/{}",
590 crate::workspace::branch_short_name(&branch)
591 );
592 self.validate_ref_against_schema(&hook, &gitdir, mem, &remote_ref)?;
593
594 // Run the underlying pull (re-runs the fetch via git CLI, but
595 // that's a no-op cache-wise and keeps the fast-forward logic
596 // co-located with the rest of the transport implementation).
597 let outcome = (hook.pull)(&gitdir, remote, &branch, mem).map_err(|e| match e {
598 BackendError::Other(msg) if msg.starts_with("UNKNOWN_REMOTE:") => {
599 EngineError::UnknownRemote(
600 msg.trim_start_matches("UNKNOWN_REMOTE:").trim().to_string(),
601 )
602 }
603 BackendError::Other(msg) if msg.starts_with("LOCAL_DIVERGENCE:") => {
604 let payload = msg.trim_start_matches("LOCAL_DIVERGENCE:");
605 let mut parts = payload.splitn(2, ':');
606 let v = parts.next().unwrap_or(mem).to_string();
607 let remote_ref = parts.next().unwrap_or("refs/remotes/?/?").to_string();
608 EngineError::LocalDivergence { mem: v, remote_ref }
609 }
610 other => EngineError::Backend(other),
611 })?;
612
613 // Rewind cached head + emit change event.
614 if outcome.previous_sha != outcome.new_sha {
615 if let Some(state) = self.mounts.get_mut(mount_idx) {
616 state.last_known_head = Some(outcome.new_sha.clone());
617 }
618 let event = crate::engine::events::MemChangedEvent {
619 mem: mem.to_string(),
620 head: outcome.new_sha.clone(),
621 previous: outcome.previous_sha.clone(),
622 n_commits: 1,
623 };
624 self.emit_mem_changed(&event);
625 }
626 Ok(outcome)
627 }
628
629 /// Push the named mem's branch to `remote`. Runs a pre-push
630 /// schema validation pass against the local branch tree; refuses
631 /// with `LOCAL_INVALID_STATE` when the local state fails schema
632 /// validation (the remote is not contacted in that case). Refuses
633 /// with `NON_FAST_FORWARD` when the push is not a fast-forward
634 /// and `force: false`; with `force: true` runs a
635 /// `--force-with-lease` push instead.
636 pub fn push(
637 &self,
638 mem: &str,
639 remote: &str,
640 force: bool,
641 ) -> Result<crate::ops::PushOutcome, EngineError> {
642 let m = self.find_mount(mem)?;
643 let (gitdir, branch) = match &m.mount.storage {
644 MountStorage::Folder { .. } | MountStorage::Archive { .. } | MountStorage::InMemory => {
645 return Err(EngineError::InvalidInput(format!(
646 "mem '{mem}' is not git-backed — `memstead_push` requires a git-branch mount",
647 )));
648 }
649 MountStorage::GitBranch { gitdir, branch } => (gitdir.clone(), branch.clone()),
650 };
651 let hook = self.git_branch_ops.ok_or_else(|| {
652 EngineError::Backend(BackendError::Other(
653 "git-branch push hook not installed (full flavour not loaded)".to_string(),
654 ))
655 })?;
656
657 // Pre-push schema validation: walk the local branch tree (the
658 // mount's declared branch, never a ref derived from the mem
659 // name), run the mem's pinned schema over every `.md` blob.
660 // Any parse failure refuses the push with
661 // `LOCAL_INVALID_STATE` — the remote is not contacted. A gate
662 // that cannot run (the ref unresolvable, the tree unreadable)
663 // refuses too: a validation that did not happen must never
664 // read as a pass.
665 let local_ref = crate::workspace::branch_full_ref(&branch);
666 match self.validate_ref_against_schema(&hook, &gitdir, mem, &local_ref) {
667 Ok(()) => {}
668 Err(EngineError::SchemaViolationInFetch { violations, .. }) => {
669 return Err(EngineError::LocalInvalidState {
670 mem: mem.to_string(),
671 remote: remote.to_string(),
672 detail: format!(
673 "{} violation(s) in local branch: {}",
674 violations.len(),
675 violations.join("; "),
676 ),
677 });
678 }
679 Err(EngineError::UnknownRef(raw)) => {
680 // Names the declared branch the gate looked for, so
681 // the reader sees what was resolved, not an invented
682 // ref.
683 return Err(EngineError::UnknownRef(raw));
684 }
685 Err(other) => {
686 return Err(EngineError::Backend(BackendError::Other(format!(
687 "pre-push schema validation could not run for mem '{mem}' \
688 (declared branch `{local_ref}`): {other}",
689 ))));
690 }
691 }
692
693 (hook.push)(&gitdir, remote, &branch, mem, force).map_err(|e| match e {
694 BackendError::Other(msg) if msg.starts_with("UNKNOWN_REMOTE:") => {
695 EngineError::UnknownRemote(
696 msg.trim_start_matches("UNKNOWN_REMOTE:").trim().to_string(),
697 )
698 }
699 BackendError::Other(msg) if msg.starts_with("NON_FAST_FORWARD:") => {
700 let payload = msg.trim_start_matches("NON_FAST_FORWARD:");
701 let mut parts = payload.splitn(2, ':');
702 let v = parts.next().unwrap_or(mem).to_string();
703 let r = parts.next().unwrap_or(remote).to_string();
704 EngineError::NonFastForward { mem: v, remote: r }
705 }
706 BackendError::Other(msg) if msg.starts_with("UNKNOWN_REF:") => {
707 EngineError::UnknownRef(msg.trim_start_matches("UNKNOWN_REF:").trim().to_string())
708 }
709 other => EngineError::Backend(other),
710 })
711 }
712
713 /// Push every mounted git-branch mem's declared branch, plus the
714 /// `__MEMSTEAD` ref of every mem-repo those mounts share, to
715 /// `remote` — fast-forward only, there is no force variant. One
716 /// `ls-remote` per mem-repo decides which refs lag; a ref whose
717 /// local and remote SHA agree is reported `in_sync` and never
718 /// sent. Each lagging mem branch goes through [`Self::push`]
719 /// (the pre-push schema gate included); `__MEMSTEAD` goes through
720 /// the same transport seam with no schema gate, it IS the
721 /// schemas. A refusal on one ref lands in `refused` with its
722 /// typed code and the run continues with the next ref — the
723 /// caller decides the exit status from the outcome. Folder,
724 /// archive and in-memory mounts have no branch and are skipped.
725 /// Only a remote that cannot be listed at all (`UNKNOWN_REMOTE`)
726 /// fails the run as a whole.
727 pub fn push_all(&self, remote: &str) -> Result<crate::ops::PushAllOutcome, EngineError> {
728 use crate::ops::{PushAllOutcome, PushedRef, RefusedRef};
729
730 let hook = self.git_branch_ops.ok_or_else(|| {
731 EngineError::Backend(BackendError::Other(
732 "git-branch push hook not installed (full flavour not loaded)".to_string(),
733 ))
734 })?;
735
736 // Mem-repos in mount order, each once; a workspace may spread
737 // its mounts over more than one gitdir.
738 let mut gitdirs: Vec<std::path::PathBuf> = Vec::new();
739 for m in &self.mounts {
740 if let MountStorage::GitBranch { gitdir, .. } = &m.mount.storage
741 && !gitdirs.contains(gitdir)
742 {
743 gitdirs.push(gitdir.clone());
744 }
745 }
746
747 let mut outcome = PushAllOutcome {
748 remote: remote.to_string(),
749 ..Default::default()
750 };
751
752 for gitdir in &gitdirs {
753 let remote_refs: std::collections::HashMap<String, String> =
754 (hook.ls_remote)(gitdir, remote)
755 .map_err(|e| match e {
756 BackendError::Other(msg) if msg.starts_with("UNKNOWN_REMOTE:") => {
757 EngineError::UnknownRemote(
758 msg.trim_start_matches("UNKNOWN_REMOTE:").trim().to_string(),
759 )
760 }
761 other => EngineError::Backend(other),
762 })?
763 .into_iter()
764 .collect();
765
766 // `__MEMSTEAD` first: schemas and mem configs before the
767 // content that depends on them, so a clone that pulls
768 // mid-run never sees a mem its schema ref lacks.
769 let memstead_ref = crate::workspace::branch_full_ref(crate::MEMSTEAD_REF_BRANCH);
770 let mut targets: Vec<(String, Option<String>, String)> = Vec::new();
771 match (hook.resolve_ref)(gitdir, &memstead_ref) {
772 Ok(Some(_)) => targets.push((
773 memstead_ref.clone(),
774 None,
775 crate::MEMSTEAD_REF_BRANCH.to_string(),
776 )),
777 Ok(None) => {}
778 Err(e) => return Err(EngineError::Backend(e)),
779 }
780 for m in &self.mounts {
781 if let MountStorage::GitBranch { gitdir: g, branch } = &m.mount.storage
782 && g == gitdir
783 {
784 targets.push((
785 crate::workspace::branch_full_ref(branch),
786 Some(m.mount.mem.clone()),
787 branch.clone(),
788 ));
789 }
790 }
791
792 for (ref_name, mem, branch) in targets {
793 let local_sha = match (hook.resolve_ref)(gitdir, &ref_name) {
794 Ok(Some(sha)) => sha,
795 Ok(None) => {
796 outcome.refused.push(RefusedRef {
797 ref_name: ref_name.clone(),
798 mem: mem.clone(),
799 code: "UNKNOWN_REF".to_string(),
800 message: format!(
801 "{ref_name} does not exist locally{}",
802 mem.as_deref()
803 .map(|m| format!(" (mem `{m}`'s declared branch)"))
804 .unwrap_or_default()
805 ),
806 });
807 continue;
808 }
809 Err(e) => return Err(EngineError::Backend(e)),
810 };
811 let previous_sha = remote_refs.get(&ref_name).cloned().unwrap_or_default();
812 if previous_sha == local_sha {
813 outcome.in_sync.push(ref_name);
814 continue;
815 }
816 let result = match &mem {
817 Some(m) => self.push(m, remote, false).map(|o| o.new_sha),
818 None => (hook.push)(gitdir, remote, &branch, crate::MEMSTEAD_REF_BRANCH, false)
819 .map(|o| o.new_sha)
820 .map_err(|e| match e {
821 BackendError::Other(msg) if msg.starts_with("NON_FAST_FORWARD:") => {
822 EngineError::NonFastForward {
823 mem: crate::MEMSTEAD_REF_BRANCH.to_string(),
824 remote: remote.to_string(),
825 }
826 }
827 BackendError::Other(msg) if msg.starts_with("UNKNOWN_REMOTE:") => {
828 EngineError::UnknownRemote(
829 msg.trim_start_matches("UNKNOWN_REMOTE:").trim().to_string(),
830 )
831 }
832 other => EngineError::Backend(other),
833 }),
834 };
835 match result {
836 Ok(new_sha) => outcome.pushed.push(PushedRef {
837 ref_name,
838 mem,
839 previous_sha,
840 new_sha,
841 }),
842 Err(e) => {
843 // The single-mem refusal names `force`, which
844 // `--all` does not offer; say what applies here.
845 let message = match &e {
846 EngineError::NonFastForward { .. } => format!(
847 "remote `{remote}` carries commits on {ref_name} this clone lacks; \
848 fetch and pull {}, then push again",
849 mem.as_deref()
850 .map(|m| format!("mem `{m}`"))
851 .unwrap_or_else(|| "the schema ref".to_string()),
852 ),
853 other => other.to_string(),
854 };
855 outcome.refused.push(RefusedRef {
856 ref_name,
857 mem,
858 code: e.code().to_string(),
859 message,
860 })
861 }
862 }
863 }
864 }
865
866 Ok(outcome)
867 }
868
869 /// Configure (or re-point) a named remote on the workspace's
870 /// mem-repo, so `fetch` / `pull` / `push` have somewhere to go.
871 /// Upsert semantics — safe to re-run with a new URL. The mem-repo
872 /// is shared by every git-branch mount, so the op is
873 /// workspace-level: any git-branch mount locates it; refuses
874 /// `INVALID_INPUT` when the workspace has none.
875 pub fn remote_add(
876 &self,
877 name: &str,
878 url: &str,
879 ) -> Result<crate::ops::RemoteAddOutcome, EngineError> {
880 // Both values become git subprocess arguments — refuse shapes
881 // that would parse as flags.
882 if name.is_empty() || name.starts_with('-') || url.is_empty() || url.starts_with('-') {
883 return Err(EngineError::InvalidInput(format!(
884 "remote name and url must be non-empty and must not start with '-' \
885 (got name '{name}', url '{url}')",
886 )));
887 }
888 let gitdir = self
889 .mounts
890 .iter()
891 .find_map(|m| match &m.mount.storage {
892 MountStorage::GitBranch { gitdir, .. } => Some(gitdir.clone()),
893 _ => None,
894 })
895 .ok_or_else(|| {
896 EngineError::InvalidInput(
897 "no git-branch mounts — `remote-add` requires a mem-repo workspace".to_string(),
898 )
899 })?;
900 let hook = self.git_branch_ops.ok_or_else(|| {
901 EngineError::Backend(BackendError::Other(
902 "git-branch remote_add hook not installed (full flavour not loaded)".to_string(),
903 ))
904 })?;
905 (hook.remote_add)(&gitdir, name, url).map_err(EngineError::Backend)
906 }
907
908 /// Pre-merge schema validation pass: walks every `.md` blob at
909 /// `ref_name` and runs `parse_entries` with the mem's pinned
910 /// schema. Returns `Ok(())` when the tree is schema-clean;
911 /// returns `EngineError::SchemaViolationInFetch` with the list of
912 /// per-entity violation messages otherwise. The validation is
913 /// strict on parse-time errors — any `(path, error)` pair from
914 /// `parse_entries` triggers a refusal.
915 ///
916 /// `ref_name` is the prospective state (a `refs/remotes/*` ref
917 /// for pull, `refs/heads/*` for push). The engine layer maps the
918 /// returned error into the surface code it needs
919 /// (`SCHEMA_VIOLATION_IN_FETCH` for pull, `LOCAL_INVALID_STATE`
920 /// for push).
921 fn validate_ref_against_schema(
922 &self,
923 hook: &crate::engine::GitBranchOps,
924 gitdir: &std::path::Path,
925 mem: &str,
926 ref_name: &str,
927 ) -> Result<(), EngineError> {
928 // The REF first, then the schema. A branch that does not exist is the
929 // clearer answer, and asking for the schema first turned "that branch
930 // is not there" into "the schema pin did not resolve", which is true
931 // of a quarantined mem and tells the reader nothing about why their
932 // push failed (04/05).
933 let blobs = (hook.read_tree)(gitdir, ref_name).map_err(|e| match e {
934 BackendError::Other(msg) if msg.starts_with("UNKNOWN_REF:") => {
935 EngineError::UnknownRef(msg.trim_start_matches("UNKNOWN_REF:").trim().to_string())
936 }
937 other => EngineError::Backend(other),
938 })?;
939 let schema = self
940 .schemas
941 .get(mem)
942 .ok_or_else(|| EngineError::SchemaNotFound {
943 mem: mem.to_string(),
944 pin: "<missing engine-side resolution>".to_string(),
945 // Internal invariant breach (an already-resolved schema
946 // absent from the per-mem map), not a source-resolution
947 // failure — no per-source diagnostics apply.
948 sources: Vec::new(),
949 install_hint: None,
950 })?
951 .clone();
952
953 let mut source_entries: Vec<crate::entity::source::SourceEntry> = Vec::new();
954 for (rel_path, content) in blobs {
955 source_entries.push(crate::entity::source::SourceEntry {
956 relative_path: rel_path.clone(),
957 source_path: std::path::PathBuf::from(rel_path),
958 content,
959 });
960 }
961
962 // First pass: permissive parse via the engine's loader so we
963 // can build Entity values for the strict validator. The
964 // loader silently absorbs frontmatter / title / section
965 // drift; the strict pass below is what catches it.
966 let load_result = crate::entity::loader::parse_entries(
967 source_entries.clone(),
968 Vec::new(),
969 mem,
970 schema.as_ref(),
971 );
972 let mut violations: Vec<String> = load_result
973 .errors
974 .iter()
975 .map(|(path, msg)| format!("{}: {msg}", path.display()))
976 .collect();
977
978 // Strict per-entity validator: enforces "looks like a mem
979 // entity" invariants (frontmatter shape, title presence,
980 // required sections, unknown sections, relationship syntax,
981 // wiki-link shape) that the permissive loader doesn't refuse.
982 // Re-runs against the same source bytes so unparseable
983 // frontmatter surfaces here even when the loader's tolerant
984 // path produces an Entity stub.
985 let entities_by_path: std::collections::HashMap<String, &crate::entity::Entity> =
986 load_result
987 .entities
988 .iter()
989 .map(|p| (p.entity.file_path.clone(), &p.entity))
990 .collect();
991 for source in &source_entries {
992 let Some(entity) = entities_by_path.get(&source.relative_path) else {
993 continue;
994 };
995 let type_def = match schema.get_type(&entity.entity_type) {
996 Some(t) => t,
997 None => {
998 violations.push(format!(
999 "{}: unknown entity_type '{}' in schema",
1000 source.relative_path, entity.entity_type,
1001 ));
1002 continue;
1003 }
1004 };
1005 if let Err(e) = crate::validator::strict::validate_strict(
1006 &source.content,
1007 entity,
1008 type_def.as_ref(),
1009 &source.relative_path,
1010 ) {
1011 violations.push(format!("{}: {e}", source.relative_path));
1012 }
1013 }
1014
1015 if violations.is_empty() {
1016 Ok(())
1017 } else {
1018 Err(EngineError::SchemaViolationInFetch {
1019 mem: mem.to_string(),
1020 ref_name: ref_name.to_string(),
1021 violations,
1022 })
1023 }
1024 }
1025
1026 /// Reset a mem's branch pointer to `target_sha`. The only
1027 /// engine surface that moves a branch pointer over existing
1028 /// commits — every other mutation appends. Refuses if any commit
1029 /// that would be discarded by the reset is already reachable from
1030 /// a `refs/remotes/*` ref (the engine's definition of "pushed").
1031 ///
1032 /// `target_sha` accepts anything `gix::rev_parse_single` admits:
1033 /// a SHA, an abbreviated SHA, a branch name, a tag. The branch
1034 /// itself (the mount's declared branch) must exist.
1035 ///
1036 /// Refusal codes:
1037 /// - [`EngineError::UnknownMem`] (`UNKNOWN_MEM`)
1038 /// - [`EngineError::UnknownRef`] (`UNKNOWN_REF`) — branch or
1039 /// target ref does not resolve.
1040 /// - [`EngineError::PushedCommitsProtected`]
1041 /// (`PUSHED_COMMITS_PROTECTED`) — at least one discarded commit
1042 /// is pushed. The error carries the offending SHAs verbatim.
1043 /// - [`EngineError::InvalidInput`] (`INVALID_INPUT`) — mem is
1044 /// folder / archive-backed (history rewriting only makes sense
1045 /// for git-branch mounts).
1046 ///
1047 /// Emits a [`crate::engine::events::MemChangedEvent`] on
1048 /// success when the SHA actually changed; the reset's effect is
1049 /// observable through the same change-event surface every commit
1050 /// flows through. Engine's cached `last_known_head` for the
1051 /// affected mount is rewound to the new SHA so the next drift
1052 /// probe doesn't flag the reset as a sibling-writer surprise.
1053 pub fn branch_reset(
1054 &mut self,
1055 mem: &str,
1056 target_sha: &str,
1057 expected_head: Option<&str>,
1058 ) -> Result<crate::ops::BranchResetOutcome, EngineError> {
1059 let mount_idx = self
1060 .mounts
1061 .iter()
1062 .position(|m| m.mount.mem == mem)
1063 .ok_or_else(|| self.unknown_mem_error(mem))?;
1064
1065 // History rewriting is a write — read-only and archive mounts
1066 // refuse before any dispatch (parity with the mutation surface).
1067 if self.mounts[mount_idx].mount.capability != crate::workspace::MountCapability::Write {
1068 return Err(EngineError::ReadOnlyMount(mem.to_string()));
1069 }
1070
1071 let outcome = match &self.mounts[mount_idx].mount.storage {
1072 MountStorage::Folder { .. } | MountStorage::Archive { .. } | MountStorage::InMemory => {
1073 return Err(EngineError::InvalidInput(format!(
1074 "mem '{mem}' is not git-backed — `memstead_branch_reset` requires a git-branch mount",
1075 )));
1076 }
1077 MountStorage::GitBranch { gitdir, branch } => match self.git_branch_ops.as_ref() {
1078 Some(hook) => (hook.branch_reset)(gitdir, branch, target_sha, expected_head)
1079 .map_err(|e| match e {
1080 BackendError::Other(msg) if msg.starts_with("UNKNOWN_REF:") => {
1081 let raw = msg.trim_start_matches("UNKNOWN_REF:").trim().to_string();
1082 EngineError::UnknownRef(raw)
1083 }
1084 BackendError::Other(msg) if msg.starts_with("EXPECTED_HEAD_MISMATCH:") => {
1085 let current = msg
1086 .trim_start_matches("EXPECTED_HEAD_MISMATCH:")
1087 .trim()
1088 .to_string();
1089 EngineError::BranchResetHeadMoved {
1090 mem: mem.to_string(),
1091 expected: expected_head.unwrap_or_default().to_string(),
1092 current,
1093 }
1094 }
1095 BackendError::Other(msg)
1096 if msg.starts_with("PUSHED_COMMITS_PROTECTED:") =>
1097 {
1098 let payload =
1099 msg.trim_start_matches("PUSHED_COMMITS_PROTECTED:").trim();
1100 let pushed_shas = payload
1101 .split(',')
1102 .map(|s| s.trim().to_string())
1103 .filter(|s| !s.is_empty())
1104 .collect();
1105 EngineError::PushedCommitsProtected {
1106 mem: mem.to_string(),
1107 target_sha: target_sha.to_string(),
1108 pushed_shas,
1109 }
1110 }
1111 other => EngineError::Backend(other),
1112 })?,
1113 None => {
1114 return Err(EngineError::Backend(BackendError::Other(
1115 "git-branch branch_reset hook not installed (full flavour not loaded)"
1116 .to_string(),
1117 )));
1118 }
1119 },
1120 };
1121
1122 // Rewind the engine's cached HEAD so subsequent drift probes
1123 // don't surface MEM_RELOADED for the reset we just made.
1124 // Then emit a change event so subscribers see the transition
1125 // (skipping the no-op case where previous == new).
1126 if outcome.previous_sha != outcome.new_sha {
1127 if let Some(state) = self.mounts.get_mut(mount_idx) {
1128 state.last_known_head = Some(outcome.new_sha.clone());
1129 }
1130 let event = crate::engine::events::MemChangedEvent {
1131 mem: mem.to_string(),
1132 head: outcome.new_sha.clone(),
1133 previous: outcome.previous_sha.clone(),
1134 // n_commits stays at 1 for reset events. The wire
1135 // shape is the same `MemChangedEvent` consumers
1136 // already key on; semantics: "the head moved by this
1137 // operation". Replay-aware consumers branch on the
1138 // commit-vs-reset distinction by inspecting the
1139 // produced commit (a reset's new head is an existing
1140 // commit, not a freshly minted one).
1141 n_commits: 1,
1142 };
1143 self.emit_mem_changed(&event);
1144 }
1145 Ok(outcome)
1146 }
1147
1148 /// Cross-mem references that a reset of `mem` to `target_sha` would
1149 /// strand: incoming edges from entities in *other* mems whose target
1150 /// exists at the current head but would not exist at the target
1151 /// commit — entities created after the target, or renamed to their
1152 /// current id after it (the reset re-materialises the old id, so
1153 /// references to the new id dangle either way).
1154 ///
1155 /// A read — computes against the live store and the commit history,
1156 /// moves nothing. The human surface calls this fresh at
1157 /// confirmation-dialog time and warns before `branch_reset`. Sorted
1158 /// (from_id, to_id, rel_type) for stable rendering.
1159 ///
1160 /// Refusals mirror `changes_since`: `UnknownMem`, `InvalidCursor`
1161 /// for an unresolvable `target_sha`, `InvalidInput` for
1162 /// non-git-backed mounts.
1163 pub fn branch_reset_stranded_refs(
1164 &self,
1165 mem: &str,
1166 target_sha: &str,
1167 ) -> Result<Vec<crate::ops::StrandedCrossMemRef>, EngineError> {
1168 use crate::ops::ChangeEnvelope;
1169
1170 let report = self.changes_since(mem, target_sha, None)?;
1171 let mut discarded: std::collections::HashSet<String> = std::collections::HashSet::new();
1172 for change in &report.changes {
1173 match change {
1174 ChangeEnvelope::Added { id, .. } => {
1175 discarded.insert(id.to_string());
1176 }
1177 ChangeEnvelope::Renamed { to_id, .. } => {
1178 discarded.insert(to_id.to_string());
1179 }
1180 ChangeEnvelope::Updated { .. } | ChangeEnvelope::Removed { .. } => {}
1181 }
1182 }
1183 if discarded.is_empty() {
1184 return Ok(Vec::new());
1185 }
1186
1187 let mut stranded: Vec<crate::ops::StrandedCrossMemRef> = self
1188 .store
1189 .all_entities()
1190 .filter(|e| e.mem != mem)
1191 .flat_map(|e| {
1192 e.relationships
1193 .iter()
1194 .filter(|r| discarded.contains(&r.target.to_string()))
1195 .map(|r| crate::ops::StrandedCrossMemRef {
1196 from_id: e.id.to_string(),
1197 from_mem: e.mem.clone(),
1198 to_id: r.target.to_string(),
1199 rel_type: r.rel_type.clone(),
1200 })
1201 .collect::<Vec<_>>()
1202 })
1203 .collect();
1204 stranded.sort_by(|a, b| {
1205 (&a.from_id, &a.to_id, &a.rel_type).cmp(&(&b.from_id, &b.to_id, &b.rel_type))
1206 });
1207 Ok(stranded)
1208 }
1209
1210 /// Two-ref structural diff. Produces a per-entity [`crate::ops::Diff`]
1211 /// comparing the trees at `ref_a` and `ref_b` for the named
1212 /// mem's storage. Folder and archive backends carry no git
1213 /// refs and refuse via [`EngineError::InvalidInput`]; the
1214 /// git-branch backend routes through [`GitBranchOps::diff`] when
1215 /// the full flavour is loaded.
1216 ///
1217 /// `mem` selects the storage context (the gitdir, for
1218 /// git-branch mounts). `ref_a` / `ref_b` are arbitrary refs the
1219 /// underlying git layer accepts — branch names, commit SHAs, tag
1220 /// names — so cross-mem diffs work via fully-qualified refs
1221 /// (the other mount's declared branch) without a separate API. A
1222 /// bare `HEAD` token re-anchors onto this mem's declared branch.
1223 ///
1224 /// Refusal codes:
1225 /// - [`EngineError::UnknownMem`] (`UNKNOWN_MEM`) — no mount
1226 /// for `mem`.
1227 /// - [`EngineError::UnknownRef`] (`UNKNOWN_REF`) — either ref
1228 /// does not resolve. Surfaces verbatim from the git layer's
1229 /// `rev_parse` refusal.
1230 /// - [`EngineError::RenameSimilarityOutOfRange`] (`INVALID_INPUT`)
1231 /// — `config.rename_similarity` outside `[0.1, 1.0]`.
1232 /// - [`EngineError::InvalidInput`] (`INVALID_INPUT`) — mem is
1233 /// folder or archive-backed (no refs to diff).
1234 pub fn diff(
1235 &self,
1236 mem: &str,
1237 ref_a: &str,
1238 ref_b: &str,
1239 config: Option<crate::ops::DiffConfig>,
1240 ) -> Result<crate::ops::Diff, EngineError> {
1241 let m = self.find_mount(mem)?;
1242 let config = config.unwrap_or_default();
1243
1244 if config.rename_similarity < crate::ops::RENAME_SIMILARITY_MIN
1245 || config.rename_similarity > crate::ops::RENAME_SIMILARITY_MAX
1246 {
1247 return Err(EngineError::RenameSimilarityOutOfRange {
1248 requested: config.rename_similarity,
1249 allowed_min: crate::ops::RENAME_SIMILARITY_MIN,
1250 allowed_max: crate::ops::RENAME_SIMILARITY_MAX,
1251 });
1252 }
1253
1254 match &m.mount.storage {
1255 MountStorage::Folder { .. } | MountStorage::Archive { .. } | MountStorage::InMemory => {
1256 Err(EngineError::InvalidInput(format!(
1257 "mem '{mem}' is not git-backed — `memstead_diff` requires a git-branch mount",
1258 )))
1259 }
1260 MountStorage::GitBranch { gitdir, branch } => match self.git_branch_ops.as_ref() {
1261 Some(hook) => {
1262 (hook.diff)(gitdir, branch, mem, ref_a, ref_b, &config).map_err(|e| match e {
1263 // Map the standard backend-side "ref not found" shape into the
1264 // typed engine-level refusal. The git-branch dispatcher uses
1265 // `BackendError::Other` with a leading marker so the engine can
1266 // recover the typed code without re-parsing the message.
1267 BackendError::Other(msg) if msg.starts_with("UNKNOWN_REF:") => {
1268 let raw = msg.trim_start_matches("UNKNOWN_REF:").trim().to_string();
1269 EngineError::UnknownRef(raw)
1270 }
1271 other => EngineError::Backend(other),
1272 })
1273 }
1274 None => Err(EngineError::Backend(BackendError::Other(
1275 "git-branch diff hook not installed (full flavour not loaded)".to_string(),
1276 ))),
1277 },
1278 }
1279 }
1280}
1281
1282#[cfg(test)]
1283mod tests {
1284 use std::path::{Path, PathBuf};
1285
1286 use tempfile::TempDir;
1287
1288 use crate::backend::{BackendError, MemBackend};
1289 use crate::engine::test_helpers::*;
1290 use crate::engine::{DeleteEntityArgs, Engine, EngineError};
1291 use crate::entity::EntityId;
1292
1293 use crate::provenance::Provenance;
1294 use crate::storage::ArchiveBackend;
1295 use crate::vcs::CommitContext;
1296 use crate::workspace::{Mount, MountCapability, MountLifecycle, MountStorage};
1297
1298 #[test]
1299 fn engine_diff_unknown_mem_returns_typed_error() {
1300 let tmp = TempDir::new().unwrap();
1301 let engine = build_demo_engine(&tmp);
1302 let err = engine.diff("nope", "a", "b", None).unwrap_err();
1303 assert!(matches!(err, EngineError::UnknownMem(v) if v == "nope"));
1304 }
1305
1306 /// A `write_id` from a mutation response passed back as `since`
1307 /// on a folder mem refuses with the typed `INVALID_CURSOR` code
1308 /// and a message naming both the right cursor and the confusion —
1309 /// before this guard it silently replayed the whole history.
1310 #[test]
1311 fn engine_changes_since_folder_refuses_write_token_as_cursor() {
1312 let tmp = TempDir::new().unwrap();
1313 let engine = build_demo_engine(&tmp);
1314 let token = format!("{:032x}{:016x}", 1_766_000_000_000_000_000u128, 7u64);
1315 let err = engine.changes_since("specs", &token, None).unwrap_err();
1316 assert_eq!(err.code(), "INVALID_CURSOR");
1317 match &err {
1318 EngineError::InvalidTimestampCursor { mem, since } => {
1319 assert_eq!(mem, "specs");
1320 assert_eq!(since, &token);
1321 }
1322 other => panic!("expected InvalidTimestampCursor, got {other:?}"),
1323 }
1324 let msg = err.to_string();
1325 assert!(
1326 msg.contains("RFC3339"),
1327 "message names the cursor dialect: {msg}"
1328 );
1329 assert!(
1330 msg.contains("`write_id` is an identity, not a cursor"),
1331 "message names the confusion: {msg}"
1332 );
1333 // The sentinel and a real timestamp still read.
1334 assert!(
1335 engine
1336 .changes_since("specs", crate::ops::EMPTY_TREE_SHA, None)
1337 .is_ok()
1338 );
1339 assert!(engine.changes_since("specs", "", None).is_ok());
1340 }
1341
1342 #[test]
1343 fn engine_diff_folder_mount_refuses_with_invalid_input() {
1344 let tmp = TempDir::new().unwrap();
1345 let engine = build_demo_engine(&tmp);
1346 // Folder backend has no git refs — refuse cleanly via the
1347 // typed `INVALID_INPUT` code rather than collapsing through
1348 // the backend layer.
1349 let err = engine.diff("specs", "a", "b", None).unwrap_err();
1350 match err {
1351 EngineError::InvalidInput(msg) => {
1352 assert!(msg.contains("not git-backed"), "unexpected msg: {msg}");
1353 }
1354 other => panic!("expected InvalidInput, got {other:?}"),
1355 }
1356 }
1357
1358 #[test]
1359 fn engine_diff_rename_similarity_out_of_range_refuses() {
1360 let tmp = TempDir::new().unwrap();
1361 let engine = build_demo_engine(&tmp);
1362 let bad = crate::ops::DiffConfig {
1363 rename_similarity: 2.0,
1364 ..Default::default()
1365 };
1366 let err = engine.diff("specs", "a", "b", Some(bad)).unwrap_err();
1367 assert!(matches!(
1368 err,
1369 EngineError::RenameSimilarityOutOfRange { .. }
1370 ));
1371 }
1372
1373 #[test]
1374 fn engine_changes_since_archive_mount_returns_empty_report() {
1375 // Archive backends have no diff surface; the engine wrapper
1376 // produces an empty `ChangesReport` with the cursor echoed.
1377 let tmp = TempDir::new().unwrap();
1378 let archive_path = build_archive(tmp.path(), "ext", &[("a.md", b"a")]);
1379 let mount = archive_mount("ext", archive_path.clone());
1380 let engine = Engine::from_mounts(vec![(
1381 mount,
1382 Box::new(ArchiveBackend::new(archive_path)) as Box<dyn MemBackend>,
1383 )])
1384 .unwrap();
1385 let report = engine.changes_since("ext", "abc", None).expect("known mem");
1386 assert_eq!(report.mem, "ext");
1387 assert_eq!(report.since, "abc");
1388 assert_eq!(report.head, "abc");
1389 assert!(report.changes.is_empty());
1390 assert!(report.warnings.is_empty());
1391 }
1392
1393 #[test]
1394 fn engine_changes_since_unknown_mem_returns_typed_error() {
1395 let tmp = TempDir::new().unwrap();
1396 let engine = build_demo_engine(&tmp);
1397 let err = engine
1398 .changes_since("does-not-exist", "abc", None)
1399 .unwrap_err();
1400 assert!(matches!(err, EngineError::UnknownMem(_)));
1401 }
1402
1403 #[test]
1404 fn engine_changes_since_refuses_rename_similarity_below_min() {
1405 let tmp = TempDir::new().unwrap();
1406 let engine = build_demo_engine(&tmp);
1407 // 0.05 is below RENAME_SIMILARITY_MIN (0.1); typed refusal,
1408 // not a silent clamp.
1409 let err = engine
1410 .changes_since("specs", "abc", Some(0.05))
1411 .expect_err("out-of-range refuses");
1412 match err {
1413 EngineError::RenameSimilarityOutOfRange {
1414 requested,
1415 allowed_min,
1416 allowed_max,
1417 } => {
1418 assert!((requested - 0.05).abs() < f32::EPSILON);
1419 assert!((allowed_min - crate::ops::RENAME_SIMILARITY_MIN).abs() < f32::EPSILON);
1420 assert!((allowed_max - crate::ops::RENAME_SIMILARITY_MAX).abs() < f32::EPSILON);
1421 }
1422 other => panic!("expected RenameSimilarityOutOfRange, got {other:?}"),
1423 }
1424 }
1425
1426 #[test]
1427 fn engine_changes_since_refuses_rename_similarity_above_max() {
1428 let tmp = TempDir::new().unwrap();
1429 let engine = build_demo_engine(&tmp);
1430 // 1.5 is above RENAME_SIMILARITY_MAX (1.0); typed refusal.
1431 let err = engine
1432 .changes_since("specs", "abc", Some(1.5))
1433 .expect_err("out-of-range refuses");
1434 match err {
1435 EngineError::RenameSimilarityOutOfRange { requested, .. } => {
1436 assert!((requested - 1.5).abs() < f32::EPSILON);
1437 }
1438 other => panic!("expected RenameSimilarityOutOfRange, got {other:?}"),
1439 }
1440 }
1441
1442 #[test]
1443 fn engine_changes_since_no_warning_when_rename_similarity_in_range() {
1444 let tmp = TempDir::new().unwrap();
1445 let engine = build_demo_engine(&tmp);
1446 // 0.5 is comfortably inside the valid range; no warning.
1447 let report = engine
1448 .changes_since("specs", "", Some(0.5))
1449 .expect("known mem");
1450 assert!(report.warnings.is_empty());
1451 }
1452
1453 #[test]
1454 fn engine_changes_since_no_warning_when_rename_similarity_omitted() {
1455 // Caller passes None → wrapper falls back to the default;
1456 // no clamping, no warning.
1457 let tmp = TempDir::new().unwrap();
1458 let engine = build_demo_engine(&tmp);
1459 let report = engine.changes_since("specs", "", None).expect("known mem");
1460 assert!(report.warnings.is_empty());
1461 }
1462
1463 #[test]
1464 fn engine_changes_since_enriches_envelope_title_and_type_from_store() {
1465 // `build_demo_engine` creates three entities via the engine's
1466 // mutation pipeline, which appends Create events to the folder
1467 // backend's changelog. `Engine::changes_since` synthesises
1468 // BackendChanges from the changelog (id-only envelopes), then
1469 // enriches title / entity_type from the in-memory store.
1470 let tmp = TempDir::new().unwrap();
1471 let engine = build_demo_engine(&tmp);
1472 let report = engine
1473 .changes_since("specs", crate::ops::EMPTY_TREE_SHA, None)
1474 .expect("known mem");
1475
1476 // Three Create events → three Added envelopes, each enriched.
1477 assert_eq!(report.changes.len(), 3);
1478 for env in &report.changes {
1479 match env {
1480 crate::ops::ChangeEnvelope::Added {
1481 id,
1482 title,
1483 entity_type,
1484 } => {
1485 assert!(title.is_some(), "title enriched for {id}");
1486 assert_eq!(entity_type.as_deref(), Some("spec"), "type for {id}");
1487 }
1488 other => panic!("expected Added envelope, got {other:?}"),
1489 }
1490 }
1491 }
1492
1493 #[test]
1494 fn engine_changes_since_removed_envelope_keeps_title_and_type_none() {
1495 // Create-then-delete net effect = Removed. Even though the
1496 // store may still know the entity, the engine wrapper
1497 // unconditionally strips title / entity_type on Removed.
1498 let tmp = TempDir::new().unwrap();
1499 let mut engine = build_demo_engine(&tmp);
1500 let (actor, client) = cli_actor();
1501 let id = EntityId::new("specs", "lonely-three");
1502 let hash = engine
1503 .get_entity(&id)
1504 .expect("seeded entity present")
1505 .content_hash
1506 .clone();
1507 engine
1508 .delete_entity(
1509 DeleteEntityArgs {
1510 id: id.clone(),
1511 expected_hash: Some(hash),
1512 },
1513 actor,
1514 Some(&client),
1515 None,
1516 )
1517 .unwrap();
1518 let report = engine
1519 .changes_since("specs", crate::ops::EMPTY_TREE_SHA, None)
1520 .unwrap();
1521 let removed = report
1522 .changes
1523 .iter()
1524 .find(|e| {
1525 matches!(e,
1526 crate::ops::ChangeEnvelope::Removed { id: rid, .. } if rid == &id)
1527 })
1528 .expect("removed envelope for lonely-three");
1529 match removed {
1530 crate::ops::ChangeEnvelope::Removed {
1531 title, entity_type, ..
1532 } => {
1533 assert!(title.is_none());
1534 assert!(entity_type.is_none());
1535 }
1536 other => panic!("expected Removed, got {other:?}"),
1537 }
1538 }
1539
1540 // ---- Engine::cross_mem_link_allowed ---------------------------
1541
1542 #[test]
1543 fn reload_if_stale_returns_empty_for_folder_only_engine() {
1544 // Folder mems now carry a changelog-derived drift cursor, so
1545 // this pins the QUIET case: no sibling wrote between probes,
1546 // so repeated checks stay warning-free (the first probe
1547 // captures the baseline silently, the second sees no advance).
1548 let tmp = TempDir::new().unwrap();
1549 let mut engine = build_demo_engine(&tmp);
1550 let warnings = engine.reload_if_stale(None);
1551 assert!(warnings.is_empty());
1552 let warnings = engine.reload_if_stale(Some("specs"));
1553 assert!(warnings.is_empty());
1554 }
1555
1556 #[test]
1557 fn reload_if_stale_short_circuits_for_unknown_mem_filter() {
1558 // Filtering by an unknown mem produces zero candidates;
1559 // the method returns an empty Vec without panicking.
1560 let tmp = TempDir::new().unwrap();
1561 let mut engine = build_demo_engine(&tmp);
1562 let warnings = engine.reload_if_stale(Some("does-not-exist"));
1563 assert!(warnings.is_empty());
1564 }
1565
1566 /// Test fixture: a `MemBackend` whose `current_head` and
1567 /// (read-side) entity surface are externally mutable so a test
1568 /// can simulate a sibling writer advancing the head between
1569 /// drift-check probes. Write methods are no-ops; the engine's
1570 /// drift-check path never invokes them.
1571 struct ManualHeadBackend {
1572 head: std::sync::Mutex<Option<String>>,
1573 entities: std::sync::Mutex<Vec<(PathBuf, Vec<u8>)>>,
1574 }
1575
1576 impl ManualHeadBackend {
1577 fn new(initial_head: Option<&str>) -> Self {
1578 Self {
1579 head: std::sync::Mutex::new(initial_head.map(String::from)),
1580 entities: std::sync::Mutex::new(Vec::new()),
1581 }
1582 }
1583
1584 fn set_head(&self, head: Option<&str>) {
1585 *self.head.lock().unwrap() = head.map(String::from);
1586 }
1587 }
1588
1589 impl MemBackend for ManualHeadBackend {
1590 fn list_entities(&self) -> Result<Vec<PathBuf>, BackendError> {
1591 Ok(self
1592 .entities
1593 .lock()
1594 .unwrap()
1595 .iter()
1596 .map(|(p, _)| p.clone())
1597 .collect())
1598 }
1599 fn read_entity(&self, rel: &Path) -> Result<Option<Vec<u8>>, BackendError> {
1600 Ok(self
1601 .entities
1602 .lock()
1603 .unwrap()
1604 .iter()
1605 .find(|(p, _)| p == rel)
1606 .map(|(_, b)| b.clone()))
1607 }
1608 fn write_entity(&self, _: &Path, _: &[u8]) -> Result<(), BackendError> {
1609 Ok(())
1610 }
1611 fn delete_entity(&self, _: &Path) -> Result<(), BackendError> {
1612 Ok(())
1613 }
1614 fn move_entity(&self, _: &Path, _: &Path) -> Result<(), BackendError> {
1615 Ok(())
1616 }
1617 fn commit(
1618 &self,
1619 _: &str,
1620 _: &CommitContext<'_>,
1621 ) -> Result<crate::storage::CommitId, BackendError> {
1622 Ok("synthetic".to_string())
1623 }
1624 fn append_provenance(&self, _: &Provenance) -> Result<(), BackendError> {
1625 Ok(())
1626 }
1627 fn read_provenance(&self, _: Option<&str>) -> Result<Vec<Provenance>, BackendError> {
1628 Ok(Vec::new())
1629 }
1630 fn current_head(&self) -> Result<Option<String>, BackendError> {
1631 Ok(self.head.lock().unwrap().clone())
1632 }
1633 }
1634
1635 #[test]
1636 fn reload_if_stale_emits_mem_reloaded_when_head_advances() {
1637 // Use an Arc<ManualHeadBackend> so the test retains a handle
1638 // for mutation after the engine has taken ownership of a
1639 // Box<dyn MemBackend> wrapper around it.
1640 struct ArcBackend(std::sync::Arc<ManualHeadBackend>);
1641 impl MemBackend for ArcBackend {
1642 fn list_entities(&self) -> Result<Vec<PathBuf>, BackendError> {
1643 self.0.list_entities()
1644 }
1645 fn read_entity(&self, rel: &Path) -> Result<Option<Vec<u8>>, BackendError> {
1646 self.0.read_entity(rel)
1647 }
1648 fn write_entity(&self, p: &Path, b: &[u8]) -> Result<(), BackendError> {
1649 self.0.write_entity(p, b)
1650 }
1651 fn delete_entity(&self, p: &Path) -> Result<(), BackendError> {
1652 self.0.delete_entity(p)
1653 }
1654 fn move_entity(&self, f: &Path, t: &Path) -> Result<(), BackendError> {
1655 self.0.move_entity(f, t)
1656 }
1657 fn commit(
1658 &self,
1659 m: &str,
1660 c: &CommitContext<'_>,
1661 ) -> Result<crate::storage::CommitId, BackendError> {
1662 self.0.commit(m, c)
1663 }
1664 fn append_provenance(&self, r: &Provenance) -> Result<(), BackendError> {
1665 self.0.append_provenance(r)
1666 }
1667 fn read_provenance(&self, c: Option<&str>) -> Result<Vec<Provenance>, BackendError> {
1668 self.0.read_provenance(c)
1669 }
1670 fn current_head(&self) -> Result<Option<String>, BackendError> {
1671 self.0.current_head()
1672 }
1673 }
1674
1675 let shared = std::sync::Arc::new(ManualHeadBackend::new(Some("aaa")));
1676 let backend = Box::new(ArcBackend(shared.clone()));
1677 let mount = Mount {
1678 mem: "specs".to_string(),
1679 schema: Some(pin("default")),
1680 storage: MountStorage::Folder {
1681 path: PathBuf::from("/dev/null"),
1682 },
1683 capability: MountCapability::Write,
1684 lifecycle: MountLifecycle::Eager,
1685 cross_linkable: true,
1686 migration_target: None,
1687 };
1688 let mut engine = Engine::from_mounts(vec![(mount, backend)]).unwrap();
1689
1690 // No drift on first probe — cached==new.
1691 let warnings = engine.reload_if_stale(Some("specs"));
1692 assert!(warnings.is_empty());
1693
1694 // Sibling writer advances the head.
1695 shared.set_head(Some("bbb"));
1696
1697 let warnings = engine.reload_if_stale(Some("specs"));
1698 assert_eq!(warnings.len(), 1);
1699 match &warnings[0] {
1700 crate::ops::WarningHint::MemReloaded {
1701 mem,
1702 old_head,
1703 new_head,
1704 ..
1705 } => {
1706 assert_eq!(mem, "specs");
1707 assert_eq!(old_head, "aaa");
1708 assert_eq!(new_head, "bbb");
1709 }
1710 other => panic!("expected MemReloaded, got {other:?}"),
1711 }
1712
1713 // Drift cleared — the engine's cached head now matches the
1714 // backend's current head; another probe is a no-op.
1715 let warnings = engine.reload_if_stale(Some("specs"));
1716 assert!(warnings.is_empty());
1717 }
1718
1719 #[test]
1720 fn mem_drifted_tracks_sibling_advance_until_reload() {
1721 // The read-only drift probe (built for the retired macOS app's roster): it reports
1722 // `true` once a sibling writer advances the backend past the
1723 // engine's cached head, *without* itself reloading, and clears
1724 // after the engine re-reads.
1725 struct ArcBackend(std::sync::Arc<ManualHeadBackend>);
1726 impl MemBackend for ArcBackend {
1727 fn list_entities(&self) -> Result<Vec<PathBuf>, BackendError> {
1728 self.0.list_entities()
1729 }
1730 fn read_entity(&self, rel: &Path) -> Result<Option<Vec<u8>>, BackendError> {
1731 self.0.read_entity(rel)
1732 }
1733 fn write_entity(&self, p: &Path, b: &[u8]) -> Result<(), BackendError> {
1734 self.0.write_entity(p, b)
1735 }
1736 fn delete_entity(&self, p: &Path) -> Result<(), BackendError> {
1737 self.0.delete_entity(p)
1738 }
1739 fn move_entity(&self, f: &Path, t: &Path) -> Result<(), BackendError> {
1740 self.0.move_entity(f, t)
1741 }
1742 fn commit(
1743 &self,
1744 m: &str,
1745 c: &CommitContext<'_>,
1746 ) -> Result<crate::storage::CommitId, BackendError> {
1747 self.0.commit(m, c)
1748 }
1749 fn append_provenance(&self, r: &Provenance) -> Result<(), BackendError> {
1750 self.0.append_provenance(r)
1751 }
1752 fn read_provenance(&self, c: Option<&str>) -> Result<Vec<Provenance>, BackendError> {
1753 self.0.read_provenance(c)
1754 }
1755 fn current_head(&self) -> Result<Option<String>, BackendError> {
1756 self.0.current_head()
1757 }
1758 }
1759
1760 let shared = std::sync::Arc::new(ManualHeadBackend::new(Some("aaa")));
1761 let backend = Box::new(ArcBackend(shared.clone()));
1762 let mount = Mount {
1763 mem: "specs".to_string(),
1764 schema: Some(pin("default")),
1765 storage: MountStorage::Folder {
1766 path: PathBuf::from("/dev/null"),
1767 },
1768 capability: MountCapability::Write,
1769 lifecycle: MountLifecycle::Eager,
1770 cross_linkable: true,
1771 migration_target: None,
1772 };
1773 let mut engine = Engine::from_mounts(vec![(mount, backend)]).unwrap();
1774
1775 // Fresh boot: cached == live, no drift.
1776 assert!(!engine.mem_drifted("specs").unwrap());
1777
1778 // Sibling writer advances the head — drift is visible WITHOUT a reload.
1779 shared.set_head(Some("bbb"));
1780 assert!(engine.mem_drifted("specs").unwrap());
1781 // Probing did not reload — still drifted on a second read.
1782 assert!(engine.mem_drifted("specs").unwrap());
1783
1784 // Re-reading through the engine clears it.
1785 let _ = engine.reload_if_stale(Some("specs"));
1786 assert!(!engine.mem_drifted("specs").unwrap());
1787
1788 // Unknown mem errors rather than reporting a bogus `false`.
1789 assert!(matches!(
1790 engine.mem_drifted("nope"),
1791 Err(EngineError::UnknownMem(_))
1792 ));
1793 }
1794
1795 #[test]
1796 fn reload_one_mem_report_head_before_is_prior_cursor_and_advances() {
1797 // Regression for the reload→changes_since recipe. `head_before`
1798 // must report the engine's PRIOR cursor (the SHA it last knew),
1799 // not the post-drift on-disk tip — otherwise
1800 // `changes_since(since=head_before)` spans an empty range in
1801 // exactly the sibling-drift case the recipe targets. The reload
1802 // must also advance the cursor to the new tip so the next
1803 // staleness probe is a no-op rather than a spurious reload.
1804 struct ArcBackend(std::sync::Arc<ManualHeadBackend>);
1805 impl MemBackend for ArcBackend {
1806 fn list_entities(&self) -> Result<Vec<PathBuf>, BackendError> {
1807 self.0.list_entities()
1808 }
1809 fn read_entity(&self, rel: &Path) -> Result<Option<Vec<u8>>, BackendError> {
1810 self.0.read_entity(rel)
1811 }
1812 fn write_entity(&self, p: &Path, b: &[u8]) -> Result<(), BackendError> {
1813 self.0.write_entity(p, b)
1814 }
1815 fn delete_entity(&self, p: &Path) -> Result<(), BackendError> {
1816 self.0.delete_entity(p)
1817 }
1818 fn move_entity(&self, f: &Path, t: &Path) -> Result<(), BackendError> {
1819 self.0.move_entity(f, t)
1820 }
1821 fn commit(
1822 &self,
1823 m: &str,
1824 c: &CommitContext<'_>,
1825 ) -> Result<crate::storage::CommitId, BackendError> {
1826 self.0.commit(m, c)
1827 }
1828 fn append_provenance(&self, r: &Provenance) -> Result<(), BackendError> {
1829 self.0.append_provenance(r)
1830 }
1831 fn read_provenance(&self, c: Option<&str>) -> Result<Vec<Provenance>, BackendError> {
1832 self.0.read_provenance(c)
1833 }
1834 fn current_head(&self) -> Result<Option<String>, BackendError> {
1835 self.0.current_head()
1836 }
1837 }
1838
1839 let shared = std::sync::Arc::new(ManualHeadBackend::new(Some("aaa")));
1840 let backend = Box::new(ArcBackend(shared.clone()));
1841 let mount = Mount {
1842 mem: "specs".to_string(),
1843 schema: Some(pin("default")),
1844 storage: MountStorage::Folder {
1845 path: PathBuf::from("/dev/null"),
1846 },
1847 capability: MountCapability::Write,
1848 lifecycle: MountLifecycle::Eager,
1849 cross_linkable: true,
1850 migration_target: None,
1851 };
1852 let mut engine = Engine::from_mounts(vec![(mount, backend)]).unwrap();
1853
1854 // Sibling writer advances the head past the engine's cursor.
1855 shared.set_head(Some("bbb"));
1856
1857 let report = engine.reload_one_mem_report("specs").unwrap();
1858 // head_before is the prior cursor "aaa", not the drifted tip.
1859 assert_eq!(report.head_before, "aaa");
1860 assert_eq!(report.head_after, "bbb");
1861
1862 // Cursor advanced to "bbb": a follow-up staleness probe is a
1863 // no-op, not a spurious MEM_RELOADED.
1864 let warnings = engine.reload_if_stale(Some("specs"));
1865 assert!(
1866 warnings.is_empty(),
1867 "cursor should have advanced to bbb, got {warnings:?}"
1868 );
1869 }
1870
1871 #[test]
1872 fn reload_if_stale_fires_every_call_no_throttle() {
1873 // Two back-to-back probes with the head advancing between
1874 // them: the second must reload and warn. There is no throttle
1875 // window — the ref check is the correctness floor.
1876 let shared = std::sync::Arc::new(ManualHeadBackend::new(Some("aaa")));
1877 struct ArcBackend(std::sync::Arc<ManualHeadBackend>);
1878 impl MemBackend for ArcBackend {
1879 fn list_entities(&self) -> Result<Vec<PathBuf>, BackendError> {
1880 self.0.list_entities()
1881 }
1882 fn read_entity(&self, rel: &Path) -> Result<Option<Vec<u8>>, BackendError> {
1883 self.0.read_entity(rel)
1884 }
1885 fn write_entity(&self, p: &Path, b: &[u8]) -> Result<(), BackendError> {
1886 self.0.write_entity(p, b)
1887 }
1888 fn delete_entity(&self, p: &Path) -> Result<(), BackendError> {
1889 self.0.delete_entity(p)
1890 }
1891 fn move_entity(&self, f: &Path, t: &Path) -> Result<(), BackendError> {
1892 self.0.move_entity(f, t)
1893 }
1894 fn commit(
1895 &self,
1896 m: &str,
1897 c: &CommitContext<'_>,
1898 ) -> Result<crate::storage::CommitId, BackendError> {
1899 self.0.commit(m, c)
1900 }
1901 fn append_provenance(&self, r: &Provenance) -> Result<(), BackendError> {
1902 self.0.append_provenance(r)
1903 }
1904 fn read_provenance(&self, c: Option<&str>) -> Result<Vec<Provenance>, BackendError> {
1905 self.0.read_provenance(c)
1906 }
1907 fn current_head(&self) -> Result<Option<String>, BackendError> {
1908 self.0.current_head()
1909 }
1910 }
1911
1912 let backend = Box::new(ArcBackend(shared.clone()));
1913 let mount = Mount {
1914 mem: "specs".to_string(),
1915 schema: Some(pin("default")),
1916 storage: MountStorage::Folder {
1917 path: PathBuf::from("/dev/null"),
1918 },
1919 capability: MountCapability::Write,
1920 lifecycle: MountLifecycle::Eager,
1921 cross_linkable: true,
1922 migration_target: None,
1923 };
1924 let mut engine = Engine::from_mounts(vec![(mount, backend)]).unwrap();
1925
1926 // First probe observes cached==new; no warning.
1927 let warnings = engine.reload_if_stale(Some("specs"));
1928 assert!(warnings.is_empty());
1929
1930 // Sibling advances head — the very next probe reloads and
1931 // warns, with no throttle window to mask it.
1932 shared.set_head(Some("bbb"));
1933 let warnings = engine.reload_if_stale(Some("specs"));
1934 assert_eq!(
1935 warnings.len(),
1936 1,
1937 "no throttle window — the moved ref reloads on the next probe"
1938 );
1939 }
1940}