memstead_base/engine/mutation/

update.rs

//! `Engine::update_entity` and `Engine::batch_update` — rewrite an
//! entity's sections / metadata in place, optimistically-locked.

use std::path::Path;

use crate::engine_fallback_type;
use crate::entity::EntityId;
use crate::entity::parser::parse_markdown;
use crate::entity::store_builder::push_entities_into_store;
use crate::ops::{ModifiedMetadata, ModifiedSections, WarningHint};
use crate::provenance::{Provenance, ProvenanceKind};
use crate::runtime_validator::{
    parse_metadata_value, validate_section_content, validate_section_keys,
    validate_unsettable_metadata_key, validate_updatable_section, validate_writable_metadata_key,
};
use crate::vcs::{Actor, ClientId, CommitContext};
use crate::workspace::MountCapability;

use super::super::{Engine, EngineError, UpdateEntityArgs, UpdateEntityOutcome};
use super::{
    PATCH_OLD_NOT_FOUND_CONTENT_CAP, make_stub, unknown_type_error,
    validate_relation_target_grammar,
};
use crate::engine::outcomes::RelationDeclared;
use crate::entity::{Entity, Relationship};

use std::sync::Arc;

/// Result of [`Engine::prepare_update`] — the validation + markdown
/// step split out of the commit so the batch path can prepare every
/// item before committing the whole set atomically.
enum PrepareOutcome {
    /// No commit is needed: the no-op short-circuit (content unchanged)
    /// and the dry-run preview both return a finished outcome here.
    Done(UpdateEntityOutcome),
    /// A real change whose post-mutation markdown is ready to stage +
    /// commit.
    Prepared(PreparedUpdate),
}

/// Everything the commit step needs to stage one prepared update's
/// disk write and build its outcome. Carries no commit SHA — that's
/// produced when the (single or batched) commit lands.
struct PreparedUpdate {
    mount_idx: usize,
    id: EntityId,
    mem: String,
    type_def: Arc<memstead_schema::TypeDefinition>,
    file_path: String,
    markdown: String,
    /// Body wiki-link targets the entity had *before* this mutation —
    /// the GC sweep scopes orphan-stub detection to these.
    prev_body_targets: std::collections::HashSet<EntityId>,
    modified_date: String,
    modified_sections: ModifiedSections,
    modified_metadata: ModifiedMetadata,
    warnings: Vec<WarningHint>,
    relations_declared: Vec<RelationDeclared>,
    /// Validated anchors to merge into this entity's sidecar row — staged
    /// into the same commit as the disk write on the commit step. Empty
    /// when the update carried no `anchors[]`.
    anchors: Vec<crate::anchor::Anchor>,
    /// Validated explicit anchor removals, applied before the `anchors`
    /// merge in the same staged write. Empty when the update carried no
    /// `anchors_unset[]`.
    anchor_unsets: Vec<crate::anchor::AnchorUnset>,
    /// True when this update's *sole* delta is the anchors sidecar —
    /// sections, metadata, and relationships are byte-identical to the
    /// on-disk entity. Such a commit earns the distinct
    /// `memstead: anchor <id>` subject (parsed `tool_verb == "anchor"`)
    /// so an anchor-only refresh — which the `_hash` excludes and which
    /// therefore produces zero entity deltas — is still observable to an
    /// `--include-notes` reader. A content-changing update (even one that
    /// also carries anchors) keeps the `memstead: update <id>` subject:
    /// its content change already bumps `_hash` and surfaces as a delta,
    /// so the anchor activity riding it is already visible.
    anchor_only: bool,
}

/// The store-side results of applying a prepared write — filled in
/// after the commit lands by [`Engine::apply_prepared_to_store`].
struct AppliedWrite {
    content_hash: String,
    title: String,
    orphan_stubs_removed: Vec<EntityId>,
}

impl Engine {
    /// Update an entity's sections and/or metadata.
    ///
    /// Same six-concern shape as [`Engine::create_entity`]. Optimistic
    /// locking via `args.expected_hash`: when `Some`, must match the
    /// store's current `content_hash` or returns
    /// [`EngineError::HashMismatch`]. The new engine's MCP-facing
    /// callers should always pass the hash; `None` is the
    /// `--force`-style escape hatch.
    ///
    /// Internally a two-step pipeline: [`Self::prepare_update`] runs
    /// all validation and computes the post-mutation markdown without
    /// committing, then [`Self::commit_prepared_update`] stages and
    /// commits the result. The split lets [`Self::batch_update`]
    /// prepare every item up front and commit the whole batch as one
    /// atomic unit.
    pub fn update_entity(
        &mut self,
        args: UpdateEntityArgs,
        actor: Actor,
        client: Option<&ClientId>,
        note: Option<&str>,
    ) -> Result<UpdateEntityOutcome, EngineError> {
        // Reload-before-operation: probe the mem ref and reload if a
        // sibling advanced it, so the `expected_hash` compare inside
        // `prepare_update` runs against current truth. A stale hash for
        // the targeted entity then trips a real `HASH_MISMATCH`; an
        // unrelated concurrent write leaves this entity's hash intact
        // and the update proceeds. The drift notice rides the outcome.
        let mut drift_warnings = self.reload_if_stale(Some(args.id.mem()));
        // Declared relations on an ACYCLIC rel-type (or one in an
        // `acyclic_sets` set, whose guard walks the set's UNION
        // subgraph) run the same whole-subgraph cycle guard relate
        // runs — full load first, or a cycle through a deferred mem's
        // edge is invisible (see the relate path's comment for the
        // demonstrated failure). Declared signals on the entity's
        // schema or any relation target's schema need the full load
        // too: the threshold-crossing diff counts edges that can
        // originate in any mem.
        if args.declare_relations.iter().any(|r| {
            self.schemas.get(args.id.mem()).is_some_and(|s| {
                s.relationship_acyclic(&r.rel_type)
                    || s.acyclic_set_containing(&r.rel_type).is_some()
            }) || self
                .schemas
                .get(r.target.mem())
                .is_some_and(|s| s.types.values().any(|td| !td.signals.is_empty()))
        }) || self
            .schemas
            .get(args.id.mem())
            .is_some_and(|s| s.types.values().any(|td| !td.signals.is_empty()))
        {
            self.ensure_mems_loaded(None);
        }
        let mut outcome = match self.prepare_update(args)? {
            PrepareOutcome::Done(outcome) => outcome,
            PrepareOutcome::Prepared(prepared) => {
                self.commit_prepared_update(prepared, actor, client, note)?
            }
        };
        drift_warnings.append(&mut outcome.warnings);
        outcome.warnings = drift_warnings;
        Ok(outcome)
    }

    /// Stage the prepared disk write, commit it as one commit, append
    /// provenance, and apply the change to the in-memory store — the
    /// single-update tail of [`Self::update_entity`]. The batch path
    /// drives the same steps but commits once across all items.
    fn commit_prepared_update(
        &mut self,
        prepared: PreparedUpdate,
        actor: Actor,
        client: Option<&ClientId>,
        note: Option<&str>,
    ) -> Result<UpdateEntityOutcome, EngineError> {
        // Aggregate signals: the entities an update can move are the
        // updated entity itself, the endpoints of every edge it adds
        // or removes (pre-write counterparts in both directions plus
        // the post-write markdown's targets), and — through the
        // neighbour filter — the counterparts of an entity whose
        // `neighbour_field` value the write changes (covered by the
        // same pre-write counterpart set). Captured before disk or
        // store mutate, diffed after the store applies.
        let signal_snapshot = {
            let mut candidates: Vec<EntityId> = vec![prepared.id.clone()];
            candidates.extend(
                self.store
                    .outgoing(&prepared.id)
                    .iter()
                    .map(|e| e.target.clone()),
            );
            candidates.extend(
                self.store
                    .incoming(&prepared.id)
                    .iter()
                    .map(|e| e.from.clone()),
            );
            if let Ok(parsed) = parse_markdown(
                &prepared.markdown,
                &prepared.file_path,
                prepared.type_def.as_ref(),
                &prepared.mem,
            ) {
                candidates.extend(parsed.entity.relationships.iter().map(|r| r.target.clone()));
            }
            crate::ops::signals::snapshot_levels(&self.store, &self.schemas, candidates.iter())
        };
        let backend = self.mounts[prepared.mount_idx].backend.as_ref();
        backend.write_entity(Path::new(&prepared.file_path), prepared.markdown.as_bytes())?;
        // Stage the anchors sidecar into the same commit as the entity
        // write. Only when the update carried anchors or anchor unsets.
        if !prepared.anchors.is_empty() || !prepared.anchor_unsets.is_empty() {
            super::stage_anchors_sidecar(
                backend,
                &prepared.id,
                &prepared.anchor_unsets,
                prepared.anchors.clone(),
            )?;
        }
        // Derivation baselines (agent-trust plan 12): declared
        // relations on a derivation rel-type record the target's
        // current hash, riding the same commit as the entity write.
        if let Some(schema) = self.schemas.get(prepared.id.mem()) {
            for r in prepared
                .relations_declared
                .iter()
                .filter(|r| super::rel_type_declares_derivation(schema, &r.rel_type))
            {
                let hash = self
                    .store
                    .get(&r.target)
                    .map(|e| e.content_hash.clone())
                    .unwrap_or_default();
                let (from, rel, to) = (
                    prepared.id.to_string(),
                    r.rel_type.clone(),
                    r.target.to_string(),
                );
                super::stage_derivation_sidecar(backend, |s| s.set(&from, &rel, &to, &hash))?;
            }
        }
        // Anchor-only commits carry the distinct `anchor` verb so their
        // otherwise-invisible sidecar change is legible in the note log;
        // every other update keeps `update`. The verb is a subject-only
        // signal — delta computation reads the tree diff, not the
        // subject, so the zero-entity-delta guarantee is untouched.
        let commit_subject = if prepared.anchor_only {
            format!("memstead: anchor {}", prepared.id)
        } else {
            format!("memstead: update {}", prepared.id)
        };
        let ctx = CommitContext {
            actor,
            client: client.cloned(),
            tool: Some("update_entity"),
            note: note.map(String::from),
            role: self.current_role,
            logical_operation_id: None,
            entity_ids: None,
        };
        let write_id = backend.commit(&commit_subject, &ctx)?;
        backend.append_provenance(
            &Provenance::new(
                std::time::SystemTime::now(),
                ProvenanceKind::Update,
                Some(prepared.id.to_string()),
                actor,
                client.cloned(),
                note.map(String::from),
            )
            .with_role(self.current_role),
        )?;
        self.record_self_write(prepared.mount_idx, &write_id);
        let stamp_warnings = self.stamp_mutation_versions(prepared.mount_idx);

        let applied = self.apply_prepared_to_store(&prepared)?;

        self.invalidate_communities();
        // Incremental (flywheel W8/01): the updated entity is the
        // whole touched set — declared-relation stubs are never
        // indexed.
        self.maintain_search_indexes(std::slice::from_ref(&prepared.id));

        // `require_notes` provenance nudge — single engine-level
        // enforcement point. Only reached on the real-commit path; the
        // no-op and dry-run prepare outcomes never demand a note.
        let mut warnings = prepared.warnings;
        warnings.extend(stamp_warnings);
        // Signal crossings — out-of-band beside the success payload,
        // never error-shaped.
        warnings.extend(crate::ops::signals::crossing_warnings(
            &self.store,
            &self.schemas,
            &signal_snapshot,
        ));
        if let Some(w) = self.note_missing_warning("update_entity", note) {
            warnings.push(w);
        }

        Ok(UpdateEntityOutcome {
            id: prepared.id.clone(),
            title: applied.title,
            file_path: prepared.file_path,
            content_hash: applied.content_hash,
            write_id,
            modified_date: prepared.modified_date,
            orphan_stubs_removed: applied.orphan_stubs_removed,
            modified_sections: prepared.modified_sections,
            modified_metadata: prepared.modified_metadata,
            prospective_hash: None,
            warnings,
            relations_declared: prepared.relations_declared,
        })
    }

    /// Parse the prepared markdown, push it into the in-memory store,
    /// re-map alias-target edge sources, and GC any stub the mutation
    /// orphaned. Shared post-commit store-application step for the
    /// single-update and batch paths — does NOT touch the backend or
    /// commit (the caller has already staged + committed the disk
    /// write).
    fn apply_prepared_to_store(
        &mut self,
        prepared: &PreparedUpdate,
    ) -> Result<AppliedWrite, EngineError> {
        let parse_result = parse_markdown(
            &prepared.markdown,
            &prepared.file_path,
            prepared.type_def.as_ref(),
            &prepared.mem,
        )
        .map_err(|e| EngineError::ParseAfterWrite(e.to_string()))?;
        let content_hash = parse_result.entity.content_hash.clone();
        let title = parse_result.entity.title.clone();
        let fallback = engine_fallback_type();
        push_entities_into_store(&mut self.store, vec![parse_result], fallback.as_ref(), None);
        crate::entity::store_builder::remap_alias_target_edge_sources(
            &mut self.store,
            &self.schemas,
        );
        let orphan_stubs_removed =
            super::gc_orphan_stubs_among(&mut self.store, &prepared.prev_body_targets);
        Ok(AppliedWrite {
            content_hash,
            title,
            orphan_stubs_removed,
        })
    }

    /// Validate an update and compute its post-mutation markdown
    /// *without* committing. Returns [`PrepareOutcome::Done`] for the
    /// no-op / dry-run short-circuits (which never commit) and
    /// [`PrepareOutcome::Prepared`] for a real change whose write the
    /// caller stages + commits. May mutate the store in place via the
    /// alias-synthesis auto-stub upsert; the batch path snapshots the
    /// store before preparing so a refused batch can roll that back.
    fn prepare_update(&mut self, args: UpdateEntityArgs) -> Result<PrepareOutcome, EngineError> {
        let id = &args.id;
        let mem = id.mem().to_string();

        let mount_idx = self
            .mounts
            .iter()
            .position(|m| m.mount.mem == mem)
            .ok_or_else(|| self.unknown_mem_error(&mem))?;
        if self.mounts[mount_idx].mount.capability != MountCapability::Write {
            return Err(EngineError::ReadOnlyMount(mem));
        }

        let entity = self
            .store
            .get(id)
            .ok_or_else(|| EngineError::NotFound { id: id.to_string() })?;

        // Snapshot the prev entity's body wiki-link set before any
        // subsequent `&mut self` reborrow burns the `entity` borrow.
        // Fed to the alias-synthesis pass so the GC step can compare
        // prev vs. next body links and drop pointer-rel-type relations
        // whose target was a body link before but isn't any more.
        let prev_body_targets = super::collect_body_link_targets(entity);

        // Stub guard — stubs have no body, no metadata, no
        // schema-resolved type to validate against. The recovery is
        // `memstead_create` (stub adoption preserves incoming
        // references). Pre-Item-02 the update path fell through to
        // the `type_def` lookup below and surfaced the cryptic
        // `UnknownType { name: "" }` cascade. Mirrors the
        // `StubCannotRelate` guard on `memstead_relate`.
        if entity.stub {
            return Err(EngineError::StubNotUpdatable { id: id.to_string() });
        }

        // Skip the hash check on dry_run — full's dry_run is the
        // designated stale-hash recovery path. Agents preview a
        // change without holding a fresh hash, get back the current
        // `content_hash` and a `prospective_hash`, then call the
        // real update with `expected_hash = content_hash`.
        if !args.dry_run
            && let Some(expected) = args.expected_hash.as_deref()
            && entity.content_hash != expected
        {
            return Err(EngineError::HashMismatch {
                id: id.to_string(),
                current: entity.content_hash.clone(),
                is_stub: entity.stub,
            });
        }

        // Empty-mutation guard. After existence/stub/hash
        // gates so the more-specific errors fire first. A payload
        // with no recognised mutation content refuses BEFORE any
        // mutation work runs so a misspelled or omitted mutation
        // key (which deserialised to empty defaults under the
        // lenient pre-fix posture) doesn't silently land as
        // `succeeded: N, action: "updated", write_id: ""`.
        // Distinct from `UPDATE_NOOP` (a warning that fires when
        // mutation content was provided but matched the current
        // entity state) — the two are different states and ship
        // different envelopes.
        if args.sections.is_empty()
            && args.append_sections.is_empty()
            && args.patch_sections.is_empty()
            && args.metadata.is_empty()
            && args.metadata_unset.is_empty()
            && args.declare_relations.is_empty()
            && args.relations_unset.is_empty()
            && args.anchors.is_empty()
            && args.anchors_unset.is_empty()
        {
            return Err(EngineError::EmptyUpdate { id: id.to_string() });
        }

        // Validate any `anchors[]` / `anchors_unset[]` payload up front —
        // a malformed element refuses the whole update with a typed
        // `INVALID_ANCHOR` envelope before any disk write, on every path
        // (real / dry-run / no-op). Empty payloads → empty vecs (no
        // sidecar write; byte-identical to a pre-anchor update).
        let validated_anchors = self.validate_anchor_inputs(&mem, &args.anchors)?;
        let validated_anchor_unsets = Self::validate_anchor_unsets(&args.anchors_unset)?;

        let schema = self
            .schemas
            .get(&mem)
            .expect("schema present for every registered mount")
            .clone();
        let type_def = schema
            .get_type(&entity.entity_type)
            .ok_or_else(|| unknown_type_error(schema.as_ref(), &entity.entity_type))?;

        // Mode-conflict: the same section key may not appear in
        // more than one of `sections`, `append_sections`,
        // `patch_sections`. Mirrors full's
        // `EngineError::ConflictingSectionModes`. Three-way check:
        // build the conflict list per key and reject when ≥2 modes
        // claim it.
        for key in args.sections.keys() {
            let mut modes = vec!["sections".to_string()];
            if args.append_sections.contains_key(key) {
                modes.push("append_sections".to_string());
            }
            if args.patch_sections.contains_key(key) {
                modes.push("patch_sections".to_string());
            }
            if modes.len() > 1 {
                return Err(EngineError::ConflictingSectionModes {
                    section: key.clone(),
                    modes,
                });
            }
        }
        for key in args.append_sections.keys() {
            if args.patch_sections.contains_key(key) {
                return Err(EngineError::ConflictingSectionModes {
                    section: key.clone(),
                    modes: vec!["append_sections".to_string(), "patch_sections".to_string()],
                });
            }
        }

        validate_section_keys(
            args.sections
                .keys()
                .chain(args.append_sections.keys())
                .chain(args.patch_sections.keys())
                .map(String::as_str),
            type_def.as_ref(),
        )?;
        let mut heading_buf: Vec<&str> = Vec::new();
        #[allow(unused_assignments)]
        let mut catch_all = None;
        // Refuse embedded `^## ` in section content on every update path
        // that writes section bodies: `sections` (replace) and
        // `append_sections` (append). `patch_sections` replaces a
        // substring — its `new` text feeds into the eventual section
        // body so it gets the same gate.
        validate_section_content(
            args.sections
                .iter()
                .map(|(k, v)| (k.as_str(), v.as_str()))
                .chain(
                    args.append_sections
                        .iter()
                        .map(|(k, v)| (k.as_str(), v.as_str())),
                )
                .chain(
                    args.patch_sections
                        .iter()
                        .map(|(k, p)| (k.as_str(), p.new.as_str())),
                ),
            {
                let t: &memstead_schema::TypeDefinition = type_def.as_ref();
                catch_all = crate::runtime_validator::catch_all_context(t, &mut heading_buf);
                catch_all
            },
        )?;
        for key in args.sections.keys() {
            validate_updatable_section(key.as_str(), type_def.as_ref())?;
        }
        for key in args.append_sections.keys() {
            validate_updatable_section(key.as_str(), type_def.as_ref())?;
        }
        for key in args.patch_sections.keys() {
            validate_updatable_section(key.as_str(), type_def.as_ref())?;
        }
        for key in args.metadata.keys() {
            validate_writable_metadata_key(key.as_str(), type_def.as_ref())?;
        }
        // Unset has its own gate: the reserved `mem`/`id`/`type` triple
        // is unset-ALLOWED (the sanctioned repair for entities that
        // acquired a smuggled reserved key before the write gates
        // closed) while engine-stamped timestamp fields stay refused —
        // see `validate_unsettable_metadata_key`.
        for key in &args.metadata_unset {
            validate_unsettable_metadata_key(key.as_str(), type_def.as_ref())?;
        }

        // Reject the same key appearing in `metadata` (set) and
        // `metadata_unset` — the wire contract is that the conflict is
        // a hard error. Caught before any required-field /
        // parse-metadata check so the resolution ("pick one map") is
        // unambiguous regardless of whether the overlapping key is
        // required.
        let mut overlap: Vec<String> = args
            .metadata
            .keys()
            .filter(|k| args.metadata_unset.iter().any(|u| u == k.as_str()))
            .cloned()
            .collect();
        if !overlap.is_empty() {
            overlap.sort();
            overlap.dedup();
            return Err(EngineError::SetAndUnsetConflict { keys: overlap });
        }

        // Repair-power gate:
        // repair-shaped input is accepted only when the entity
        // currently fails the conformance check against the effective
        // schema. Conformance is per-entity-local and cheap, so the
        // gate runs in pre-validation; no agent-settable flag exists —
        // the entity's own state is the evidence. A pure-consistency
        // break does not open the gate (those have ungated repair
        // paths: `memstead_relate(remove)` and the additive params).
        if !args.relations_unset.is_empty() {
            let findings = crate::ops::integrity::entity_conformance_findings(
                &self.store,
                entity,
                schema.as_ref(),
                &self.schemas,
            );
            if findings.is_empty() {
                return Err(EngineError::RepairNotNeeded {
                    id: id.to_string(),
                    recovery: "use memstead_relate(remove=true) to detach an edge from a                                conformant entity, or the additive memstead_update params                                to evolve it"
                        .to_string(),
                });
            }
        }

        let mut next = entity.clone();

        // Repair-shaped removals, applied before declarations so a
        // repair can drop and re-shape relations in one atomic
        // update. Absent (rel_type, target) pairs are silent no-ops,
        // symmetric with `metadata_unset`. The strict post-state
        // validation below still runs — repair widens accepted
        // inputs, never admissible outputs.
        for unset in &args.relations_unset {
            let canonical = crate::entity::id::validate_rel_type(&unset.rel_type)
                .unwrap_or_else(|_| unset.rel_type.clone());
            next.relationships
                .retain(|r| !(r.rel_type == canonical && r.target == unset.target));
        }

        // Atomic batched relation declarations. Validated and applied
        // before the section/metadata changes so the strict
        // wiki-link/relation validator at the end of this fn sees
        // the freshly-declared relations as part of the post-state.
        // Same vocabulary + shape + grammar gates `memstead_relate`
        // runs; auto-stubs absent Write-mem targets identically
        // to the relate path. Returns the (rel_type, target,
        // target_was_stubbed) triples in `relations_declared` on
        // the outcome so the agent sees what landed.
        let relations_declared = apply_declare_relations(
            self,
            &mut next,
            &args.declare_relations,
            &mem,
            mount_idx,
            type_def.as_ref(),
            schema.as_ref(),
        )?;

        // Keys this update touches, captured before the args maps are
        // consumed — the section-format evaluation below judges
        // exactly these on their composed bodies.
        let format_touched: std::collections::HashSet<String> = args
            .sections
            .keys()
            .chain(args.append_sections.keys())
            .chain(args.patch_sections.keys())
            .cloned()
            .collect();

        let mut modified_sections: Vec<String> = Vec::new();
        for (key, body) in args.sections {
            modified_sections.push(key.clone());
            next.sections.insert(key, body);
        }

        // Apply append_sections after replace. Empty/absent body
        // is replaced wholesale with the append value; otherwise
        // a `\n` separator joins the two. Mirrors full.
        let mut modified_sections_appended: Vec<String> = Vec::new();
        for (key, value) in args.append_sections {
            let existing = next.sections.get(&key).cloned().unwrap_or_default();
            let new_content = if existing.trim().is_empty() {
                value
            } else {
                format!("{existing}\n{value}")
            };
            next.sections.insert(key.clone(), new_content);
            modified_sections_appended.push(key);
        }

        // Apply patch_sections after append. Find-and-replace the
        // `old` substring with `new`; `all` flips between
        // first-occurrence (replacen 1) and every-occurrence
        // (replace). Empty/absent section is rejected with
        // PatchSectionEmpty; missing-`old` rejected with
        // PatchOldNotFound carrying a UTF-8-safe truncated snapshot
        // of the current body. Mirrors full.
        let mut modified_sections_patched: Vec<String> = Vec::new();
        for (key, patch) in args.patch_sections {
            let existing = next
                .sections
                .get(&key)
                .ok_or_else(|| EngineError::PatchSectionEmpty {
                    section: key.clone(),
                })?
                .clone();
            if !existing.contains(&patch.old) {
                let cap = PATCH_OLD_NOT_FOUND_CONTENT_CAP;
                let truncated = existing.len() > cap;
                // Truncate at a UTF-8 char boundary to avoid
                // splitting a code point.
                let mut cut = cap.min(existing.len());
                while cut > 0 && !existing.is_char_boundary(cut) {
                    cut -= 1;
                }
                let current_content = if truncated {
                    existing[..cut].to_string()
                } else {
                    existing.clone()
                };
                return Err(EngineError::PatchOldNotFound {
                    section: key,
                    current_content,
                    truncated,
                });
            }
            let patched = if patch.all {
                existing.replace(&patch.old, &patch.new)
            } else {
                existing.replacen(&patch.old, &patch.new, 1)
            };
            next.sections.insert(key.clone(), patched);
            modified_sections_patched.push(key);
        }

        let mut modified_metadata_set: Vec<String> = Vec::new();
        for (key, value) in &args.metadata {
            let parsed = parse_metadata_value(key.as_str(), value.as_str(), type_def.as_ref())?;
            modified_metadata_set.push(key.clone());
            next.metadata.insert(key.clone(), parsed);
        }

        let mut modified_metadata_unset: Vec<String> = Vec::new();
        for key in args.metadata_unset {
            // Reserved identity/discriminator keys bypass the
            // required-field gate below: unsetting one is the
            // sanctioned repair for a historically smuggled key, and
            // it can only move the entity toward the invariant. For
            // `type` the engine immediately re-seeds the authoritative
            // discriminator from the entity's own type — the
            // frontmatter can never go typeless (a missing `type:`
            // would silently re-type the entity to the mem's default
            // on the next parse), so on a healthy entity the unset is
            // a no-op. `mem`/`id` are never engine-seeded in the map;
            // removing a smuggled one is a real (recorded) removal.
            if crate::runtime_validator::READ_ONLY_METADATA_KEYS.contains(&key.as_str()) {
                if key == "type" {
                    let authoritative =
                        crate::entity::MetadataValue::String(next.entity_type.clone());
                    if next
                        .metadata
                        .shift_remove("type")
                        .is_some_and(|removed| removed != authoritative)
                    {
                        modified_metadata_unset.push(key);
                    }
                    next.metadata.insert("type".to_string(), authoritative);
                } else if next.metadata.shift_remove(&key).is_some() {
                    modified_metadata_unset.push(key);
                }
                continue;
            }
            // Reject unset on required fields — the pre-remove check
            // carries the recovery payload (field_description,
            // enum_values, type_write_rules) so MCP envelopes surface
            // the full REQUIRED_FIELD_UNSET shape.
            let field_def = type_def.metadata_field(&key);
            let is_required = field_def.map(|f| f.is_required()).unwrap_or(false);
            if is_required {
                let (field_description, enum_values) = match field_def {
                    Some(f) => (
                        Some(f.description.clone()),
                        f.enum_values.clone().unwrap_or_default(),
                    ),
                    None => (None, Vec::new()),
                };
                return Err(EngineError::RequiredFieldUnset {
                    field: key,
                    entity_type: type_def.name.clone(),
                    field_description,
                    enum_values,
                    type_write_rules: type_def.write_rules.clone(),
                    // Update path — caller passed
                    // `metadata_unset: ["field"]` against a required
                    // field. The wording ("cannot unset required
                    // field …") is semantically correct for this
                    // path.
                    on_create: false,
                    // The unset path targets one field per call by
                    // definition, so the multi-field accumulator
                    // stays empty here — the singular fields above
                    // are authoritative.
                    missing: Vec::new(),
                });
            }
            if next.metadata.shift_remove(&key).is_some() {
                modified_metadata_unset.push(key);
            }
        }

        // Delay the auto-stamp until AFTER the no-op short-circuit. Pre-fix
        // this branch overwrote `last_modified` with `today_iso()`
        // before the bytes-compare, so the prospective markdown
        // always differed from the on-disk bytes (the schema's
        // `last_modified` was already populated with a full ISO
        // timestamp on the prior write, but `today_iso()` returns
        // date-only), and the no-op compare never matched. Compute
        // `today` for later use but don't stamp `next` yet.
        let today = self.now_iso();

        // Alias-synthesis pass: for schemas declaring
        // `alias_target_rel_type`, append engine-emitted relations of
        // that rel-type for every body wiki-link not already backed,
        // and GC pointer-rel-type relations whose target was a body
        // wiki-link in the prev state but isn't in next. Cross-mem
        // refusal aborts the update — no partial state.
        //
        // The returned `Vec<Relationship>` is the per-call set of
        // synthesised relations; feed it into the auto-stub warning
        // emission below.
        let (synthesised_relations, self_link_ignored) =
            super::synthesise_alias_relations(self, &prev_body_targets, &mut next)?;

        // Alias-existence invariant: every body wiki-link must be
        // backed by an entry in `entity.relationships`. The validator
        // runs against the *full* post-mutation state (not just the
        // delta), so a mutation that leaves an existing unbacked link
        // in place still fails — forcing cleanup of historical drift.
        let missing = super::scan_wikilinks_without_relation(&next)?;
        if !missing.is_empty() {
            return Err(EngineError::WikiLinkWithoutRelation {
                from_id: id.to_string(),
                missing: missing
                    .into_iter()
                    .map(|(section_key, target)| crate::engine::MissingWikiLink {
                        section_key,
                        target_id: target.to_string(),
                    })
                    .collect(),
            });
        }

        let file_path = next.file_path.clone();

        // The bytes-compare runs against the pre-stamp markdown so the
        // auto-timestamp doesn't synthesise a false delta. When the
        // user-visible payload (sections, user-set metadata,
        // declared relations) didn't change, the pre-stamp markdown
        // matches the on-disk bytes byte-for-byte; we short-circuit
        // and return with `last_modified` preserved at its pre-call
        // value. Real changes fall through; we stamp + regenerate
        // below.
        let markdown_pre_stamp = super::render_for_write(&next, type_def.as_ref())?;

        // Whether the user-visible content (sections, user-set metadata,
        // declared relations) is byte-identical to the on-disk entity —
        // the pre-stamp markdown's hash matches the current
        // `content_hash`. When this holds past the no-op guard below, the
        // *only* remaining delta is the anchors sidecar (an anchor-only
        // update). `_hash` excludes anchors, so an anchor-only commit
        // produces zero entity deltas; the distinct `anchor` verb is the
        // compensating signal that makes it observable. `next.content_hash`
        // is the on-disk value (cloned from the source entity and never
        // recomputed since — same rationale as the no-op branch reads it).
        let content_unchanged =
            crate::entity::parser::compute_hash(&markdown_pre_stamp) == next.content_hash;

        // No-op short-circuit. When the pre-stamp markdown's hash
        // matches the entity's current `content_hash`, the
        // post-mutation user-visible state equals the on-disk state.
        // Skip the disk write, the commit, the provenance append,
        // and the store re-parse. Mirrors `relate.rs`'s
        // `NoOpAlreadyPresent` / `NoOpAbsent` and `rename.rs`'s
        // slug-noop short-circuits: empty `write_id`, unchanged
        // `content_hash`, preserved `last_modified`, typed
        // `UpdateNoop` warning. Skipped on the dry_run path because
        // the dry_run preview semantics document a separate
        // non-committing shape with `prospective_hash: Some(_)` and
        // the unchanged `content_hash`; conflating the two would
        // lose the prospective-hash channel callers use to chain a
        // follow-up real update with `expected_hash`.
        if !args.dry_run {
            // An update carrying `anchors[]` or `anchors_unset[]` is never
            // a no-op even when the entity content is unchanged — the
            // anchors sidecar must be written, so fall through to the
            // real-commit path.
            if content_unchanged
                && validated_anchors.is_empty()
                && validated_anchor_unsets.is_empty()
            {
                // No-op: report the preserved `last_modified` from
                // the pre-stamp `next` (which still carries the
                // entity's on-disk value because we haven't run the
                // auto-stamp yet on this branch).
                let modified_date = next
                    .metadata
                    .get("last_modified")
                    .and_then(|v| v.as_str().map(str::to_string))
                    .unwrap_or_default();
                return Ok(PrepareOutcome::Done(UpdateEntityOutcome {
                    id: id.clone(),
                    title: next.title.clone(),
                    file_path,
                    content_hash: next.content_hash.clone(),
                    write_id: String::new(),
                    modified_date,
                    // No-op: the prospective hash equals the on-disk hash,
                    // so nothing landed. `modified_*` report the *applied*
                    // delta (empty), consistent with the empty `write_id`
                    // and unchanged hash — not the request-derived keys,
                    // which would claim a change that did not happen
                    // (F1). The request vecs
                    // (`modified_metadata_set` etc.) are intentionally
                    // dropped on this branch.
                    modified_sections: ModifiedSections::default(),
                    modified_metadata: ModifiedMetadata::default(),
                    prospective_hash: None,
                    // No write happened on the no-op path, so nothing
                    // could have orphaned a stub.
                    orphan_stubs_removed: Vec::new(),
                    warnings: vec![WarningHint::UpdateNoop { id: id.clone() }],
                    relations_declared,
                }));
            }
        }

        // Real change: apply the auto-stamp now and regenerate the
        // markdown so the subsequent hash + write reflect it.
        // Exception — the anchor-only leg (`content_unchanged` true,
        // reachable only because anchors/unsets are present): the
        // sidecar is the sole delta and `_hash` excludes it, so the
        // entity bytes must stay untouched. Stamping here would move
        // `last_modified` (and with it `_hash`) whenever the update
        // lands in a different second than the previous write —
        // breaking the "anchors never move `_hash`" contract.
        if !content_unchanged {
            super::auto_stamp_timestamps(&mut next, type_def.as_ref(), &today);
        }
        let markdown = super::render_for_write(&next, type_def.as_ref())?;

        let mut warnings: Vec<WarningHint> = Vec::new();

        // Heading-divergence check: when a written section's declared
        // heading differs from a heading the file already carried for
        // the same key (derives to it), warn — the write commits and
        // the regenerated file replaces the old heading text, which
        // the caller should see rather than discover on the next read.
        // `next.raw_section_headings` is the pre-mutation parse
        // artefact (cloned from the store entity; nothing in this
        // pipeline rewrites it).
        for key in modified_sections
            .iter()
            .chain(modified_sections_appended.iter())
            .chain(modified_sections_patched.iter())
        {
            let Some(def) = type_def.section(key) else {
                continue;
            };
            if let Some(existing) = next.raw_section_headings.iter().find(|h| {
                h.as_str() != def.heading && memstead_schema::derive_section_key(h) == *key
            }) {
                warnings.push(WarningHint::SectionHeadingDivergence {
                    entity_id: id.clone(),
                    section_key: key.clone(),
                    writing_heading: def.heading.clone(),
                    existing_heading: existing.clone(),
                });
            }
        }

        // Required-outgoing evaluation — the warning the tool
        // descriptions have promised all along. `next` carries this
        // update's final edge set (declared relations applied,
        // alias-synthesis run), evaluated through the same function
        // the health sweep uses (one implementation; the two surfaces
        // cannot disagree). A warning, never a refusal.
        // Section-format evaluation (plan 08), composed-body rule: a
        // section touched by this update (replace, append, or patch)
        // is judged on its COMPOSED final body — the delta-only
        // byte-class guard keeps its scope, shape needs the
        // composition point. Untouched sections stay lenient (their
        // pre-existing violations are health findings; the next write
        // is the sanctioned repair point).
        for def in &type_def.sections {
            if def.format_severity != memstead_schema::ConstraintSeverity::Block {
                continue;
            }
            if !format_touched.contains(def.key.as_str()) {
                continue;
            }
            let Some(body) = next.sections.get(def.key.as_str()) else {
                continue;
            };
            if let Some(first) = crate::section_format::check_section_format(def, body)
                .into_iter()
                .next()
            {
                return Err(EngineError::SectionFormatRefused {
                    entity_type: next.entity_type.clone(),
                    entity_id: id.to_string(),
                    violation: first,
                });
            }
        }

        let unsatisfied =
            crate::ops::health::unsatisfied_required_outgoing(&next, type_def.as_ref());
        if !unsatisfied.is_empty() {
            // `severity: block` promotes the warning to a refusal —
            // evaluated in the prepare step, before any disk write or
            // commit, so a refused update leaves nothing behind.
            let blocked: Vec<_> = unsatisfied
                .iter()
                .filter(|b| b.severity == memstead_schema::ConstraintSeverity::Block)
                .cloned()
                .collect();
            if !blocked.is_empty() {
                return Err(EngineError::RequiredOutgoingUnsatisfied {
                    entity_type: next.entity_type.clone(),
                    entity_id: id.to_string(),
                    missing: blocked,
                });
            }
            warnings.push(WarningHint::MissingRequiredOutgoing {
                entity_type: next.entity_type.clone(),
                entity_id: id.clone(),
                missing: unsatisfied,
            });
        }

        // Declared-constraints evaluation — same single evaluation the
        // health `constraints` include runs, against this update's
        // final state. Block-tier violations refuse; warn-tier warn.
        let violated = crate::ops::health::unsatisfied_constraints(
            &self.store,
            &next,
            type_def.as_ref(),
            Some(id),
        );
        if !violated.is_empty() {
            let blocked: Vec<_> = violated
                .iter()
                .filter(|v| v.severity() == memstead_schema::ConstraintSeverity::Block)
                .cloned()
                .collect();
            if !blocked.is_empty() {
                return Err(EngineError::ConstraintUnsatisfied {
                    entity_type: next.entity_type.clone(),
                    entity_id: id.to_string(),
                    violations: blocked,
                });
            }
            warnings.push(WarningHint::ConstraintUnsatisfied {
                entity_type: next.entity_type.clone(),
                entity_id: id.clone(),
                violations: violated,
            });
        }

        // Mirror the create-path emission shape — drive the warning from
        // the synthesised relations the alias pass just emitted, not
        // from a re-parse of the generated markdown. `parse_markdown`
        // filters its `inline_links` against the entity's
        // `relationships` vec (which the synthesis pass has already
        // appended to), so the pre-fix path saw `inline_links: []`
        // and silently dropped the warning the docstring promises.
        let auto_stubbed: Vec<EntityId> = synthesised_relations
            .iter()
            .filter_map(|rel| {
                if !self.store.contains(&rel.target) {
                    Some(rel.target.clone())
                } else {
                    None
                }
            })
            .collect();
        if !auto_stubbed.is_empty() {
            warnings.push(WarningHint::InlineWikiLinkAutoStubbed {
                from: id.clone(),
                stubs: auto_stubbed,
            });
        }
        // F11: surface a dropped self-referential body link (the alias
        // pass omitted the vacuous self-edge).
        if self_link_ignored {
            warnings.push(WarningHint::SelfLinkIgnored { id: id.clone() });
        }

        // Dry-run: compute prospective hash from the in-memory
        // entity and return without touching disk, store, or
        // commits. Mirrors full's `UpdateArgs.dry_run` semantics —
        // `content_hash` carries the unchanged on-disk hash so the
        // caller can use it as `expected_hash` on the follow-up
        // real call (designated stale-hash recovery path).
        if args.dry_run {
            let prospective = crate::entity::parser::compute_hash(&markdown);
            // `next.content_hash` was cloned from the source
            // entity and not modified since; equals the on-disk
            // value.
            let current_hash = next.content_hash.clone();
            let modified_date = if type_def.metadata_fields.iter().any(|f| f.auto_timestamp) {
                today.clone()
            } else {
                String::new()
            };
            return Ok(PrepareOutcome::Done(UpdateEntityOutcome {
                id: id.clone(),
                title: next.title.clone(),
                file_path,
                content_hash: current_hash,
                write_id: String::new(),
                modified_date,
                modified_sections: ModifiedSections {
                    replaced: modified_sections,
                    appended: modified_sections_appended,
                    patched: modified_sections_patched,
                },
                modified_metadata: ModifiedMetadata {
                    set: modified_metadata_set,
                    unset: modified_metadata_unset,
                },
                prospective_hash: Some(prospective),
                // Dry-run touches neither store nor disk, so no stub
                // could have been GC'd.
                orphan_stubs_removed: Vec::new(),
                warnings,
                relations_declared: relations_declared.clone(),
            }));
        }

        // Real change prepared. Compute `modified_date` (mirrors full's
        // UpdateResult.modified_date — the `today` the auto-stamp loop
        // used; empty when the schema has no auto_timestamp field), then
        // hand the staged write to the caller to commit. The single
        // path commits immediately; the batch path commits the whole
        // set at once.
        let modified_date = if content_unchanged {
            // Anchor-only: nothing was stamped; report the preserved
            // on-entity value, mirroring the no-op branch.
            next.metadata
                .get("last_modified")
                .and_then(|v| v.as_str().map(str::to_string))
                .unwrap_or_default()
        } else if type_def.metadata_fields.iter().any(|f| f.auto_timestamp) {
            today.clone()
        } else {
            String::new()
        };

        Ok(PrepareOutcome::Prepared(PreparedUpdate {
            mount_idx,
            id: id.clone(),
            mem,
            type_def,
            file_path,
            markdown,
            prev_body_targets,
            modified_date,
            modified_sections: ModifiedSections {
                replaced: modified_sections,
                appended: modified_sections_appended,
                patched: modified_sections_patched,
            },
            modified_metadata: ModifiedMetadata {
                set: modified_metadata_set,
                unset: modified_metadata_unset,
            },
            // F5: `InlineWikiLinkAutoStubbed` rides on the outcome so the
            // update path matches create's contract.
            warnings,
            relations_declared,
            // Anchor-only: content byte-identical to disk, so the sidecar
            // is the sole delta. Reachable here only past the no-op guard,
            // which already returned when content is unchanged AND no
            // anchors or unsets — so `content_unchanged` here implies
            // anchor work is present. The explicit `!is_empty()` keeps the
            // predicate self-evidently correct without leaning on that
            // invariant.
            anchor_only: content_unchanged
                && (!validated_anchors.is_empty() || !validated_anchor_unsets.is_empty()),
            anchors: validated_anchors,
            anchor_unsets: validated_anchor_unsets,
        }))
    }

    /// Apply a batch of [`UpdateEntityArgs`] **atomically** — all or
    /// nothing. Surfaces `BatchResult` for `memstead batch-update`
    /// consumers.
    ///
    /// The batch validates and prepares every item first (each with
    /// its own optimistic-lock check), then commits the whole set as
    /// **one** commit per mem. If any item fails — validation error,
    /// `HASH_MISMATCH`, entity-not-found, any per-item refusal —
    /// **nothing is committed**: the on-disk mem and the in-memory
    /// store are restored to exactly their pre-call state, and the
    /// result is marked `applied: false` with EVERY failing item
    /// carrying a typed `{code, message, details}` error envelope
    /// (the family's report-all contract — bounded at
    /// [`Self::BATCH_ERROR_REPORT_CAP`] detailed envelopes, with
    /// `errors_suppressed` counting the rest) and every valid item
    /// marked `"not_applied"`, so one repair cycle fixes the file.
    ///
    /// On success the returned `write_id` is the single batch commit
    /// — an honest `memstead_changes_since` cursor / revert handle. Each
    /// item's per-entry note rides into its own provenance record.
    ///
    /// Empty batches return `applied: true` with zero counts and no
    /// commit. A batch where every item is a no-op (content unchanged)
    /// likewise applies with an empty `write_id`.
    ///
    /// **Rehearsal** (`dry_run: true`): the FULL per-item validation
    /// pass runs — identical refusals, identical report-all envelope —
    /// then the batch stops before any write or commit. A legal batch
    /// returns the would-be receipt (`applied: true`, per-entry
    /// actions) with the marker form's empty `write_id`; an illegal
    /// one returns the same refusal a real call would. Nothing is
    /// staged, committed, or stamped.
    ///
    /// Atomicity is per-mem: for the common single-mem batch a
    /// commit-time backend failure rolls the whole batch back. A batch
    /// spanning multiple mems commits each mem in turn; if a later
    /// mem's commit fails, already-committed mems stay committed
    /// (true cross-mem two-phase commit is out of scope) — but the
    /// dominant failure mode, a per-item validation/hash refusal, is
    /// always fully atomic because no commit happens until every item
    /// has passed.
    pub fn batch_update(
        &mut self,
        updates: Vec<(UpdateEntityArgs, Option<String>)>,
        actor: Actor,
        client: Option<&ClientId>,
        dry_run: bool,
    ) -> Result<crate::ops::BatchResult, EngineError> {
        if updates.is_empty() {
            return Ok(crate::ops::BatchResult {
                warnings: Vec::new(),
                orphan_stubs_removed: Vec::new(),
                errors_suppressed: 0,
                applied: true,
                results: Vec::new(),
                succeeded: 0,
                failed: 0,
                write_id: String::new(),
            });
        }

        // Reload-before-operation: refresh every mem this batch
        // touches *before* preparing items, so each item's
        // `expected_hash` check runs against current truth (the batch
        // is the one multi-op-per-process path, so a sibling commit
        // between boot and this call is plausible). Notices stash on
        // the engine for the caller to drain.
        let mut touched_mems: Vec<String> = updates
            .iter()
            .map(|(a, _)| a.id.mem().to_string())
            .collect();
        touched_mems.sort();
        touched_mems.dedup();
        for v in &touched_mems {
            self.reload_if_stale(Some(v));
        }
        // Same acyclic-guard rule as the single-item path: declared
        // relations on an ACYCLIC rel-type (or one in an
        // `acyclic_sets` set) walk the whole subgraph, so the walk
        // must see every mem — deferred ones included (see the
        // batch_relate comment). Declared signals on any involved
        // schema need the full load too (see the single-item path).
        if updates.iter().any(|(a, _)| {
            a.declare_relations.iter().any(|r| {
                self.schemas.get(a.id.mem()).is_some_and(|s| {
                    s.relationship_acyclic(&r.rel_type)
                        || s.acyclic_set_containing(&r.rel_type).is_some()
                }) || self
                    .schemas
                    .get(r.target.mem())
                    .is_some_and(|s| s.types.values().any(|td| !td.signals.is_empty()))
            }) || self
                .schemas
                .get(a.id.mem())
                .is_some_and(|s| s.types.values().any(|td| !td.signals.is_empty()))
        }) {
            self.ensure_mems_loaded(None);
        }

        // Snapshot the in-memory store so a refused batch (or a
        // commit-time backend failure) can roll back any auto-stubs
        // and store pushes that earlier items already applied during
        // preparation. The on-disk side rolls back by discarding each
        // backend's staged-but-uncommitted pending buffer.
        let store_snapshot = self.store.clone();

        // What each item is, in submission order, so the result
        // entries echo the input order. `Prepared` is a real write
        // (its `PreparedUpdate` lives in `prepared`); `Noop` is an
        // applied no-op (content unchanged, no write); `Error` is a
        // refusal (its envelope lives in `errors` by index).
        enum Item {
            Prepared,
            Noop,
            Error,
        }
        let mut items: Vec<(EntityId, Item)> = Vec::with_capacity(updates.len());
        let mut prepared: Vec<PreparedUpdate> = Vec::new();
        let mut notes: Vec<Option<String>> = Vec::new();
        let mut errors: Vec<(usize, EngineError)> = Vec::new();

        // --- Phase 1: validate + prepare every item (no commits).
        // Report-all: a failing item never stops preparation — every
        // remaining item still validates so the refusal can name every
        // failing entry at once (the family's upgraded contract).
        for (i, (args, note)) in updates.into_iter().enumerate() {
            let id = args.id.clone();
            // Rehearsal is batch-level (the `dry_run` parameter) —
            // force the per-entry flag off so `Done` below always
            // means a genuine content no-op, never a per-entry
            // dry-run short-circuit misread as one.
            let mut args = args;
            args.dry_run = false;
            match self.prepare_update(args) {
                Ok(PrepareOutcome::Done(_)) => {
                    // No-op: applied, no write.
                    items.push((id, Item::Noop));
                }
                Ok(PrepareOutcome::Prepared(p)) => {
                    prepared.push(p);
                    notes.push(note);
                    items.push((id, Item::Prepared));
                }
                Err(e) => {
                    items.push((id, Item::Error));
                    errors.push((i, e));
                }
            }
        }

        if !errors.is_empty() {
            // Refuse the whole batch. Roll back store + disk, then
            // report every failing entry — bounded at
            // `BATCH_ERROR_REPORT_CAP` detailed envelopes with
            // `errors_suppressed` counting the rest.
            self.store = store_snapshot;
            self.discard_all_pending();
            let failed = errors.len();
            let mut error_map: std::collections::HashMap<usize, EngineError> =
                errors.into_iter().collect();
            let mut reported = 0usize;
            let mut suppressed = 0usize;
            let results: Vec<crate::ops::BatchEntry> = items
                .into_iter()
                .enumerate()
                .map(|(i, (id, _))| match error_map.remove(&i) {
                    Some(e) => {
                        if reported < Self::BATCH_ERROR_REPORT_CAP {
                            reported += 1;
                            crate::ops::BatchEntry {
                                id,
                                action: "error".to_string(),
                                error: Some(batch_error_envelope(&e)),
                            }
                        } else {
                            suppressed += 1;
                            crate::ops::BatchEntry {
                                id,
                                action: "error".to_string(),
                                error: None,
                            }
                        }
                    }
                    None => crate::ops::BatchEntry {
                        id,
                        action: "not_applied".to_string(),
                        error: None,
                    },
                })
                .collect();
            return Ok(crate::ops::BatchResult {
                warnings: Vec::new(),
                orphan_stubs_removed: Vec::new(),
                errors_suppressed: suppressed,
                applied: false,
                results,
                succeeded: 0,
                failed,
                write_id: String::new(),
            });
        }

        // Rehearsal: every item validated (the pass above is the same
        // one a real batch runs), nothing failed — stop before any
        // write. Roll back the prepare pass's store effects, discard
        // any pending buffers, and return the would-be receipt with
        // the marker form's empty `write_id`.
        if dry_run {
            self.store = store_snapshot;
            self.discard_all_pending();
            let succeeded = items.len();
            let results: Vec<crate::ops::BatchEntry> = items
                .into_iter()
                .map(|(id, item)| crate::ops::BatchEntry {
                    id,
                    action: match item {
                        Item::Prepared => "updated".to_string(),
                        Item::Noop => "noop".to_string(),
                        Item::Error => unreachable!("refusal path returned above"),
                    },
                    error: None,
                })
                .collect();
            return Ok(crate::ops::BatchResult {
                warnings: Vec::new(),
                orphan_stubs_removed: Vec::new(),
                errors_suppressed: 0,
                applied: true,
                results,
                succeeded,
                failed: 0,
                write_id: String::new(),
            });
        }

        // --- Phase 2: stage every prepared write, then commit once
        // per mem. ---
        for p in &prepared {
            if let Err(e) = self.mounts[p.mount_idx]
                .backend
                .write_entity(Path::new(&p.file_path), p.markdown.as_bytes())
            {
                self.store = store_snapshot;
                self.discard_all_pending();
                return Err(e.into());
            }
            // Stage each item's anchors into the same per-mem pending
            // buffer so they ride the batch commit atomically.
            if (!p.anchors.is_empty() || !p.anchor_unsets.is_empty())
                && let Err(e) = super::stage_anchors_sidecar(
                    self.mounts[p.mount_idx].backend.as_ref(),
                    &p.id,
                    &p.anchor_unsets,
                    p.anchors.clone(),
                )
            {
                self.store = store_snapshot;
                self.discard_all_pending();
                return Err(e);
            }
            // Derivation baselines (plan 12) — same predicate and
            // staging as the single update; rides the batch commit.
            if let Some(schema) = self.schemas.get(p.id.mem()) {
                for r in p
                    .relations_declared
                    .iter()
                    .filter(|r| super::rel_type_declares_derivation(schema, &r.rel_type))
                {
                    let hash = self
                        .store
                        .get(&r.target)
                        .map(|e| e.content_hash.clone())
                        .unwrap_or_default();
                    let (from, rel, to) =
                        (p.id.to_string(), r.rel_type.clone(), r.target.to_string());
                    if let Err(e) = super::stage_derivation_sidecar(
                        self.mounts[p.mount_idx].backend.as_ref(),
                        |s| s.set(&from, &rel, &to, &hash),
                    ) {
                        self.store = store_snapshot;
                        self.discard_all_pending();
                        return Err(e);
                    }
                }
            }
        }

        // Distinct mount indices in first-seen order — one commit each.
        let mut distinct_mounts: Vec<usize> = Vec::new();
        for p in &prepared {
            if !distinct_mounts.contains(&p.mount_idx) {
                distinct_mounts.push(p.mount_idx);
            }
        }
        let mut mount_commits: Vec<(usize, String)> = Vec::with_capacity(distinct_mounts.len());
        for &m in &distinct_mounts {
            let entity_ids: Vec<String> = prepared
                .iter()
                .filter(|p| p.mount_idx == m)
                .map(|p| p.id.to_string())
                .collect();
            let count = entity_ids.len();
            let subject = format!("memstead: batch-update ({count} entities)");
            // Per-entry notes ride the batch commit's note record as
            // `<id>: <note>` lines (decision 3) — `append_provenance` is a
            // no-op on the git-branch backend, so this record is where
            // they survive. No notes → no note record.
            let note_lines: Vec<String> = prepared
                .iter()
                .zip(notes.iter())
                .filter(|(p, _)| p.mount_idx == m)
                .filter_map(|(p, n)| n.as_ref().map(|n| format!("{}: {n}", p.id)))
                .collect();
            let ctx = CommitContext {
                actor,
                client: client.cloned(),
                tool: Some("batch_update"),
                note: if note_lines.is_empty() {
                    None
                } else {
                    Some(note_lines.join("\n"))
                },
                role: self.current_role,
                logical_operation_id: None,
                // F13: name every entity this batch commit touched so an
                // `--include-notes` reader can recover them from the note
                // record alone — the subject only says `(N entities)`.
                entity_ids: Some(entity_ids),
            };
            match self.mounts[m].backend.commit(&subject, &ctx) {
                Ok(sha) => mount_commits.push((m, sha)),
                Err(e) => {
                    // A commit failed. Roll back the store and any
                    // still-pending backends. Mems already committed
                    // in this loop stay committed (per-mem atomicity).
                    self.store = store_snapshot;
                    self.discard_all_pending();
                    return Err(e.into());
                }
            }
        }

        // Provenance + store application per item, now that the commits
        // landed. `record_self_write` marks the commit as engine-self
        // so drift detection ignores it.
        let mut batch_warnings: Vec<WarningHint> = Vec::new();
        for (p, note) in prepared.iter().zip(notes.iter()) {
            let write_id = mount_commits
                .iter()
                .find(|(m, _)| *m == p.mount_idx)
                .map(|(_, s)| s.clone())
                .unwrap_or_default();
            self.mounts[p.mount_idx].backend.append_provenance(
                &Provenance::new(
                    std::time::SystemTime::now(),
                    ProvenanceKind::Update,
                    Some(p.id.to_string()),
                    actor,
                    client.cloned(),
                    note.clone(),
                )
                .with_role(self.current_role),
            )?;
            self.record_self_write(p.mount_idx, &write_id);
            batch_warnings.extend(self.stamp_mutation_versions(p.mount_idx));
            self.apply_prepared_to_store(p)?;
        }

        self.invalidate_communities();
        self.invalidate_search_indexes();

        // Single-mem batches name their one commit; multi-mem names
        // the last mem committed (see the method docstring).
        let write_id = mount_commits
            .last()
            .map(|(_, s)| s.clone())
            .unwrap_or_default();
        let succeeded = items.len();
        let results: Vec<crate::ops::BatchEntry> = items
            .into_iter()
            .map(|(id, item)| crate::ops::BatchEntry {
                id,
                action: match item {
                    Item::Prepared => "updated".to_string(),
                    Item::Noop => "noop".to_string(),
                    Item::Error => unreachable!("refusal path returned above"),
                },
                error: None,
            })
            .collect();

        Ok(crate::ops::BatchResult {
            warnings: batch_warnings,
            orphan_stubs_removed: Vec::new(),
            errors_suppressed: 0,
            applied: true,
            results,
            succeeded,
            failed: 0,
            write_id,
        })
    }

    /// Best-effort discard of every backend's staged-but-uncommitted
    /// pending buffer — the disk-side half of an atomic-batch rollback.
    /// Discard errors (a poisoned pending mutex) are swallowed: we are
    /// already unwinding a refused batch and have nothing better to do.
    pub(super) fn discard_all_pending(&self) {
        for mount in &self.mounts {
            let _ = mount.backend.discard_pending();
        }
    }

    /// CommitContext-bundling wrapper around [`Self::update_entity`].
    /// See [`Self::create_entity_with_ctx`] for the rationale.
    pub fn update_entity_with_ctx(
        &mut self,
        args: UpdateEntityArgs,
        ctx: &CommitContext<'_>,
    ) -> Result<UpdateEntityOutcome, EngineError> {
        self.update_entity(args, ctx.actor, ctx.client.as_ref(), ctx.note.as_deref())
    }
}

/// Build a per-item structured error envelope for [`Engine::batch_update`].
/// Mirrors the `{code, message, details}` shape single-update failures
/// carry on the MCP wire so a mixed-success batch is structurally uniform.
/// Variants without a typed recovery payload (boundary / internal failures
/// like `ParseAfterWrite`, `Backend`) return an empty details object —
/// the code and message channels still discriminate.
pub(super) fn batch_error_envelope(err: &EngineError) -> crate::ops::BatchError {
    // The per-item envelope reads the centralised
    // `EngineError::details()` helper so every typed variant ships the
    // same recovery payload the singleton MCP/CLI surfaces emit, so
    // agents' "fix from `details` rather than re-fetching" loop works
    // the same in batch mode.
    let code = err.code().to_string();
    let message = err.to_string();
    let details = err.details();
    crate::ops::BatchError {
        code,
        message,
        details,
    }
}

/// Validate, auto-stub, and append a batch of relation declarations
/// onto `next.relationships`. Returns the canonical
/// `RelationDeclared` summary echoed back on the outcome.
///
/// Validates each declared relation against the same gates
/// `memstead_relate` runs (target-id grammar, rel-type vocabulary,
/// schema shape, cross-mem policy, ReadOnly-target rule). On the
/// add path with an absent Write-mem target, the target is
/// auto-stubbed via `make_stub` — matching the
/// `WarningHint::AutoStubCreated` semantics of the relate flow.
///
/// Defined at module scope (rather than a method on `Engine`) so
/// the borrow on `engine.store` for the auto-stub upsert can run
/// alongside the `&mut next` borrow.
fn apply_declare_relations(
    engine: &mut Engine,
    next: &mut Entity,
    declarations: &[crate::ops::RelateArg],
    source_mem: &str,
    source_mount_idx: usize,
    type_def: &memstead_schema::TypeDefinition,
    schema: &memstead_schema::Schema,
) -> Result<Vec<RelationDeclared>, EngineError> {
    let _ = type_def; // Reserved for future per-type policy hooks.
    let _ = source_mount_idx; // Reserved for parity with delete.
    let mut declared: Vec<RelationDeclared> = Vec::with_capacity(declarations.len());
    for rel in declarations {
        // Canonicalise rel_type to UPPER_SNAKE_CASE so the validator
        // and the stored edge see the same wire-contract form.
        let canonical = crate::entity::id::validate_rel_type(&rel.rel_type)
            .unwrap_or_else(|_| rel.rel_type.clone());

        validate_relation_target_grammar(&rel.target)?;

        let target_mem = rel.target.mem().to_string();
        // Grant + ReadOnly-missing-target checks both live in the
        // shared add-path funnel.
        super::validate_cross_mem_add_policy(engine, source_mem, &rel.target)?;

        // Rel-type + shape validation, routed through the engine's
        // cross-mem-aware edge validator. Cross-different-schema
        // edges check vocabulary + shape against the source schema's
        // `cross_mem_relationships:` entry; same-schema edges fall
        // through to the intra-mem `relationships.definitions`.
        // Open-mode admits unknown rel-types silently (no
        // per-declaration warning surfaced here — symmetry with the
        // pre-cross-mem behaviour).
        let target_type = engine
            .store
            .get(&rel.target)
            .map(|e| e.entity_type.clone())
            .filter(|t| !t.is_empty());
        // Deferred-mem target (flywheel W7/02): the real type comes
        // from the one resolved blob, never from loading the mem.
        let target_type = match target_type {
            Some(t) => Some(t),
            None => super::peek_deferred_target_type(engine, &rel.target)?,
        };
        let _ = super::route_edge_validation(
            engine,
            &canonical,
            next.entity_type.as_str(),
            target_type.as_deref(),
            source_mem,
            &target_mem,
            &next.id,
            &rel.target,
            /* check_shape = */ true,
        )?;

        // Per-edge description posture. Normalise first so
        // empty/whitespace-only inputs collapse to `None` and the
        // posture check sees a canonical input that matches what
        // the renderer will emit.
        let normalised_description =
            crate::entity::normalise_description(rel.description.as_deref());
        super::validate_description_posture(
            engine,
            &canonical,
            normalised_description.as_deref(),
            source_mem,
            &target_mem,
            &next.id,
            &rel.target,
        )?;
        // declare_relations is an explicit-author
        // boundary too — gate on manual_authoring posture.
        super::validate_manual_authoring_posture(
            engine,
            &canonical,
            source_mem,
            &next.id,
            &rel.target,
        )?;

        // Cycle family — the same shared gate `memstead_relate` runs
        // (self-loop on listed no-self-loop rel-types, long cycle on acyclic
        // types), against the current store state.
        super::validate_edge_acyclicity(
            &engine.store,
            schema,
            &next.id,
            next.entity_type.as_str(),
            &rel.target,
            &canonical,
        )?;

        // Append to the entity's relationships list. Duplicate
        // declarations are idempotent — same (rel_type, target) pair
        // is a silent no-op so the agent can re-issue the same call
        // without surprise.
        let exists = next
            .relationships
            .iter()
            .any(|r| r.rel_type == canonical && r.target == rel.target);
        if !exists {
            next.relationships.push(Relationship {
                rel_type: canonical.clone(),
                target: rel.target.clone(),
                description: normalised_description,
            });
        }

        // Auto-stub absent Write-mem targets. Same mechanic as
        // `memstead_relate`'s relate path. ReadOnly cross-mem targets
        // were caught above; same-mem and cross-mem-to-Write
        // both fall through here.
        let target_was_stubbed = !engine.store.contains(&rel.target);
        if target_was_stubbed && !exists {
            let kind = super::deferred_verified_stub_kind(engine, &rel.target)?;
            engine
                .store
                .upsert(rel.target.clone(), make_stub(&rel.target, kind));
        }

        declared.push(RelationDeclared {
            rel_type: canonical,
            target: rel.target.clone(),
            target_was_stubbed,
        });
    }
    Ok(declared)
}

#[cfg(test)]
mod tests {

    use indexmap::IndexMap;
    use tempfile::TempDir;

    use crate::backend::MemBackend;
    use crate::engine::test_helpers::*;
    use crate::engine::{
        CreateEntityArgs, Engine, EngineError, RelateEntityArgs, UpdateEntityArgs,
    };
    use crate::entity::EntityId;

    use crate::storage::{ArchiveBackend, FilesystemMemWriter};
    use crate::vcs::Actor;

    /// A mutation writing a section whose declared heading differs
    /// from a heading the file already carried for the same key warns
    /// (`SECTION_HEADING_DIVERGENCE`, naming both headings) and still
    /// commits. Refusal complement: once the file carries the matching
    /// heading, the same update emits no such warning.
    #[test]
    fn update_warns_on_section_heading_divergence_and_still_commits() {
        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        // Pre-existing file whose heading derives to `identity` but is
        // not the schema's declared `Identity`.
        std::fs::write(
            mem_dir.join("diverged.md"),
            "---\ntype: spec\n---\n# Diverged\n\n## IDENTITY\n\nold text.\n",
        )
        .unwrap();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        let (actor, client) = cli_actor();
        let id = EntityId::new("specs", "diverged");

        let update_identity = |engine: &mut Engine, body: &str| {
            let current = engine.get_entity(&id).unwrap().content_hash.clone();
            let mut sections = IndexMap::new();
            sections.insert("identity".to_string(), body.to_string());
            engine
                .update_entity(
                    UpdateEntityArgs {
                        anchors: Vec::new(),
                        id: id.clone(),
                        expected_hash: Some(current),
                        sections,
                        append_sections: IndexMap::new(),
                        patch_sections: IndexMap::new(),
                        metadata: IndexMap::new(),
                        metadata_unset: Vec::new(),
                        declare_relations: Vec::new(),
                        dry_run: false,
                        relations_unset: Vec::new(),
                        anchors_unset: Vec::new(),
                    },
                    actor,
                    Some(&client),
                    None,
                )
                .unwrap()
        };

        let outcome = update_identity(&mut engine, "new text.");
        assert!(!outcome.write_id.is_empty(), "the mutation still commits");
        let divergences: Vec<_> = outcome
            .warnings
            .iter()
            .filter_map(|w| match w {
                crate::ops::WarningHint::SectionHeadingDivergence {
                    section_key,
                    writing_heading,
                    existing_heading,
                    ..
                } => Some((
                    section_key.clone(),
                    writing_heading.clone(),
                    existing_heading.clone(),
                )),
                _ => None,
            })
            .collect();
        assert_eq!(
            divergences,
            vec![(
                "identity".to_string(),
                "Identity".to_string(),
                "IDENTITY".to_string()
            )],
            "warning names both headings; all warnings = {:?}",
            outcome.warnings
        );

        // The regenerated file now carries the declared heading — a
        // second update to the same section must not warn.
        let outcome2 = update_identity(&mut engine, "third text.");
        assert!(
            !outcome2
                .warnings
                .iter()
                .any(|w| matches!(w, crate::ops::WarningHint::SectionHeadingDivergence { .. })),
            "matching heading emits no divergence warning: {:?}",
            outcome2.warnings
        );
    }

    #[test]
    fn batch_update_empty_batch_returns_zero_counts() {
        // No updates → BatchResult with zero counts + empty
        // write_id. No engine mutation happens.
        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();

        let result = engine
            .batch_update(Vec::new(), Actor::Cli, None, false)
            .unwrap();
        assert!(result.applied, "empty batch is a vacuous success");
        assert_eq!(result.results.len(), 0);
        assert_eq!(result.succeeded, 0);
        assert_eq!(result.failed, 0);
        assert_eq!(result.write_id, "");
    }

    #[test]
    fn batch_update_refuses_whole_batch_when_one_item_fails() {
        // Atomic semantics: a 2-item batch where item 1 is valid and
        // item 2 targets a missing id refuses the WHOLE batch. Nothing
        // is committed — item 1 is NOT applied (its section change does
        // not land), `applied` is false, `write_id` is empty, the
        // missing item carries the typed ENTITY_NOT_FOUND envelope, and
        // the valid item is marked `not_applied`.
        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();

        // Seed: create an entity.
        let create_args = CreateEntityArgs {
            anchors: Vec::new(),
            mem: "specs".to_string(),
            title: "Seed".to_string(),
            entity_type: "spec".to_string(),
            sections: IndexMap::from_iter([
                ("identity".to_string(), "seed identity".to_string()),
                ("purpose".to_string(), "seed purpose".to_string()),
            ]),
            metadata: IndexMap::new(),
            relations: Vec::new(),
            dry_run: false,
        };
        let created = engine
            .create_entity(create_args, Actor::Cli, None, None)
            .unwrap();

        // Batch: update the seed entity AND a missing id.
        let valid_update = UpdateEntityArgs {
            anchors: Vec::new(),
            id: created.id.clone(),
            expected_hash: Some(created.content_hash.clone()),
            sections: IndexMap::from_iter([("identity".to_string(), "updated body".to_string())]),
            append_sections: IndexMap::new(),
            patch_sections: IndexMap::new(),
            metadata: IndexMap::new(),
            metadata_unset: Vec::new(),
            declare_relations: Vec::new(),
            dry_run: false,
            relations_unset: Vec::new(),
            anchors_unset: Vec::new(),
        };
        let missing_update = UpdateEntityArgs {
            anchors: Vec::new(),
            id: EntityId("specs--nonexistent".to_string()),
            expected_hash: None,
            sections: IndexMap::new(),
            append_sections: IndexMap::new(),
            patch_sections: IndexMap::new(),
            metadata: IndexMap::new(),
            metadata_unset: Vec::new(),
            declare_relations: Vec::new(),
            dry_run: false,
            relations_unset: Vec::new(),
            anchors_unset: Vec::new(),
        };

        let result = engine
            .batch_update(
                vec![(valid_update, None), (missing_update, None)],
                Actor::Cli,
                None,
                false,
            )
            .unwrap();
        // Whole batch refused: nothing applied, no commit.
        assert!(!result.applied, "a failing item must refuse the batch");
        assert_eq!(result.results.len(), 2);
        assert_eq!(result.succeeded, 0);
        assert_eq!(result.failed, 1);
        assert_eq!(result.write_id, "", "refused batch must not commit");
        // First entry: the valid item, marked not_applied (the batch
        // was refused before it could land).
        assert_eq!(result.results[0].action, "not_applied");
        assert!(result.results[0].error.is_none());
        // Second entry: the failing item carries the typed envelope.
        assert_eq!(result.results[1].action, "error");
        let err = result.results[1]
            .error
            .as_ref()
            .expect("failed entry must carry a structured error envelope");
        assert_eq!(err.code, "ENTITY_NOT_FOUND");
        assert!(err.message.contains("not found"), "got: {}", err.message);

        // The valid item's section change must NOT have landed — the
        // store is byte-identical to pre-call.
        let seed = engine.get_entity(&created.id).unwrap();
        assert_eq!(
            seed.sections.get("identity").map(String::as_str),
            Some("seed identity"),
            "refused batch must leave the in-memory store untouched",
        );
        assert_eq!(
            seed.content_hash, created.content_hash,
            "refused batch must not change the entity's content hash",
        );
    }

    #[test]
    fn batch_update_applies_all_valid_items_as_one_commit() {
        // A 2-item batch where both items are valid
        // applies both and produces exactly one commit; the response's
        // write_id names it and both entries report "updated".
        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();

        let mk = |title: &str| CreateEntityArgs {
            anchors: Vec::new(),
            mem: "specs".to_string(),
            title: title.to_string(),
            entity_type: "spec".to_string(),
            sections: IndexMap::from_iter([
                ("identity".to_string(), "id".to_string()),
                ("purpose".to_string(), "purp".to_string()),
            ]),
            metadata: IndexMap::new(),
            relations: Vec::new(),
            dry_run: false,
        };
        let a = engine
            .create_entity(mk("A"), Actor::Cli, None, None)
            .unwrap();
        let b = engine
            .create_entity(mk("B"), Actor::Cli, None, None)
            .unwrap();

        let upd = |id: EntityId, hash: String, body: &str| UpdateEntityArgs {
            anchors: Vec::new(),
            id,
            expected_hash: Some(hash),
            sections: IndexMap::from_iter([("identity".to_string(), body.to_string())]),
            append_sections: IndexMap::new(),
            patch_sections: IndexMap::new(),
            metadata: IndexMap::new(),
            metadata_unset: Vec::new(),
            declare_relations: Vec::new(),
            dry_run: false,
            relations_unset: Vec::new(),
            anchors_unset: Vec::new(),
        };

        let result = engine
            .batch_update(
                vec![
                    (upd(a.id.clone(), a.content_hash.clone(), "A body"), None),
                    (upd(b.id.clone(), b.content_hash.clone(), "B body"), None),
                ],
                Actor::Cli,
                None,
                false,
            )
            .unwrap();
        assert!(result.applied);
        assert_eq!(result.succeeded, 2);
        assert_eq!(result.failed, 0);
        assert!(
            !result.write_id.is_empty(),
            "applied batch carries the commit"
        );
        assert!(result.results.iter().all(|e| e.action == "updated"));
        // Both section changes landed.
        assert_eq!(
            engine
                .get_entity(&a.id)
                .unwrap()
                .sections
                .get("identity")
                .map(String::as_str),
            Some("A body"),
        );
        assert_eq!(
            engine
                .get_entity(&b.id)
                .unwrap()
                .sections
                .get("identity")
                .map(String::as_str),
            Some("B body"),
        );
    }

    /// Rehearsal contract (agent-trust plan 07): `batch_update` with
    /// `dry_run: true` runs the full per-item validation, reports the
    /// would-be receipt with the marker form's empty `write_id`, and
    /// persists NOTHING — on-disk bodies and hashes stay untouched.
    /// The follow-up real call with the SAME expected hashes succeeds,
    /// proving both the identical-validation contract and the
    /// side-effect-freeness (a persisted rehearsal would have moved
    /// the hashes and refused the real call).
    #[test]
    fn batch_update_dry_run_reports_receipt_and_writes_nothing() {
        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();

        let mk = |title: &str| CreateEntityArgs {
            anchors: Vec::new(),
            mem: "specs".to_string(),
            title: title.to_string(),
            entity_type: "spec".to_string(),
            sections: IndexMap::from_iter([
                ("identity".to_string(), "id".to_string()),
                ("purpose".to_string(), "purp".to_string()),
            ]),
            metadata: IndexMap::new(),
            relations: Vec::new(),
            dry_run: false,
        };
        let a = engine
            .create_entity(mk("A"), Actor::Cli, None, None)
            .unwrap();
        let b = engine
            .create_entity(mk("B"), Actor::Cli, None, None)
            .unwrap();

        let upd = |id: EntityId, hash: String, body: &str| UpdateEntityArgs {
            anchors: Vec::new(),
            id,
            expected_hash: Some(hash),
            sections: IndexMap::from_iter([("identity".to_string(), body.to_string())]),
            append_sections: IndexMap::new(),
            patch_sections: IndexMap::new(),
            metadata: IndexMap::new(),
            metadata_unset: Vec::new(),
            declare_relations: Vec::new(),
            dry_run: false,
            relations_unset: Vec::new(),
            anchors_unset: Vec::new(),
        };
        let batch = || {
            vec![
                (upd(a.id.clone(), a.content_hash.clone(), "A body"), None),
                (upd(b.id.clone(), b.content_hash.clone(), "B body"), None),
            ]
        };

        let rehearsed = engine
            .batch_update(batch(), Actor::Cli, None, true)
            .unwrap();
        assert!(rehearsed.applied, "{rehearsed:?}");
        assert_eq!(rehearsed.succeeded, 2);
        assert!(rehearsed.write_id.is_empty(), "marker form: empty write_id");
        assert!(rehearsed.results.iter().all(|e| e.action == "updated"));
        // Nothing persisted: body and hash unchanged.
        let a_now = engine.get_entity(&a.id).unwrap();
        assert_eq!(
            a_now.sections.get("identity").map(String::as_str),
            Some("id")
        );
        assert_eq!(a_now.content_hash, a.content_hash);

        // The real call with the same (pre-rehearsal) hashes lands.
        let real = engine
            .batch_update(batch(), Actor::Cli, None, false)
            .unwrap();
        assert!(real.applied, "{real:?}");
        assert!(!real.write_id.is_empty());
        assert_eq!(
            engine
                .get_entity(&a.id)
                .unwrap()
                .sections
                .get("identity")
                .map(String::as_str),
            Some("A body"),
        );
    }

    /// Rehearsal refusal parity: a failing batch refuses under
    /// `dry_run: true` with the SAME per-entry envelope (code,
    /// message, details) the real refusal carries.
    #[test]
    fn batch_update_dry_run_refuses_identically_to_real() {
        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        let created = engine
            .create_entity(
                CreateEntityArgs {
                    anchors: Vec::new(),
                    mem: "specs".to_string(),
                    title: "Valid".to_string(),
                    entity_type: "spec".to_string(),
                    sections: IndexMap::from_iter([
                        ("identity".to_string(), "x".to_string()),
                        ("purpose".to_string(), "p".to_string()),
                    ]),
                    metadata: IndexMap::new(),
                    relations: Vec::new(),
                    dry_run: false,
                },
                Actor::Cli,
                None,
                None,
            )
            .unwrap();
        let upd = |id: EntityId, hash: Option<String>| UpdateEntityArgs {
            anchors: Vec::new(),
            id,
            expected_hash: hash,
            sections: IndexMap::from_iter([("identity".to_string(), "new".to_string())]),
            append_sections: IndexMap::new(),
            patch_sections: IndexMap::new(),
            metadata: IndexMap::new(),
            metadata_unset: Vec::new(),
            declare_relations: Vec::new(),
            dry_run: false,
            relations_unset: Vec::new(),
            anchors_unset: Vec::new(),
        };
        let batch = || {
            vec![
                (
                    upd(created.id.clone(), Some("wrong-hash".to_string())),
                    None,
                ),
                (upd(EntityId("specs--missing".to_string()), None), None),
            ]
        };

        let rehearsed = engine
            .batch_update(batch(), Actor::Cli, None, true)
            .unwrap();
        let real = engine
            .batch_update(batch(), Actor::Cli, None, false)
            .unwrap();
        assert!(!rehearsed.applied && !real.applied);
        let envelope = |r: &crate::ops::BatchResult| {
            r.results
                .iter()
                .map(|e| {
                    (
                        e.id.to_string(),
                        e.action.clone(),
                        e.error.as_ref().map(|err| {
                            (err.code.clone(), err.message.clone(), err.details.clone())
                        }),
                    )
                })
                .collect::<Vec<_>>()
        };
        assert_eq!(envelope(&rehearsed), envelope(&real), "identical refusals");
        // Neither run persisted anything.
        assert_eq!(
            engine
                .get_entity(&created.id)
                .unwrap()
                .sections
                .get("identity")
                .map(String::as_str),
            Some("x"),
        );
    }

    /// Report-all (the family's upgraded contract): a batch with TWO
    /// failing items names both with their typed codes — a failing
    /// item no longer stops preparation at the first error.
    #[test]
    fn batch_update_reports_every_failing_item() {
        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        let created = engine
            .create_entity(
                CreateEntityArgs {
                    anchors: Vec::new(),
                    mem: "specs".to_string(),
                    title: "Seed".to_string(),
                    entity_type: "spec".to_string(),
                    sections: IndexMap::from_iter([
                        ("identity".to_string(), "seed identity".to_string()),
                        ("purpose".to_string(), "seed purpose".to_string()),
                    ]),
                    metadata: IndexMap::new(),
                    relations: Vec::new(),
                    dry_run: false,
                },
                Actor::Cli,
                None,
                None,
            )
            .unwrap();

        let upd = |id: EntityId, hash: Option<String>| UpdateEntityArgs {
            anchors: Vec::new(),
            id,
            expected_hash: hash,
            sections: IndexMap::from_iter([("identity".to_string(), "new body".to_string())]),
            append_sections: IndexMap::new(),
            patch_sections: IndexMap::new(),
            metadata: IndexMap::new(),
            metadata_unset: Vec::new(),
            declare_relations: Vec::new(),
            dry_run: false,
            relations_unset: Vec::new(),
            anchors_unset: Vec::new(),
        };
        let result = engine
            .batch_update(
                vec![
                    (upd(created.id.clone(), None), None),
                    (upd(EntityId("specs--missing-one".to_string()), None), None),
                    (upd(EntityId("specs--missing-two".to_string()), None), None),
                ],
                Actor::Cli,
                None,
                false,
            )
            .unwrap();
        assert!(!result.applied);
        assert_eq!(result.failed, 2, "{result:?}");
        assert_eq!(result.write_id, "");
        let codes: Vec<(usize, &str)> = result
            .results
            .iter()
            .enumerate()
            .filter(|(_, r)| r.action == "error")
            .map(|(i, r)| (i, r.error.as_ref().map(|e| e.code.as_str()).unwrap_or("")))
            .collect();
        assert_eq!(
            codes,
            vec![(1, "ENTITY_NOT_FOUND"), (2, "ENTITY_NOT_FOUND")],
            "BOTH failing items named, not just the first: {result:?}"
        );
        assert_eq!(result.results[0].action, "not_applied");
        // The valid item's change did not land.
        assert_eq!(
            engine
                .get_entity(&created.id)
                .unwrap()
                .sections
                .get("identity")
                .map(String::as_str),
            Some("seed identity"),
        );
    }

    #[test]
    fn batch_update_rolls_back_in_memory_store_auto_stub_on_refusal() {
        // The subtle invariant: an earlier item that auto-stubs a
        // relation target during preparation must have that stub rolled
        // OUT of the in-memory store when a later item refuses the
        // batch. Item 1 declares a relation to an absent target (which
        // upserts a forward-reference stub during prepare); item 2
        // targets a missing entity and fails. The refusal must leave no
        // trace of the stub.
        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir.clone()),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        engine.set_workspace_root(mem_dir);
        let (actor, client) = cli_actor();

        let a = engine
            .create_entity(
                empty_create_args("specs", "Anchor"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        let stub_target = EntityId::new("specs", "would-be-stub");
        let item1 = UpdateEntityArgs {
            anchors: Vec::new(),
            relations_unset: Vec::new(),
            anchors_unset: Vec::new(),
            id: a.id.clone(),
            expected_hash: Some(a.content_hash.clone()),
            sections: IndexMap::new(),
            append_sections: IndexMap::new(),
            patch_sections: IndexMap::new(),
            metadata: IndexMap::new(),
            metadata_unset: Vec::new(),
            declare_relations: vec![crate::ops::RelateArg {
                rel_type: "USES".to_string(),
                target: stub_target.clone(),
                description: None,
            }],
            dry_run: false,
        };
        let item2 = UpdateEntityArgs {
            anchors: Vec::new(),
            id: EntityId::new("specs", "nonexistent"),
            expected_hash: None,
            sections: IndexMap::from_iter([("identity".to_string(), "x".to_string())]),
            append_sections: IndexMap::new(),
            patch_sections: IndexMap::new(),
            metadata: IndexMap::new(),
            metadata_unset: Vec::new(),
            declare_relations: Vec::new(),
            dry_run: false,
            relations_unset: Vec::new(),
            anchors_unset: Vec::new(),
        };

        // Sanity: the would-be stub does not exist before the batch.
        assert!(engine.get_entity(&stub_target).is_none());

        let result = engine
            .batch_update(
                vec![(item1, None), (item2, None)],
                actor,
                Some(&client),
                false,
            )
            .unwrap();
        assert!(!result.applied, "missing item 2 must refuse the batch");

        // The auto-stub item 1 created during preparation was rolled
        // back with the store snapshot — no orphaned stub survives.
        assert!(
            engine.get_entity(&stub_target).is_none(),
            "refused batch must roll the in-memory auto-stub back out of the store",
        );
        // The anchor's relation set is unchanged too.
        let anchor = engine.get_entity(&a.id).unwrap();
        assert!(
            !anchor.relationships.iter().any(|r| r.target == stub_target),
            "refused batch must not leave the declared relation on the anchor",
        );
    }

    #[test]
    fn update_entity_replaces_a_section_and_logs_provenance() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Updatable");
        let (actor, client) = cli_actor();

        let mut sections = IndexMap::new();
        sections.insert("identity".to_string(), "Updated body.".to_string());

        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                Some("section update"),
            )
            .unwrap();

        assert_eq!(
            outcome.modified_sections.replaced,
            vec!["identity".to_string()]
        );
        assert_ne!(
            outcome.content_hash, seeded.content_hash,
            "hash must change"
        );
        // Store carries the new content.
        let entity = engine.get_entity(&seeded.id).unwrap();
        assert!(
            entity
                .sections
                .get("identity")
                .unwrap()
                .contains("Updated body.")
        );
        // Provenance log records the update.
        let log = std::fs::read_to_string(tmp.path().join(".memstead/changes.jsonl")).unwrap();
        assert!(log.contains("\"kind\":\"update\""));
        assert!(log.contains("\"note\":\"section update\""));
    }

    #[test]
    fn update_entity_rejects_hash_mismatch() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Hash Guarded");
        let (actor, client) = cli_actor();
        let err = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some("wrong-hash".to_string()),
                    sections: IndexMap::new(),
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap_err();
        match err {
            EngineError::HashMismatch {
                id,
                current,
                is_stub,
            } => {
                assert_eq!(id, seeded.id.to_string());
                assert_eq!(current, seeded.content_hash);
                assert!(!is_stub, "real entity must not flag as stub");
            }
            other => panic!("expected HashMismatch, got {other:?}"),
        }
    }

    #[test]
    fn update_entity_rejects_unknown_id() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, _) = engine_with_seed(&tmp, "Anchor");
        let (actor, client) = cli_actor();
        let err = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: crate::EntityId::new("specs", "ghost"),
                    expected_hash: None,
                    sections: IndexMap::new(),
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap_err();
        assert!(matches!(err, EngineError::NotFound { .. }));
    }

    #[test]
    fn update_entity_rejects_read_only_mount() {
        let tmp = TempDir::new().unwrap();
        let archive_path = build_archive(
            tmp.path(),
            "ext",
            &[(
                "a.md",
                b"---\ntype: spec\n---\n# A\n\n## Identity\n\nbody.\n",
            )],
        );
        let mut engine = Engine::from_mounts(vec![(
            archive_mount("external", archive_path.clone()),
            Box::new(ArchiveBackend::new(archive_path)),
        )])
        .unwrap();
        let (actor, client) = cli_actor();
        let id = crate::EntityId::new("external", "a");
        let err = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id,
                    expected_hash: None,
                    sections: IndexMap::new(),
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap_err();
        assert!(matches!(err, EngineError::ReadOnlyMount(v) if v == "external"));
    }

    #[test]
    fn update_entity_patches_section_with_find_and_replace() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Patch Subject");
        let (actor, client) = cli_actor();

        // Pre-write a known body via the replace path so the patch
        // test has a deterministic substring to target.
        let mut replace = IndexMap::new();
        replace.insert("identity".to_string(), "hello world hello".to_string());
        let replaced = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections: replace,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        // First-occurrence patch (all = false).
        let mut patches = IndexMap::new();
        patches.insert(
            "identity".to_string(),
            crate::ops::PatchArg {
                old: "hello".to_string(),
                new: "HI".to_string(),
                all: false,
            },
        );
        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(replaced.content_hash.clone()),
                    sections: IndexMap::new(),
                    append_sections: IndexMap::new(),
                    patch_sections: patches,
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        assert_eq!(outcome.modified_sections.patched, vec!["identity"]);
        let body = engine
            .get_entity(&seeded.id)
            .unwrap()
            .sections
            .get("identity")
            .unwrap()
            .clone();
        assert!(body.contains("HI world hello"), "first-only: {body:?}");
    }

    #[test]
    fn update_entity_patch_rejects_missing_old_substring() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Patch Miss");
        let (actor, client) = cli_actor();
        let mut patches = IndexMap::new();
        patches.insert(
            "identity".to_string(),
            crate::ops::PatchArg {
                old: "this-substring-does-not-exist".to_string(),
                new: "nope".to_string(),
                all: false,
            },
        );
        let err = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections: IndexMap::new(),
                    append_sections: IndexMap::new(),
                    patch_sections: patches,
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap_err();
        match err {
            EngineError::PatchOldNotFound { section, .. } => {
                assert_eq!(section, "identity");
            }
            other => panic!("expected PatchOldNotFound, got {other:?}"),
        }
    }

    #[test]
    fn update_entity_appends_to_existing_section_with_newline_separator() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Append Subject");
        let (actor, client) = cli_actor();

        let mut appends = IndexMap::new();
        appends.insert("identity".to_string(), "appended tail.".to_string());

        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections: IndexMap::new(),
                    append_sections: appends,
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        // modified_sections.appended carries the append key;
        // modified_sections.replaced stays empty.
        assert_eq!(outcome.modified_sections.appended, vec!["identity"]);
        assert!(outcome.modified_sections.replaced.is_empty());

        // The section body now contains the appended tail.
        let updated = engine.get_entity(&seeded.id).unwrap();
        let body = updated.sections.get("identity").expect("identity section");
        assert!(
            body.contains("appended tail."),
            "appended body missing: {body:?}"
        );
    }

    /// Criteria 5 and 6. The state is reproduced the way it actually arrives:
    /// a hand-edited file on disk, reloaded. The engine cannot author it, so a
    /// fixture that went through `update_entity` would prove nothing.
    fn engine_with_open_fence(tmp: &TempDir) -> (Engine, crate::EntityId) {
        let (_engine, seeded) = engine_with_seed(tmp, "Fenced");
        let id = seeded.id.clone();
        let path = tmp.path().join(&seeded.file_path);
        let raw = std::fs::read_to_string(&path).expect("seeded file");
        // Open a fence inside `identity`. In CommonMark its range runs to end
        // of text, so `## Purpose` below is masked and absorbed into it.
        let doctored = raw.replace("fixture identity body", "intro\n\n```rust\nfn main() {}");
        assert_ne!(doctored, raw, "the seeded body must be there to doctor");
        std::fs::write(&path, doctored).unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        drop(seeded);
        (engine, id)
    }

    #[test]
    fn a_write_that_does_not_resolve_an_open_fence_is_refused() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, id) = engine_with_open_fence(&tmp);
        let (actor, client) = cli_actor();
        // The absorption is real before the write is attempted, and this is
        // the exact shape criterion 4 names: `purpose` is present and EMPTY
        // while its content sits verbatim inside `identity`. A surface that
        // reports "empty section" here is telling the truth about the parse
        // and a lie about the entity.
        let stored = engine.get_entity(&id).expect("entity loads");
        assert!(
            stored
                .sections
                .get("purpose")
                .is_some_and(|v| v.trim().is_empty()),
            "purpose should read as empty: {:?}",
            stored.sections.get("purpose")
        );
        assert!(
            stored.sections["identity"].contains("## Purpose"),
            "its content is inside identity: {:?}",
            stored.sections.get("identity")
        );
        let hash = stored.content_hash.clone();

        let err = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: id.clone(),
                    expected_hash: Some(hash),
                    sections: IndexMap::from_iter([(
                        "purpose".to_string(),
                        "a new purpose".to_string(),
                    )]),
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap_err();
        match err {
            EngineError::UnterminatedFenceInStoredBody {
                ref section,
                ref fence,
                ref swallowed,
                ..
            } => {
                assert_eq!(section, "identity");
                assert_eq!(fence, "```");
                // Every declared section after the open fence, not just the
                // first: the fence's range reaches end of text, so it took
                // all of them.
                assert_eq!(
                    swallowed,
                    &vec![
                        "Purpose".to_string(),
                        "Specifies".to_string(),
                        "Constraints".to_string(),
                        "Rationale".to_string(),
                    ]
                );
            }
            other => panic!("expected UnterminatedFenceInStoredBody, got {other:?}"),
        }
        assert_eq!(err.code(), "UNTERMINATED_FENCE_IN_STORED_BODY");
    }

    #[test]
    fn replacing_the_absorbing_section_is_the_way_out() {
        // Criterion 6. The refusal above strands nothing: the caller lifts the
        // swallowed content back out in the same call, and editing the file
        // directly is forbidden by the workspace rule, so this route has to
        // exist through the engine.
        let tmp = TempDir::new().unwrap();
        let (mut engine, id) = engine_with_open_fence(&tmp);
        let (actor, client) = cli_actor();
        let hash = engine.get_entity(&id).unwrap().content_hash.clone();
        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: id.clone(),
                    expected_hash: Some(hash),
                    sections: IndexMap::from_iter([
                        (
                            "identity".to_string(),
                            "intro\n\n```rust\nfn main() {}\n```".to_string(),
                        ),
                        ("purpose".to_string(), "the recovered purpose".to_string()),
                    ]),
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .expect("a corrected body for the absorbing section is admitted");
        assert!(
            outcome
                .modified_sections
                .replaced
                .contains(&"identity".to_string())
        );
        let fixed = engine.get_entity(&id).unwrap();
        assert_eq!(
            fixed.sections.get("purpose").map(String::as_str),
            Some("the recovered purpose"),
            "the swallowed section is a section again"
        );
        assert!(
            crate::markdown::closing_fence_if_unterminated(fixed.sections.get("identity").unwrap())
                .is_none()
        );
    }

    /// The bypass the first version of this fix left open. The gate lived in
    /// `update_entity`, so every OTHER verb that regenerates the file walked
    /// past it and froze the absorption anyway. `relate` is the cheapest
    /// witness; `rename` reaches the same render. The gate now sits at the
    /// render itself, so this holds for any verb that writes bytes.
    #[test]
    fn every_verb_that_regenerates_the_file_is_gated_not_only_update() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, id) = engine_with_open_fence(&tmp);
        let (actor, client) = cli_actor();
        let before =
            std::fs::read_to_string(tmp.path().join(&engine.get_entity(&id).unwrap().file_path))
                .unwrap();
        let hash = engine.get_entity(&id).unwrap().content_hash.clone();

        let err = engine
            .relate_entity(
                RelateEntityArgs {
                    source: id.clone(),
                    expected_hash: Some(hash),
                    rel_type: "USES".to_string(),
                    target: crate::EntityId::new("specs", "some-target"),
                    remove: false,
                    description: None,
                    dry_run: false,
                },
                actor,
                Some(&client),
                None,
            )
            .expect_err("relate must not be able to freeze the absorption");
        assert_eq!(err.code(), "UNTERMINATED_FENCE_IN_STORED_BODY");

        let err = engine
            .rename_entity(
                crate::engine::RenameEntityArgs {
                    id: id.clone(),
                    new_title: "Renamed Fenced".to_string(),
                    expected_hash: Some(engine.get_entity(&id).unwrap().content_hash.clone()),
                },
                actor,
                Some(&client),
                None,
            )
            .expect_err("rename must not be able to freeze the absorption either");
        assert_eq!(err.code(), "UNTERMINATED_FENCE_IN_STORED_BODY");

        // And nothing was written: a refused mutation leaves the file alone.
        let after =
            std::fs::read_to_string(tmp.path().join(&engine.get_entity(&id).unwrap().file_path))
                .unwrap();
        assert_eq!(before, after, "a refused write must not touch the file");
    }

    #[test]
    fn an_entity_with_no_open_fence_updates_exactly_as_before() {
        // Criterion 7 at the write tier: the new gate is invisible to every
        // entity that does not carry the condition.
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Ordinary");
        let (actor, client) = cli_actor();
        engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections: IndexMap::from_iter([(
                        "purpose".to_string(),
                        "a new purpose".to_string(),
                    )]),
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .expect("an ordinary update is untouched by the fence gate");
    }

    /// Item 02: `memstead_update` against a stub must surface a typed
    /// `StubNotUpdatable` envelope rather than the pre-fix
    /// `UnknownType { name: "" }` cascade. Mirrors the
    /// `StubCannotRelate` guard that `memstead_relate` already runs;
    /// before Item 02 the docstring list advertised the
    /// `STUB_NOT_UPDATABLE` code but no engine path emitted it.
    #[test]
    fn update_entity_against_stub_surfaces_typed_stub_not_updatable() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, source) = engine_with_seed(&tmp, "Source");
        let (actor, client) = cli_actor();
        // Materialise a stub by relating from a real entity to an
        // absent target. The relate path upserts the stub.
        let stub_id = crate::EntityId::new("specs", "stub-update-target");
        engine
            .relate_entity(
                RelateEntityArgs {
                    source: source.id.clone(),
                    expected_hash: Some(source.content_hash.clone()),
                    rel_type: "USES".to_string(),
                    target: stub_id.clone(),
                    remove: false,
                    description: None,
                    dry_run: false,
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        let err = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: stub_id.clone(),
                    expected_hash: Some(String::new()),
                    sections: IndexMap::from_iter([("identity".to_string(), "body".to_string())]),
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap_err();
        match err {
            EngineError::StubNotUpdatable { id } => assert_eq!(id, stub_id.to_string()),
            other => panic!("expected StubNotUpdatable, got {other:?}"),
        }
    }

    #[test]
    fn update_entity_rejects_conflicting_section_modes() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Conflict");
        let (actor, client) = cli_actor();

        let mut sections = IndexMap::new();
        sections.insert("identity".to_string(), "replace".to_string());
        let mut appends = IndexMap::new();
        appends.insert("identity".to_string(), "append".to_string());

        let err = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections,
                    append_sections: appends,
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap_err();

        match err {
            EngineError::ConflictingSectionModes { section, modes } => {
                assert_eq!(section, "identity");
                assert_eq!(modes, vec!["sections", "append_sections"]);
            }
            other => panic!("expected ConflictingSectionModes, got {other:?}"),
        }
    }

    #[test]
    fn update_entity_rejects_overlapping_metadata_and_metadata_unset_keys() {
        // Wire contract: setting and unsetting the same key is a hard
        // error. The check runs before the required-field check so the
        // resolution (pick one map) is unambiguous regardless of
        // whether the conflicting key is required.
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Overlap Subject");
        let (actor, client) = cli_actor();

        let mut metadata = IndexMap::new();
        // `tags` is a non-required field on the default `spec` schema —
        // so this conflict is purely about the overlap, not about
        // unsetting-a-required-field.
        metadata.insert("tags".to_string(), "foo".to_string());

        let err = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections: IndexMap::new(),
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata,
                    metadata_unset: vec!["tags".to_string()],
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap_err();
        match err {
            EngineError::SetAndUnsetConflict { keys } => {
                assert_eq!(keys, vec!["tags".to_string()]);
            }
            other => panic!("expected SetAndUnsetConflict, got {other:?}"),
        }
    }

    #[test]
    fn update_entity_pointer_schema_auto_synthesises_references_from_body_link() {
        // Under the default schema's `alias_target_rel_type: REFERENCES`
        // pointer, a body wiki-link no longer trips the strict validator
        // — the alias-synthesis pass emits the REFERENCES relation
        // first, the validator finds the link backed, the body lands.
        // (Schemas without the pointer continue to refuse with
        // `WIKILINK_WITHOUT_RELATION`; that path is covered by the
        // dedicated no-pointer fixture test elsewhere.)
        use crate::EntityId;
        use crate::engine::UpdateEntityArgs;
        use indexmap::IndexMap;
        use tempfile::TempDir;

        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir.clone()),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        engine.set_workspace_root(mem_dir.clone());
        let (actor, client) = cli_actor();

        let target = engine
            .create_entity(
                empty_create_args("specs", "Target"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        let source = engine
            .create_entity(
                empty_create_args("specs", "Source"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        let mut sections: IndexMap<String, String> = IndexMap::new();
        sections.insert(
            "purpose".to_string(),
            "see [[target]] for context".to_string(),
        );
        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: source.id.clone(),
                    expected_hash: Some(source.content_hash.clone()),
                    sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .expect("auto-synthesis must satisfy the alias-existence invariant");
        // Body landed.
        assert!(
            outcome
                .modified_sections
                .replaced
                .iter()
                .any(|s| s == "purpose"),
        );
        let in_mem = engine.get_entity(&source.id).unwrap();
        assert_eq!(
            in_mem
                .sections
                .get("purpose")
                .map(String::as_str)
                .unwrap_or(""),
            "see [[target]] for context",
        );
        // REFERENCES relation synthesised from the body wiki-link.
        assert!(
            in_mem
                .relationships
                .iter()
                .any(|r| r.rel_type == "REFERENCES" && r.target == target.id),
            "synthesis must emit REFERENCES → target; relationships: {:?}",
            in_mem.relationships,
        );
        // Defeat unused-import warnings for the helper imports.
        let _ = EntityId::new("specs", "x");
    }

    #[test]
    fn update_entity_declare_relations_passes_strict_validator_in_one_call() {
        // The agent declares the relation + adds the body wiki-link
        // in a single `memstead_update` call. Without
        // `declare_relations`, the strict validator would refuse
        // (no backing relation yet); with the batched declaration,
        // the relation lands *before* the strict validator runs so
        // the body link passes the gate.
        use crate::engine::UpdateEntityArgs;
        use crate::ops::RelateArg;
        use indexmap::IndexMap;
        use tempfile::TempDir;

        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir.clone()),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        engine.set_workspace_root(mem_dir.clone());
        let (actor, client) = cli_actor();

        let target = engine
            .create_entity(
                empty_create_args("specs", "Target"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        let source = engine
            .create_entity(
                empty_create_args("specs", "Source"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        // Atomic declare + body update. USES (not REFERENCES) — under
        // the default schema's `alias_target_rel_type: REFERENCES`
        // pointer, explicit declare_relations type=REFERENCES is
        // refused; the body wiki-link is auto-emitted via synthesis.
        // The test's intent — that declare_relations atomically lands
        // alongside body changes — holds for any rel-type that admits
        // explicit authoring.
        let mut sections: IndexMap<String, String> = IndexMap::new();
        sections.insert(
            "purpose".to_string(),
            "see [[target]] for context".to_string(),
        );
        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                    id: source.id.clone(),
                    expected_hash: Some(source.content_hash.clone()),
                    sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    dry_run: false,
                    declare_relations: vec![RelateArg {
                        rel_type: "USES".to_string(),
                        target: target.id.clone(),
                        description: None,
                    }],
                },
                actor,
                Some(&client),
                None,
            )
            .expect("declare_relations + body update must succeed in one call");

        assert_eq!(outcome.relations_declared.len(), 1);
        assert_eq!(outcome.relations_declared[0].rel_type, "USES");
        assert_eq!(outcome.relations_declared[0].target, target.id);
        assert!(
            !outcome.relations_declared[0].target_was_stubbed,
            "target was already present in store; target_was_stubbed must be false"
        );

        let in_mem = engine.get_entity(&source.id).unwrap();
        assert!(
            in_mem.relationships.iter().any(|r| r.target == target.id),
            "declared relation must land in entity.relationships; got {:?}",
            in_mem.relationships
        );
    }

    #[test]
    fn update_entity_declare_relations_auto_stubs_absent_target() {
        // When the declared target doesn't exist yet, the engine
        // auto-stubs it (same mechanic as `memstead_relate`) and flags
        // `target_was_stubbed: true` in the outcome.
        use crate::EntityId;
        use crate::engine::UpdateEntityArgs;
        use crate::ops::RelateArg;
        use indexmap::IndexMap;

        let tmp = TempDir::new().unwrap();
        let (mut engine, source) = engine_with_seed(&tmp, "Source");
        let (actor, client) = cli_actor();
        let absent_target = EntityId::new("specs", "not-yet-existing");
        assert!(!engine.store().contains(&absent_target));

        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                    id: source.id.clone(),
                    expected_hash: Some(source.content_hash.clone()),
                    sections: IndexMap::new(),
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    dry_run: false,
                    declare_relations: vec![RelateArg {
                        rel_type: "USES".to_string(),
                        target: absent_target.clone(),
                        description: None,
                    }],
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        assert_eq!(outcome.relations_declared.len(), 1);
        assert!(
            outcome.relations_declared[0].target_was_stubbed,
            "absent target must be auto-stubbed; got target_was_stubbed=false"
        );
        // Stub now exists in the store.
        assert!(engine.store().contains(&absent_target));
        let stub = engine.get_entity(&absent_target).unwrap();
        assert!(stub.stub);
    }

    #[test]
    fn update_entity_alias_synthesis_runs_unconditionally_for_pointer_schemas() {
        // Under the alias model with a pointer-set schema (default
        // schema's `alias_target_rel_type: REFERENCES`), a fresh
        // workspace's first body-wiki-link write triggers the
        // alias-synthesis pass and the mutation lands with the
        // REFERENCES relation auto-emitted.
        use crate::engine::UpdateEntityArgs;
        use indexmap::IndexMap;
        use tempfile::TempDir;

        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir.clone()),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        engine.set_workspace_root(mem_dir.clone());
        let (actor, client) = cli_actor();
        let target = engine
            .create_entity(
                empty_create_args("specs", "Target"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        let source = engine
            .create_entity(
                empty_create_args("specs", "Source"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        let mut sections: IndexMap<String, String> = IndexMap::new();
        sections.insert(
            "purpose".to_string(),
            "see [[target]] for context".to_string(),
        );
        engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: source.id.clone(),
                    expected_hash: Some(source.content_hash.clone()),
                    sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .expect("synthesis must back the wiki-link and let the body land");
        let in_mem = engine.get_entity(&source.id).unwrap();
        assert!(
            in_mem
                .relationships
                .iter()
                .any(|r| r.rel_type == "REFERENCES" && r.target == target.id),
            "synthesis must emit REFERENCES → target; relationships: {:?}",
            in_mem.relationships,
        );
    }

    #[test]
    fn update_entity_dry_run_returns_prospective_hash_without_writing() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Preview Subject");
        let (actor, client) = cli_actor();
        let original_hash = seeded.content_hash.clone();

        let mut sections = IndexMap::new();
        sections.insert("identity".to_string(), "preview body".to_string());

        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    // Stale-hash recovery path — dry_run skips the
                    // hash check, so a wrong expected_hash is OK.
                    expected_hash: Some("wrong-hash".to_string()),
                    sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: true,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        // Wire shape: content_hash = current; prospective_hash =
        // what the write would produce; write_id empty.
        assert_eq!(outcome.content_hash, original_hash);
        let prospective = outcome
            .prospective_hash
            .expect("prospective_hash populated on dry_run");
        assert_ne!(prospective, original_hash);
        assert!(outcome.write_id.is_empty());
        // Store entity unchanged.
        let store_entity = engine.get_entity(&seeded.id).unwrap();
        assert_eq!(store_entity.content_hash, original_hash);
    }

    /// Edge-count round-trip lock. Captures the REFERENCES-drift
    /// finding:
    /// a mutation cycle (create body wiki-links → relate → update
    /// body → rename → delete) must return both `total_edges` and the
    /// REFERENCES counter to the pre-cycle values exactly. The bug
    /// was in `push_entities_into_store`: `upsert` preserved the
    /// entity's pre-existing out-edges, so `add_edge` (idempotent on
    /// `(from, to, rel_type)`) couldn't remove edges that the new
    /// parse no longer emits. Dropping a wiki-link from a body or
    /// absorbing one into an explicit relationship leaked the stale
    /// REFERENCES edge.
    ///
    /// Under the alias model the leak is structurally impossible —
    /// body wiki-links no longer emit edges, so the cleanup-on-reparse
    /// path the original test exercised has no premise. The test
    /// keeps the CRUD cycle but routes edges through atomic
    /// `relations:` declarations and explicit `memstead_relate`, which is
    /// what the model now treats as the only edge source.
    #[test]
    fn references_edges_round_trip_across_full_crud_cycle() {
        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        let (actor, client) = cli_actor();

        // Seed two link targets so the wiki-links inside the
        // probe entity's body resolve to real entities (not auto-
        // stubs we'd then have to GC).
        let foo = engine
            .create_entity(
                empty_create_args("specs", "Foo"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        let bar = engine
            .create_entity(
                empty_create_args("specs", "Bar"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        let count_references = |engine: &Engine| -> usize {
            engine
                .store()
                .all_ids()
                .flat_map(|id| engine.store().outgoing(id))
                .filter(|e| e.rel_type == "REFERENCES")
                .count()
        };

        let baseline_edges = engine.store().edge_count();
        let baseline_refs = count_references(&engine);

        // Step 1: create entity with body wiki-links — the
        // alias-synthesis pass auto-emits one REFERENCES per body
        // wiki-link (default schema's `alias_target_rel_type` →
        // REFERENCES), so the explicit `relations:` slot stays
        // empty. Net: 2 REFERENCES.
        let mut sections = IndexMap::new();
        sections.insert(
            "identity".to_string(),
            "See [[foo]] and [[bar]] inline.".to_string(),
        );
        sections.insert("purpose".to_string(), "probe purpose".to_string());
        let probe = engine
            .create_entity(
                CreateEntityArgs {
                    anchors: Vec::new(),
                    mem: "specs".to_string(),
                    title: "Probe".to_string(),
                    entity_type: "spec".to_string(),
                    sections,
                    metadata: IndexMap::new(),
                    relations: Vec::new(),
                    dry_run: false,
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        assert_eq!(count_references(&engine), baseline_refs + 2);

        // Step 2: relate INFORMED_BY → foo as a second relation to the
        // same target. The body wiki-link `[[foo]]` aliases the set of
        // relations to foo, so adding INFORMED_BY does not affect the
        // REFERENCES count — both relations coexist.
        let relate1 = engine
            .relate_entity(
                RelateEntityArgs {
                    source: probe.id.clone(),
                    expected_hash: Some(probe.content_hash.clone()),
                    rel_type: "INFORMED_BY".to_string(),
                    target: foo.id.clone(),
                    remove: false,
                    description: None,
                    dry_run: false,
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        assert_eq!(
            count_references(&engine),
            baseline_refs + 2,
            "set-membership aliasing — adding INFORMED_BY does not \
             absorb the REFERENCES relation"
        );

        // Step 3: drop the [[bar]] body link. The alias-synthesis pass
        // GCs the synthesised REFERENCES → bar atomically with the
        // body update — no second `memstead_relate --remove` needed.
        let mut sections = IndexMap::new();
        sections.insert("identity".to_string(), "See [[foo]] inline.".to_string());
        let updated = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: probe.id.clone(),
                    expected_hash: Some(relate1.content_hash.clone()),
                    sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        assert_eq!(
            count_references(&engine),
            baseline_refs + 1,
            "REFERENCES → bar must be auto-GC'd when its body link drops"
        );

        // Step 4: rename the entity. Edges follow via remove + push.
        let renamed = engine
            .rename_entity(
                crate::engine::RenameEntityArgs {
                    id: probe.id.clone(),
                    expected_hash: Some(updated.content_hash.clone()),
                    new_title: "Probe Renamed".to_string(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        assert_eq!(count_references(&engine), baseline_refs + 1);

        // Step 5: delete the renamed entity. The INFORMED_BY → foo
        // edge cascades; REFERENCES count unchanged.
        engine
            .delete_entity(
                crate::engine::DeleteEntityArgs {
                    id: renamed.new_id.clone(),
                    expected_hash: Some(renamed.content_hash.clone()),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        // Final assertion: every counter back to baseline.
        assert_eq!(
            engine.store().edge_count(),
            baseline_edges,
            "total edges must round-trip to baseline"
        );
        assert_eq!(
            count_references(&engine),
            baseline_refs,
            "REFERENCES counter must round-trip to baseline"
        );

        // Cross-check: a full reload of the mem produces the same
        // post-cycle counts. If the in-memory store and the on-disk
        // bytes drift, reload uncovers it.
        engine.reload_one_mem("specs").unwrap();
        assert_eq!(
            engine.store().edge_count(),
            baseline_edges,
            "total edges must match disk after reload"
        );
        assert_eq!(
            count_references(&engine),
            baseline_refs,
            "REFERENCES must match disk after reload"
        );
        // Sanity: foo + bar still in the store (they were not deleted).
        assert!(engine.store().contains(&foo.id));
        assert!(engine.store().contains(&bar.id));
    }

    #[test]
    fn update_entity_returns_write_id_title_modified_date_warnings_shape() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Subject");
        let (actor, client) = cli_actor();

        let mut sections = IndexMap::new();
        sections.insert("identity".to_string(), "edited body".to_string());

        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        // Folder backend produces a synthetic CommitId.
        assert!(
            !outcome.write_id.is_empty(),
            "write_id must be populated on a real update"
        );
        // Title echoed from the parsed entity post-write.
        assert_eq!(outcome.title, "Subject");
        // The default `spec` schema declares `modified_date` with
        // `auto_timestamp: true`; the unified update path
        // auto-stamps it. Asserting non-empty pins the
        // wire-shape parity with full's UpdateResult.modified_date.
        assert!(
            !outcome.modified_date.is_empty(),
            "modified_date must be auto-stamped on update for the default spec schema",
        );
        // V1: warnings always empty (typed warning surfaces are
        // separate session work). The vec is present on the outcome
        // so the wire shape parity with full's UpdateResult holds.
        assert!(outcome.warnings.is_empty());
        // Section was modified (existing behaviour, sanity check).
        assert_eq!(
            outcome.modified_sections.replaced,
            vec!["identity".to_string()]
        );
    }

    // ---- Engine::update_entity no-op detection ---------------------

    /// Re-setting a
    /// section to its current on-disk value short-circuits to
    /// `UPDATE_NOOP` and preserves `last_modified` at its pre-call
    /// value. Pre-fix the auto-timestamp stamped `last_modified` to
    /// `today_iso()` before the bytes-compare ran; the stamp
    /// synthesised a delta and the no-op never matched.
    #[test]
    fn update_entity_noop_resetting_section_to_current_value_preserves_last_modified() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Section Resetter");
        let (actor, client) = cli_actor();

        // Read the pre-update `last_modified` so we can assert it
        // survives the no-op.
        let pre_last_modified = engine
            .get_entity(&seeded.id)
            .and_then(|e| e.metadata.get("last_modified"))
            .map(|v| v.to_frontmatter_string())
            .expect("seeded entity has last_modified");

        // Re-set `identity` to its current on-disk body. The seed
        // helper writes "fixture identity body" — passing the same
        // string back must be a no-op.
        let mut sections = IndexMap::new();
        sections.insert("identity".to_string(), "fixture identity body".to_string());
        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        assert_eq!(outcome.write_id, "", "no-op must not commit");
        assert_eq!(
            outcome.content_hash, seeded.content_hash,
            "no-op must not advance content_hash",
        );
        assert!(
            outcome.warnings.iter().any(|w| w.code() == "UPDATE_NOOP"),
            "UPDATE_NOOP must fire on bytes-identical re-set",
        );
        assert_eq!(
            outcome.modified_date, pre_last_modified,
            "no-op must preserve last_modified at the pre-call value",
        );
        // The applied delta is empty on a
        // no-op — `modified_sections` must not claim `identity` was
        // replaced when nothing landed (matching the empty write_id
        // and unchanged hash above).
        assert!(
            outcome.modified_sections.replaced.is_empty()
                && outcome.modified_sections.appended.is_empty()
                && outcome.modified_sections.patched.is_empty(),
            "no-op must report an empty section delta, got {:?}",
            outcome.modified_sections,
        );

        // The on-disk entity also still carries the pre-update
        // last_modified — the no-op didn't bump it through some
        // other path.
        let post_last_modified = engine
            .get_entity(&seeded.id)
            .and_then(|e| e.metadata.get("last_modified"))
            .map(|v| v.to_frontmatter_string())
            .expect("entity still in store");
        assert_eq!(post_last_modified, pre_last_modified);
    }

    /// A payload with no
    /// recognised mutation content refuses with `EMPTY_UPDATE` BEFORE
    /// any engine work runs. Previously this same input short-
    /// circuited as a success-with-`UPDATE_NOOP`-warning, which
    /// hid a boundary-discipline failure mode (a
    /// misspelled mutation key deserialises to empty defaults and
    /// looks like a no-op success on the wire). The new refusal
    /// makes "no mutation content provided" structurally distinct
    /// from "mutation content provided but matched current state"
    /// (which still surfaces `UPDATE_NOOP` — see
    /// `update_entity_noop_same_content_surfaces_warning` below).
    #[test]
    fn update_entity_empty_payload_refuses_with_typed_code() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Empty Payload");
        let (actor, client) = cli_actor();

        let err = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections: IndexMap::new(),
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap_err();
        match err {
            EngineError::EmptyUpdate { id } => {
                assert_eq!(id, seeded.id.to_string());
            }
            other => panic!("expected EMPTY_UPDATE, got {other:?}"),
        }
        // No provenance row landed — the refusal preempts any write.
        let log_path = tmp.path().join(".memstead/changes.jsonl");
        if let Ok(log) = std::fs::read_to_string(&log_path) {
            let updates = log.matches("\"kind\":\"update\"").count();
            assert_eq!(updates, 0, "EMPTY_UPDATE refusal must not log an update");
        }
    }

    /// Complement: a
    /// payload with mutation content that matches the current entity
    /// state continues to land as success-with-`UPDATE_NOOP`-warning.
    /// The new `EMPTY_UPDATE` refusal applies only when no mutation
    /// content was provided; this path is structurally distinct.
    #[test]
    fn update_entity_noop_same_content_surfaces_warning() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Same Content Noop");
        let (actor, client) = cli_actor();

        // `empty_create_args` seeds `identity` with this exact body.
        let mut sections = IndexMap::new();
        sections.insert("identity".to_string(), "fixture identity body".to_string());

        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        assert_eq!(outcome.write_id, "");
        assert_eq!(outcome.content_hash, seeded.content_hash);
        let codes: Vec<&str> = outcome.warnings.iter().map(|w| w.code()).collect();
        assert!(
            codes.contains(&"UPDATE_NOOP"),
            "same-content update must surface UPDATE_NOOP; got {codes:?}",
        );
    }

    #[test]
    fn update_entity_noop_metadata_unset_on_absent_key() {
        // `metadata_unset=["never-set-key"]`
        // where the key was never set is a no-op — no field actually
        // changed, no commit advances, follow-up calls can chain
        // `expected_hash` without `HASH_MISMATCH`.
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Absent Key Noop");
        let (actor, client) = cli_actor();

        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections: IndexMap::new(),
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    // `tags` is declared on the `spec` schema but
                    // unset on the seeded entity. Unsetting it should
                    // be a no-op rather than producing a fresh commit.
                    metadata_unset: vec!["tags".to_string()],
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        assert_eq!(outcome.write_id, "");
        assert_eq!(outcome.content_hash, seeded.content_hash);
        assert!(
            outcome.warnings.iter().any(|w| w.code() == "UPDATE_NOOP"),
            "absent-key metadata_unset must surface UPDATE_NOOP",
        );
        // Empty applied delta on the no-op —
        // `unset` must not claim `tags` was removed when nothing landed.
        assert!(
            outcome.modified_metadata.set.is_empty() && outcome.modified_metadata.unset.is_empty(),
            "no-op must report an empty metadata delta, got {:?}",
            outcome.modified_metadata,
        );

        // Follow-up real change against the unchanged hash succeeds —
        // no HASH_MISMATCH cascade from a phantom advance.
        let mut sections = IndexMap::new();
        sections.insert("identity".to_string(), "real change".to_string());
        let real = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        assert!(!real.write_id.is_empty());
        assert_ne!(real.content_hash, seeded.content_hash);
    }

    /// The exact MCP repro — re-setting a
    /// metadata key to its current value no-ops, and the response's
    /// `modified_metadata` reports the applied delta (empty), not the
    /// requested key. Pre-fix the no-op short-circuit echoed
    /// `set: ["level"]` while `write_id` was empty and the hash
    /// unchanged — a self-contradictory response.
    #[test]
    fn update_entity_noop_setting_metadata_to_current_value_reports_empty_delta() {
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Stability Resetter");
        let (actor, client) = cli_actor();

        // `level` defaults to "M0" on the spec schema, so the seed
        // carries it. Re-setting it to "M0" changes nothing.
        let mut metadata = IndexMap::new();
        metadata.insert("level".to_string(), "M0".to_string());
        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections: IndexMap::new(),
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata,
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        assert_eq!(outcome.write_id, "", "no-op must not commit");
        assert_eq!(
            outcome.content_hash, seeded.content_hash,
            "no-op must not advance hash"
        );
        assert!(
            outcome.warnings.iter().any(|w| w.code() == "UPDATE_NOOP"),
            "re-set to current value must surface UPDATE_NOOP",
        );
        assert!(
            outcome.modified_metadata.set.is_empty() && outcome.modified_metadata.unset.is_empty(),
            "no-op must not claim `level` was set — applied delta is empty, got {:?}",
            outcome.modified_metadata,
        );
    }

    #[test]
    fn update_entity_noop_declare_already_related_edge() {
        // Re-declare an already-related edge
        // via `declare_relations` — no field, section, metadata or
        // relations list actually changes, so the bytes are identical
        // and the call no-ops.
        use crate::ops::RelateArg;
        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        let (actor, client) = cli_actor();
        let target = engine
            .create_entity(
                empty_create_args("specs", "Target Already Related"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        let source = engine
            .create_entity(
                empty_create_args("specs", "Source Already Related"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        let after_relate = engine
            .relate_entity(
                RelateEntityArgs {
                    source: source.id.clone(),
                    expected_hash: Some(source.content_hash.clone()),
                    rel_type: "USES".to_string(),
                    target: target.id.clone(),
                    remove: false,
                    description: None,
                    dry_run: false,
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        // Now re-declare the same edge via update.declare_relations.
        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                    id: source.id.clone(),
                    expected_hash: Some(after_relate.content_hash.clone()),
                    sections: IndexMap::new(),
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: vec![RelateArg {
                        rel_type: "USES".to_string(),
                        target: target.id.clone(),
                        description: None,
                    }],
                    dry_run: false,
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        assert_eq!(outcome.write_id, "");
        assert_eq!(outcome.content_hash, after_relate.content_hash);
        assert!(
            outcome.warnings.iter().any(|w| w.code() == "UPDATE_NOOP"),
            "duplicate declare must surface UPDATE_NOOP",
        );
        // `relations_declared` still records the entry — the per-
        // relation outcome is part of the surface, even for no-ops.
        assert_eq!(outcome.relations_declared.len(), 1);
        assert_eq!(outcome.relations_declared[0].rel_type, "USES");
        assert_eq!(outcome.relations_declared[0].target, target.id);
        assert!(!outcome.relations_declared[0].target_was_stubbed);
    }

    #[test]
    fn update_entity_real_change_still_commits_and_advances_hash() {
        // Regression: the no-op short-circuit must not short-circuit
        // real changes. A section replacement still produces a
        // non-empty `write_id`, advances `content_hash`, and does
        // NOT surface UPDATE_NOOP.
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Real Change Subject");
        let (actor, client) = cli_actor();

        let mut sections = IndexMap::new();
        sections.insert("identity".to_string(), "definitely new body".to_string());

        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        assert!(!outcome.write_id.is_empty(), "real change must commit");
        assert_ne!(
            outcome.content_hash, seeded.content_hash,
            "real change must advance content_hash",
        );
        assert!(
            !outcome.warnings.iter().any(|w| w.code() == "UPDATE_NOOP"),
            "real change must not surface UPDATE_NOOP",
        );
    }

    #[test]
    fn update_entity_noop_preserves_expected_hash_across_chain() {
        // A follow-up update with the original
        // hash after one or more no-ops succeeds because the hash
        // never advanced. Demonstrates the `expected_hash`-caching
        // posture the agent surface relies on.
        let tmp = TempDir::new().unwrap();
        let (mut engine, seeded) = engine_with_seed(&tmp, "Chained Noops Subject");
        let (actor, client) = cli_actor();

        // Two no-op calls in a row — both must return the same hash.
        // Pass same-content mutation
        // so UPDATE_NOOP fires (rather than EMPTY_UPDATE) and the
        // hash-chain invariant is exercised on the warning path.
        let mut noop_sections = IndexMap::new();
        noop_sections.insert("identity".to_string(), "fixture identity body".to_string());
        for _ in 0..2 {
            let outcome = engine
                .update_entity(
                    UpdateEntityArgs {
                        anchors: Vec::new(),
                        id: seeded.id.clone(),
                        expected_hash: Some(seeded.content_hash.clone()),
                        sections: noop_sections.clone(),
                        append_sections: IndexMap::new(),
                        patch_sections: IndexMap::new(),
                        metadata: IndexMap::new(),
                        metadata_unset: Vec::new(),
                        declare_relations: Vec::new(),
                        dry_run: false,
                        relations_unset: Vec::new(),
                        anchors_unset: Vec::new(),
                    },
                    actor,
                    Some(&client),
                    None,
                )
                .unwrap();
            assert_eq!(outcome.write_id, "");
            assert_eq!(outcome.content_hash, seeded.content_hash);
        }

        // Real follow-up with the original hash still works — no
        // HASH_MISMATCH cascade because the hash never advanced.
        let mut sections = IndexMap::new();
        sections.insert(
            "identity".to_string(),
            "third call: real change".to_string(),
        );
        let real = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: seeded.id.clone(),
                    expected_hash: Some(seeded.content_hash.clone()),
                    sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        assert!(!real.write_id.is_empty());
        assert_ne!(real.content_hash, seeded.content_hash);
    }

    // ---- Engine::delete_entity --------------------------------------

    // ---------------------------------------------------------------------
    // Alias-synthesis pass. Body wiki-links auto-emit
    // relations of the source schema's `alias_target_rel_type` pointer
    // and are garbage-collected when the body wiki-link disappears.
    // ---------------------------------------------------------------------

    #[test]
    fn synthesis_gc_drops_auto_emitted_reference_when_body_link_removed() {
        // Create with `[[target]]` in body → synthesis emits
        // REFERENCES. Update body to drop the wiki-link → GC drops
        // the auto-emitted REFERENCES.
        use crate::engine::{CreateEntityArgs, UpdateEntityArgs};
        use indexmap::IndexMap;
        use tempfile::TempDir;

        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir.clone()),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        engine.set_workspace_root(mem_dir.clone());
        let (actor, client) = cli_actor();

        let target = engine
            .create_entity(
                empty_create_args("specs", "Target"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        // Create source with the body wiki-link already present —
        // synthesis fires inside create.
        let mut sections: IndexMap<String, String> = IndexMap::new();
        sections.insert("identity".to_string(), "source identity".to_string());
        sections.insert(
            "purpose".to_string(),
            "see [[target]] for context".to_string(),
        );
        let source = engine
            .create_entity(
                CreateEntityArgs {
                    anchors: Vec::new(),
                    mem: "specs".to_string(),
                    title: "Source".to_string(),
                    entity_type: "spec".to_string(),
                    sections,
                    metadata: IndexMap::new(),
                    relations: Vec::new(),
                    dry_run: false,
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        assert!(
            engine
                .get_entity(&source.id)
                .unwrap()
                .relationships
                .iter()
                .any(|r| r.rel_type == "REFERENCES" && r.target == target.id),
            "create-time synthesis must emit REFERENCES → target",
        );

        // Update: drop the body wiki-link. GC should remove the
        // synthesised REFERENCES.
        let mut new_sections: IndexMap<String, String> = IndexMap::new();
        new_sections.insert("purpose".to_string(), "no link any more".to_string());
        engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: source.id.clone(),
                    expected_hash: Some(source.content_hash.clone()),
                    sections: new_sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .expect("update must succeed; GC drops the now-orphan REFERENCES");
        let in_mem = engine.get_entity(&source.id).unwrap();
        assert!(
            !in_mem
                .relationships
                .iter()
                .any(|r| r.rel_type == "REFERENCES" && r.target == target.id),
            "GC must drop the auto-emitted REFERENCES after body link removal; got {:?}",
            in_mem.relationships,
        );
    }

    #[test]
    fn update_gc_removes_orphan_stub_when_last_body_link_dropped() {
        // Create source with `[[ghost]]` body link → alias synthesis
        // auto-stubs `ghost` and emits REFERENCES → ghost. The update
        // drops the link, so the REFERENCES edge (the stub's only
        // referrer) disappears; the orphan-stub GC sweep removes the
        // stub and surfaces it in `orphan_stubs_removed`. A reload from
        // disk shows the same (decremented) stub count — proving the GC
        // was a real store mutation, not a session-local view fix.
        use crate::engine::{CreateEntityArgs, UpdateEntityArgs};
        use indexmap::IndexMap;
        use tempfile::TempDir;

        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir.clone()),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        engine.set_workspace_root(mem_dir.clone());
        let (actor, client) = cli_actor();

        let ghost = crate::EntityId::new("specs", "ghost");
        let mut sections: IndexMap<String, String> = IndexMap::new();
        sections.insert("identity".to_string(), "source identity".to_string());
        sections.insert(
            "purpose".to_string(),
            "see [[ghost]] for context".to_string(),
        );
        let source = engine
            .create_entity(
                CreateEntityArgs {
                    anchors: Vec::new(),
                    mem: "specs".to_string(),
                    title: "Source".to_string(),
                    entity_type: "spec".to_string(),
                    sections,
                    metadata: IndexMap::new(),
                    relations: Vec::new(),
                    dry_run: false,
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        assert!(
            engine.store().contains(&ghost) && engine.get_entity(&ghost).unwrap().stub,
            "body wiki-link to an absent target must auto-stub it",
        );
        assert_eq!(
            engine.health().stub_count,
            1,
            "one stub before the link drop"
        );

        let mut new_sections: IndexMap<String, String> = IndexMap::new();
        new_sections.insert("purpose".to_string(), "no link any more".to_string());
        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: source.id.clone(),
                    expected_hash: Some(source.content_hash.clone()),
                    sections: new_sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .expect("update must succeed and GC the now-orphan stub");

        assert_eq!(
            outcome.orphan_stubs_removed,
            vec![ghost.clone()],
            "the update that dropped the last body link must report the GC'd stub",
        );
        assert!(
            !engine.store().contains(&ghost),
            "orphan stub must be gone from the in-memory store",
        );
        assert_eq!(
            engine.health().stub_count,
            0,
            "stub count decremented in-session"
        );

        // Reload from disk: the source's on-disk markdown no longer
        // carries the link, so the parser re-emits no stub. The
        // decremented count holds across the reload — the GC was real.
        engine.reload_each_writable_mem().unwrap();
        assert!(
            !engine.store().contains(&ghost),
            "stub stays gone after reload-from-disk",
        );
        assert_eq!(
            engine.health().stub_count,
            0,
            "reloaded-from-disk store carries the same stub count as the in-session post-update state",
        );
    }

    #[test]
    fn update_gc_noop_when_section_edit_changes_no_body_link() {
        // An update that edits one section while leaving the `[[ghost]]`
        // link standing in another orphans nothing: `orphan_stubs_removed`
        // is present and empty (stable shape, no spurious GC), and the
        // stub survives because its referrer survives.
        use crate::engine::{CreateEntityArgs, UpdateEntityArgs};
        use indexmap::IndexMap;
        use tempfile::TempDir;

        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir.clone()),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        engine.set_workspace_root(mem_dir.clone());
        let (actor, client) = cli_actor();

        let ghost = crate::EntityId::new("specs", "ghost");
        let mut sections: IndexMap<String, String> = IndexMap::new();
        sections.insert("identity".to_string(), "original identity".to_string());
        sections.insert(
            "purpose".to_string(),
            "see [[ghost]] for context".to_string(),
        );
        let source = engine
            .create_entity(
                CreateEntityArgs {
                    anchors: Vec::new(),
                    mem: "specs".to_string(),
                    title: "Source".to_string(),
                    entity_type: "spec".to_string(),
                    sections,
                    metadata: IndexMap::new(),
                    relations: Vec::new(),
                    dry_run: false,
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        assert!(engine.store().contains(&ghost), "ghost stub materialised");

        // Replace `identity` only; the `[[ghost]]` link in `purpose`
        // stays, so no edge drops.
        let mut edit: IndexMap<String, String> = IndexMap::new();
        edit.insert("identity".to_string(), "edited identity".to_string());
        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: source.id.clone(),
                    expected_hash: Some(source.content_hash.clone()),
                    sections: edit,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .expect("update must succeed");
        assert!(
            outcome.orphan_stubs_removed.is_empty(),
            "an edit that keeps every body wiki-link orphans nothing; got {:?}",
            outcome.orphan_stubs_removed,
        );
        assert!(
            engine.store().contains(&ghost),
            "the still-referenced stub survives the unrelated section edit",
        );
    }

    #[test]
    fn update_gc_preserves_stub_with_surviving_referrer() {
        // Two sources both body-link `[[ghost]]`. Dropping the link from
        // one leaves `ghost` referenced by the other — set-membership
        // semantics keep the stub alive and `orphan_stubs_removed` empty.
        use crate::engine::{CreateEntityArgs, UpdateEntityArgs};
        use indexmap::IndexMap;
        use tempfile::TempDir;

        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir.clone()),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        engine.set_workspace_root(mem_dir.clone());
        let (actor, client) = cli_actor();

        let ghost = crate::EntityId::new("specs", "ghost");
        let make_with_link = |title: &str| {
            let mut sections: IndexMap<String, String> = IndexMap::new();
            sections.insert("identity".to_string(), format!("{title} identity"));
            sections.insert("purpose".to_string(), "see [[ghost]]".to_string());
            CreateEntityArgs {
                anchors: Vec::new(),
                mem: "specs".to_string(),
                title: title.to_string(),
                entity_type: "spec".to_string(),
                sections,
                metadata: IndexMap::new(),
                relations: Vec::new(),
                dry_run: false,
            }
        };
        let source_a = engine
            .create_entity(make_with_link("Source A"), actor, Some(&client), None)
            .unwrap();
        engine
            .create_entity(make_with_link("Source B"), actor, Some(&client), None)
            .unwrap();
        assert!(engine.store().contains(&ghost), "ghost stub materialised");

        // Drop the link from source A only.
        let mut drop_link: IndexMap<String, String> = IndexMap::new();
        drop_link.insert("purpose".to_string(), "no link here".to_string());
        let outcome = engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: source_a.id.clone(),
                    expected_hash: Some(source_a.content_hash.clone()),
                    sections: drop_link,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .expect("update must succeed");
        assert!(
            outcome.orphan_stubs_removed.is_empty(),
            "the stub keeps a referrer (source B), so nothing is GC'd; got {:?}",
            outcome.orphan_stubs_removed,
        );
        assert!(
            engine.store().contains(&ghost),
            "stub survives via the surviving referrer",
        );
    }

    #[test]
    fn synthesis_gc_preserves_non_pointer_explicit_relation_across_body_update() {
        // Explicit USES to target (USES is not the schema's
        // alias_target_rel_type pointer). A subsequent body-changing
        // update must NOT drop the USES edge — GC only touches
        // relations of the pointer rel-type. Under Option C, REFERENCES
        // can't be authored explicitly (`manual_authoring: forbidden`),
        // so the analogous "explicit REFERENCES preserved" scenario is
        // structurally impossible; USES exercises the same invariant
        // from the rel-type-discrimination side.
        use crate::engine::{RelateEntityArgs, UpdateEntityArgs};
        use indexmap::IndexMap;
        use tempfile::TempDir;

        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir.clone()),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        engine.set_workspace_root(mem_dir.clone());
        let (actor, client) = cli_actor();

        let target = engine
            .create_entity(
                empty_create_args("specs", "Target"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        let source = engine
            .create_entity(
                empty_create_args("specs", "Source"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        // Explicit relate — no body wiki-link.
        let relate = engine
            .relate_entity(
                RelateEntityArgs {
                    source: source.id.clone(),
                    expected_hash: Some(source.content_hash.clone()),
                    rel_type: "USES".to_string(),
                    target: target.id.clone(),
                    remove: false,
                    description: None,
                    dry_run: false,
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        // Update an unrelated section. The explicit USES must survive
        // — it's not the alias_target_rel_type, GC ignores it.
        let mut sections: IndexMap<String, String> = IndexMap::new();
        sections.insert("purpose".to_string(), "unrelated edit".to_string());
        engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: source.id.clone(),
                    expected_hash: Some(relate.content_hash.clone()),
                    sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .expect("update must succeed");
        let in_mem = engine.get_entity(&source.id).unwrap();
        assert!(
            in_mem
                .relationships
                .iter()
                .any(|r| r.rel_type == "USES" && r.target == target.id),
            "explicit USES must survive an unrelated body update; got {:?}",
            in_mem.relationships,
        );
    }

    #[test]
    fn synthesis_dedupes_repeated_body_links_to_same_target() {
        // Two `[[target]]` wiki-links in one body — synthesis must
        // not double-add. Result: exactly one REFERENCES.
        use crate::engine::UpdateEntityArgs;
        use indexmap::IndexMap;
        use tempfile::TempDir;

        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir.clone()),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        engine.set_workspace_root(mem_dir.clone());
        let (actor, client) = cli_actor();

        let target = engine
            .create_entity(
                empty_create_args("specs", "Target"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        let source = engine
            .create_entity(
                empty_create_args("specs", "Source"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        let mut sections: IndexMap<String, String> = IndexMap::new();
        sections.insert(
            "purpose".to_string(),
            "see [[target]] and again [[target]]".to_string(),
        );
        engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: source.id.clone(),
                    expected_hash: Some(source.content_hash.clone()),
                    sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        let in_mem = engine.get_entity(&source.id).unwrap();
        let count = in_mem
            .relationships
            .iter()
            .filter(|r| r.rel_type == "REFERENCES" && r.target == target.id)
            .count();
        assert_eq!(
            count, 1,
            "dedupe must leave exactly one REFERENCES → target; got {:?}",
            in_mem.relationships,
        );
    }

    #[test]
    fn synthesis_coexists_with_explicit_uses_to_same_target() {
        // Explicit `USES` to target AND body wiki-link to target →
        // entity carries both USES and REFERENCES edges; synthesis
        // dedupes on `(rel_type, target)` so USES never suppresses
        // REFERENCES.
        use crate::engine::{RelateEntityArgs, UpdateEntityArgs};
        use indexmap::IndexMap;
        use tempfile::TempDir;

        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir.clone()),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        engine.set_workspace_root(mem_dir.clone());
        let (actor, client) = cli_actor();

        let target = engine
            .create_entity(
                empty_create_args("specs", "Target"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        let source = engine
            .create_entity(
                empty_create_args("specs", "Source"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        // Explicit USES.
        let relate = engine
            .relate_entity(
                RelateEntityArgs {
                    source: source.id.clone(),
                    expected_hash: Some(source.content_hash.clone()),
                    rel_type: "USES".to_string(),
                    target: target.id.clone(),
                    remove: false,
                    description: None,
                    dry_run: false,
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        // Body wiki-link to the same target — synthesis emits REFERENCES.
        let mut sections: IndexMap<String, String> = IndexMap::new();
        sections.insert(
            "purpose".to_string(),
            "we also reference [[target]]".to_string(),
        );
        engine
            .update_entity(
                UpdateEntityArgs {
                    anchors: Vec::new(),
                    id: source.id.clone(),
                    expected_hash: Some(relate.content_hash.clone()),
                    sections,
                    append_sections: IndexMap::new(),
                    patch_sections: IndexMap::new(),
                    metadata: IndexMap::new(),
                    metadata_unset: Vec::new(),
                    declare_relations: Vec::new(),
                    dry_run: false,
                    relations_unset: Vec::new(),
                    anchors_unset: Vec::new(),
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        let in_mem = engine.get_entity(&source.id).unwrap();
        assert!(
            in_mem
                .relationships
                .iter()
                .any(|r| r.rel_type == "USES" && r.target == target.id),
            "USES must survive — synthesis dedupes on (rel_type, target)",
        );
        assert!(
            in_mem
                .relationships
                .iter()
                .any(|r| r.rel_type == "REFERENCES" && r.target == target.id),
            "REFERENCES must be synthesised even though USES already targets the same entity",
        );
    }

    // ---------------------------------------------------------------------
    // Alias-synthesis integration tests against custom-schema engines.
    // The default schema pins `alias_target_rel_type: REFERENCES`; these
    // tests verify the engine doesn't hardcode that name by mounting
    // schemas with a non-REFERENCES pointer (proves name-agnosticism)
    // and schemas with no pointer at all (proves the strict
    // `WIKILINK_WITHOUT_RELATION` refusal still fires for opt-out
    // schemas). Both build the engine via `from_mounts_with_schemas_dir`
    // — the production path for workspace-authored schemas.
    // ---------------------------------------------------------------------

    mod alias_synthesis_custom_schema {
        use std::path::Path;

        use indexmap::IndexMap;
        use memstead_schema::SchemaRef;
        use tempfile::TempDir;

        use crate::backend::MemBackend;
        use crate::engine::test_helpers::*;
        use crate::engine::{CreateEntityArgs, Engine, EngineError, UpdateEntityArgs};
        use crate::storage::FilesystemMemWriter;
        use crate::workspace::{Mount, MountCapability, MountLifecycle, MountStorage};

        const TYPE_BODY: &str = r#"description: t
when_to_use: tests
sections:
  - key: body
    heading: Body
    required: true
    search_weight: 10.0
    catch_all: true
    write_rules: []
metadata_fields: []
title_weight: 100.0
text_fields:
  - body
hierarchy_relationship: _default
no_self_loop_relationships: []
updatable_fields:
  - title
  - body
health_required_fields:
  - body
staleness_threshold_days: 90
write_rules: []
"#;

        fn write_schema_files(root: &Path, name: &str, manifest: &str, types: &[(&str, &str)]) {
            let dir = root.join(name);
            std::fs::create_dir_all(dir.join("types")).unwrap();
            std::fs::write(dir.join("schema.yaml"), manifest).unwrap();
            for (type_name, body) in types {
                std::fs::write(dir.join("types").join(format!("{type_name}.yaml")), body).unwrap();
            }
        }

        fn make_type_yaml(name: &str) -> String {
            format!("name: {name}\n{TYPE_BODY}")
        }

        fn folder_mount_with_pin(mem: &str, path: std::path::PathBuf, pin: SchemaRef) -> Mount {
            Mount {
                mem: mem.to_string(),
                schema: Some(pin),
                storage: MountStorage::Folder { path },
                capability: MountCapability::Write,
                lifecycle: MountLifecycle::Eager,
                cross_linkable: true,
                migration_target: None,
            }
        }

        fn engine_with_schema(
            manifest: &str,
            type_yaml_name: &str,
            schema_name: &str,
            schema_version: semver::Version,
        ) -> (Engine, TempDir) {
            let tmp = TempDir::new().unwrap();
            let schemas_dir = tmp.path().join("schemas");
            std::fs::create_dir_all(&schemas_dir).unwrap();
            write_schema_files(
                &schemas_dir,
                schema_name,
                manifest,
                &[(type_yaml_name, &make_type_yaml(type_yaml_name))],
            );
            let mem_dir = tmp.path().join("mem");
            std::fs::create_dir_all(&mem_dir).unwrap();
            let writer = FilesystemMemWriter::new(mem_dir.clone());
            let pin = SchemaRef::new(schema_name, schema_version);
            let mount = folder_mount_with_pin("v", mem_dir, pin);
            let mut engine = Engine::from_mounts_with_schemas_dir(
                vec![(mount, Box::new(writer) as Box<dyn MemBackend>)],
                Some(&schemas_dir),
            )
            .expect("engine with custom schema constructs");
            engine.set_workspace_root(tmp.path().to_path_buf());
            (engine, tmp)
        }

        #[test]
        fn non_references_alias_pointer_emits_named_rel_type_from_body_link() {
            // Schema names CITES as the alias pointer — the engine
            // must emit CITES (not REFERENCES) from a body wiki-link.
            // Proves no hard-coded "REFERENCES" string anywhere in
            // the synthesis path.
            let manifest = r#"name: aliased
version: 0.1.0
description: alias-synthesis fixture using a non-REFERENCES pointer
when_to_use: tests prove the engine does not hard-code REFERENCES
types:
  - doc
relationships:
  mode: strict
  definitions:
    - name: CITES
      description: Citation — auto-emitted from body wiki-links
      default_weight: 0.5
    - name: PART_OF
      description: Hierarchy
      default_weight: 3.0
      acyclic: true
    - name: _default
      description: Fallback
      default_weight: 1.0
alias_target_rel_type: CITES
community:
  resolution: 1.0
  seed: 42
"#;
            let (mut engine, _tmp) =
                engine_with_schema(manifest, "doc", "aliased", semver::Version::new(0, 1, 0));
            let (actor, client) = cli_actor();

            let target = engine
                .create_entity(
                    CreateEntityArgs {
                        anchors: Vec::new(),
                        mem: "v".to_string(),
                        title: "Target".to_string(),
                        entity_type: "doc".to_string(),
                        sections: IndexMap::from_iter([(
                            "body".to_string(),
                            "target body".to_string(),
                        )]),
                        metadata: IndexMap::new(),
                        relations: Vec::new(),
                        dry_run: false,
                    },
                    actor,
                    Some(&client),
                    None,
                )
                .unwrap();

            let mut sections: IndexMap<String, String> = IndexMap::new();
            sections.insert("body".to_string(), "see [[target]]".to_string());
            let source = engine
                .create_entity(
                    CreateEntityArgs {
                        anchors: Vec::new(),
                        mem: "v".to_string(),
                        title: "Source".to_string(),
                        entity_type: "doc".to_string(),
                        sections,
                        metadata: IndexMap::new(),
                        relations: Vec::new(),
                        dry_run: false,
                    },
                    actor,
                    Some(&client),
                    None,
                )
                .expect("create must succeed; CITES is auto-emitted by synthesis");

            let in_mem = engine.get_entity(&source.id).unwrap();
            assert!(
                in_mem
                    .relationships
                    .iter()
                    .any(|r| r.rel_type == "CITES" && r.target == target.id),
                "synthesis must emit CITES (the pointer rel-type), not REFERENCES; got {:?}",
                in_mem.relationships,
            );
            assert!(
                !in_mem
                    .relationships
                    .iter()
                    .any(|r| r.rel_type == "REFERENCES"),
                "engine must not hard-code REFERENCES — pointer rel-type is CITES; got {:?}",
                in_mem.relationships,
            );
        }

        #[test]
        fn no_pointer_schema_refuses_unbacked_body_wiki_link() {
            // Schema declares no `alias_target_rel_type`. Body wiki-link
            // without a backing relation must refuse with
            // `WIKILINK_WITHOUT_RELATION` — the strict validator's
            // pre-Option-C semantics, preserved for opt-out schemas.
            let manifest = r#"name: no-alias
version: 0.1.0
description: schema without alias_target_rel_type pointer
when_to_use: tests prove strict validator still fires for opt-out schemas
types:
  - doc
relationships:
  mode: strict
  definitions:
    - name: USES
      description: Use
      default_weight: 1.0
    - name: PART_OF
      description: Hierarchy
      default_weight: 3.0
      acyclic: true
    - name: _default
      description: Fallback
      default_weight: 1.0
community:
  resolution: 1.0
  seed: 42
"#;
            let (mut engine, _tmp) =
                engine_with_schema(manifest, "doc", "no-alias", semver::Version::new(0, 1, 0));
            let (actor, client) = cli_actor();

            let target = engine
                .create_entity(
                    CreateEntityArgs {
                        anchors: Vec::new(),
                        mem: "v".to_string(),
                        title: "Target".to_string(),
                        entity_type: "doc".to_string(),
                        sections: IndexMap::from_iter([(
                            "body".to_string(),
                            "target body".to_string(),
                        )]),
                        metadata: IndexMap::new(),
                        relations: Vec::new(),
                        dry_run: false,
                    },
                    actor,
                    Some(&client),
                    None,
                )
                .unwrap();
            let source = engine
                .create_entity(
                    CreateEntityArgs {
                        anchors: Vec::new(),
                        mem: "v".to_string(),
                        title: "Source".to_string(),
                        entity_type: "doc".to_string(),
                        sections: IndexMap::from_iter([(
                            "body".to_string(),
                            "source body".to_string(),
                        )]),
                        metadata: IndexMap::new(),
                        relations: Vec::new(),
                        dry_run: false,
                    },
                    actor,
                    Some(&client),
                    None,
                )
                .unwrap();

            // Add a body wiki-link with no backing relation. The
            // synthesis pass is a no-op (no pointer), so the
            // validator surfaces `WIKILINK_WITHOUT_RELATION`.
            let mut sections: IndexMap<String, String> = IndexMap::new();
            sections.insert("body".to_string(), "see [[target]]".to_string());
            let err = engine
                .update_entity(
                    UpdateEntityArgs {
                        anchors: Vec::new(),
                        id: source.id.clone(),
                        expected_hash: Some(source.content_hash.clone()),
                        sections,
                        append_sections: IndexMap::new(),
                        patch_sections: IndexMap::new(),
                        metadata: IndexMap::new(),
                        metadata_unset: Vec::new(),
                        declare_relations: Vec::new(),
                        dry_run: false,
                        relations_unset: Vec::new(),
                        anchors_unset: Vec::new(),
                    },
                    actor,
                    Some(&client),
                    None,
                )
                .unwrap_err();
            match err {
                EngineError::WikiLinkWithoutRelation { from_id, missing } => {
                    assert_eq!(from_id, source.id.to_string());
                    assert_eq!(missing.len(), 1);
                    assert_eq!(missing[0].section_key, "body");
                    assert_eq!(missing[0].target_id, target.id.to_string());
                }
                other => panic!(
                    "no-pointer schema must refuse with WikiLinkWithoutRelation; got {other:?}"
                ),
            }
        }

        /// Restoring the
        /// pre-alias-synthesis invariant that every body wiki-link
        /// target carries a grammar-valid `EntityId`. Natural-form
        /// `[[Knowledge Graph]]` no longer slips through into a
        /// malformed auto-stub; the engine refuses with the typed
        /// `InvalidWikiLinkTarget` envelope and the
        /// `title_to_slug`-derived suggestion the agent lifts
        /// directly into a retry. Covers F1 of the 2026-05-18 CLI probe.
        #[test]
        fn natural_form_body_wiki_link_refuses_with_typed_envelope() {
            let manifest = r#"name: aliased
version: 0.1.0
description: alias-synthesis fixture
when_to_use: tests prove strict wiki-link grammar at mutation entry
types:
  - doc
relationships:
  mode: strict
  definitions:
    - name: REFERENCES
      description: Reference — auto-emitted from body wiki-links
      default_weight: 0.5
    - name: PART_OF
      description: Hierarchy
      default_weight: 3.0
      acyclic: true
    - name: _default
      description: Fallback
      default_weight: 1.0
alias_target_rel_type: REFERENCES
community:
  resolution: 1.0
  seed: 42
"#;
            let (mut engine, _tmp) =
                engine_with_schema(manifest, "doc", "aliased", semver::Version::new(0, 1, 0));
            let (actor, client) = cli_actor();

            let mut sections: IndexMap<String, String> = IndexMap::new();
            sections.insert("body".to_string(), "see [[Knowledge Graph]]".to_string());
            let err = engine
                .create_entity(
                    CreateEntityArgs {
                        anchors: Vec::new(),
                        mem: "v".to_string(),
                        title: "Source".to_string(),
                        entity_type: "doc".to_string(),
                        sections,
                        metadata: IndexMap::new(),
                        relations: Vec::new(),
                        dry_run: false,
                    },
                    actor,
                    Some(&client),
                    None,
                )
                .unwrap_err();
            match err {
                EngineError::InvalidWikiLinkTarget {
                    raw,
                    suggested,
                    section,
                    link_source,
                    ..
                } => {
                    assert_eq!(raw, "Knowledge Graph");
                    assert_eq!(suggested.as_deref(), Some("knowledge-graph"));
                    assert_eq!(section, "body");
                    assert_eq!(link_source, "body_link");
                }
                other => panic!(
                    "natural-form body wiki-link must refuse with InvalidWikiLinkTarget; got {other:?}"
                ),
            }
        }

        /// Tier-2 body wiki-link with a non-conformant
        /// mem prefix refuses with the distinct `InvalidMemName`
        /// (wire code `INVALID_MEM_NAME`) — mems are fixed
        /// identifiers, not free-form text the agent can slugify, so
        /// the recovery path is different from `InvalidWikiLinkTarget`.
        #[test]
        fn tier_two_bad_mem_prefix_refuses_with_distinct_envelope() {
            let manifest = r#"name: aliased
version: 0.1.0
description: alias-synthesis fixture
when_to_use: tests prove strict mem-prefix grammar at mutation entry
types:
  - doc
relationships:
  mode: strict
  definitions:
    - name: REFERENCES
      description: Reference
      default_weight: 0.5
    - name: PART_OF
      description: Hierarchy
      default_weight: 3.0
      acyclic: true
    - name: _default
      description: Fallback
      default_weight: 1.0
alias_target_rel_type: REFERENCES
community:
  resolution: 1.0
  seed: 42
"#;
            let (mut engine, _tmp) =
                engine_with_schema(manifest, "doc", "aliased", semver::Version::new(0, 1, 0));
            let (actor, client) = cli_actor();

            let mut sections: IndexMap<String, String> = IndexMap::new();
            sections.insert("body".to_string(), "see [[Other Mem:foo]]".to_string());
            let err = engine
                .create_entity(
                    CreateEntityArgs {
                        anchors: Vec::new(),
                        mem: "v".to_string(),
                        title: "Source".to_string(),
                        entity_type: "doc".to_string(),
                        sections,
                        metadata: IndexMap::new(),
                        relations: Vec::new(),
                        dry_run: false,
                    },
                    actor,
                    Some(&client),
                    None,
                )
                .unwrap_err();
            match err {
                EngineError::InvalidWikiLinkMem { raw, section, .. } => {
                    assert_eq!(raw, "Other Mem");
                    assert_eq!(section, "body");
                }
                other => panic!(
                    "Tier-2 bad mem prefix must refuse with InvalidWikiLinkMem; got {other:?}"
                ),
            }
        }

        /// Body wiki-link
        /// containing the ambiguous `[[<segments>/<segments>--<slug>]]`
        /// form refuses with `InvalidWikiLinkTarget` carrying the
        /// colon-form (`<prefix>:<slug>`) as `suggested`. Pre-fix the
        /// dash form silently produced a same-mem phantom stub at
        /// slug `team/sub-mem--target`, losing the agent's intent.
        #[test]
        fn hierarchical_dash_form_body_link_refuses_with_colon_suggestion() {
            let manifest = r#"name: aliased
version: 0.1.0
description: alias-synthesis fixture
when_to_use: tests prove hierarchical dash-form refusal at mutation entry
types:
  - doc
relationships:
  mode: strict
  definitions:
    - name: REFERENCES
      description: Reference — auto-emitted from body wiki-links
      default_weight: 0.5
    - name: PART_OF
      description: Hierarchy
      default_weight: 3.0
      acyclic: true
    - name: _default
      description: Fallback
      default_weight: 1.0
alias_target_rel_type: REFERENCES
community:
  resolution: 1.0
  seed: 42
"#;
            let (mut engine, _tmp) =
                engine_with_schema(manifest, "doc", "aliased", semver::Version::new(0, 1, 0));
            let (actor, client) = cli_actor();

            let mut sections: IndexMap<String, String> = IndexMap::new();
            sections.insert(
                "body".to_string(),
                "see [[team/sub-mem--target]]".to_string(),
            );
            let err = engine
                .create_entity(
                    CreateEntityArgs {
                        anchors: Vec::new(),
                        mem: "v".to_string(),
                        title: "Source".to_string(),
                        entity_type: "doc".to_string(),
                        sections,
                        metadata: IndexMap::new(),
                        relations: Vec::new(),
                        dry_run: false,
                    },
                    actor,
                    Some(&client),
                    None,
                )
                .unwrap_err();
            match err {
                EngineError::InvalidWikiLinkTarget {
                    raw,
                    suggested,
                    section,
                    link_source,
                    ..
                } => {
                    assert_eq!(raw, "team/sub-mem--target");
                    assert_eq!(suggested.as_deref(), Some("team/sub-mem:target"));
                    assert_eq!(section, "body");
                    assert_eq!(link_source, "body_link");
                }
                other => panic!(
                    "hierarchical dash-form body link must refuse with InvalidWikiLinkTarget; got {other:?}"
                ),
            }

            // The entity did not land — no phantom stub for the source,
            // no phantom stub for the would-be target.
            let listed = engine.store().all_entities().collect::<Vec<_>>();
            assert!(
                listed.is_empty(),
                "refused create must not leave any entity behind, got: {listed:?}"
            );
        }
    }

    // ---- relations_unset repair-power gating -------------------------

    /// Markdown for a deliberately non-conformant `spec`: carries an
    /// undeclared metadata field (`zzz_bogus_field`) plus one USES
    /// relation. Written straight to disk before engine construction —
    /// the write path refuses non-conformant entities, so out-of-band
    /// state is the only way to seed one (which is exactly the
    /// repair-power scenario: drift entered outside the engine).
    const DRIFTED_MD: &str = "---\ntype: spec\ncreated_date: 2026-01-15\nlast_modified: 2026-04-12\nzzz_bogus_field: x\n---\n# Drifted\n\n## Identity\n\nNon-conformant fixture.\n\n## Purpose\n\nRepair-gate tests.\n\n## Relationships\n\n- **USES**: [[anchor]]\n";

    fn repair_engine() -> (TempDir, Engine) {
        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        std::fs::write(
            mem_dir.join("anchor.md"),
            "---\ntype: spec\ncreated_date: 2026-01-15\nlast_modified: 2026-04-12\n---\n# Anchor\n\n## Identity\n\nTarget.\n\n## Purpose\n\nRelation target.\n",
        )
        .unwrap();
        std::fs::write(mem_dir.join("drifted.md"), DRIFTED_MD).unwrap();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        (tmp, engine)
    }

    fn repair_args(id: EntityId, hash: Option<String>) -> UpdateEntityArgs {
        UpdateEntityArgs {
            anchors: Vec::new(),
            id,
            expected_hash: hash,
            sections: IndexMap::new(),
            append_sections: IndexMap::new(),
            patch_sections: IndexMap::new(),
            metadata: IndexMap::new(),
            metadata_unset: Vec::new(),
            declare_relations: Vec::new(),
            dry_run: false,
            relations_unset: vec![crate::ops::RelationUnsetArg {
                rel_type: "USES".to_string(),
                target: EntityId::new("specs", "anchor"),
            }],
            anchors_unset: Vec::new(),
        }
    }

    /// A conformant entity refuses repair-shaped input with
    /// `REPAIR_NOT_NEEDED` and is not modified — even though it has a
    /// relation that `relations_unset` names. The recovery text points
    /// at the focused detach path.
    #[test]
    fn relations_unset_on_conformant_entity_refuses_repair_not_needed() {
        let (_tmp, mut engine) = repair_engine();
        // `anchor` is conformant. Give it a relation first via the
        // ordinary relate path so there is something to (not) remove.
        let anchor = EntityId::new("specs", "anchor");
        let drifted = EntityId::new("specs", "drifted");
        engine
            .relate_entity(
                RelateEntityArgs {
                    source: anchor.clone(),
                    expected_hash: None,
                    rel_type: "USES".to_string(),
                    target: drifted.clone(),
                    remove: false,
                    description: None,
                    dry_run: false,
                },
                Actor::Cli,
                None,
                None,
            )
            .expect("relate on conformant entity works");
        let mut args = repair_args(anchor.clone(), None);
        args.relations_unset[0].target = drifted.clone();
        let err = engine
            .update_entity(args, Actor::Cli, None, None)
            .unwrap_err();
        match err {
            EngineError::RepairNotNeeded { id, recovery } => {
                assert_eq!(id, anchor.to_string());
                assert!(
                    recovery.contains("memstead_relate"),
                    "recovery must point at the focused tool; got {recovery}"
                );
            }
            other => panic!("expected RepairNotNeeded, got {other:?}"),
        }
        // Entity unmodified — the relation is still there.
        let entity = engine.store().get(&anchor).unwrap();
        assert!(
            entity.relationships.iter().any(|r| r.target == drifted),
            "gate must not modify the entity"
        );
    }

    /// A non-conformant entity accepts `relations_unset`: the named
    /// relation is removed atomically within the same update that also
    /// repairs the conformance break (`metadata_unset` on the
    /// undeclared field). The post-write entity is integral.
    #[test]
    fn relations_unset_repairs_non_conformant_entity_atomically() {
        let (_tmp, mut engine) = repair_engine();
        let drifted = EntityId::new("specs", "drifted");
        // Pre-state really is non-conformant.
        let pre = engine.conformance_findings("specs", None).unwrap();
        assert!(
            pre.iter().any(|f| f.id == drifted.to_string()),
            "fixture must lint non-conformant; got {pre:?}"
        );
        let mut args = repair_args(drifted.clone(), None);
        args.metadata_unset = vec!["zzz_bogus_field".to_string()];
        engine
            .update_entity(args, Actor::Cli, None, None)
            .expect("repair update lands");
        let entity = engine.store().get(&drifted).unwrap();
        assert!(
            entity.relationships.is_empty(),
            "relation must be removed; got {:?}",
            entity.relationships
        );
        assert!(
            !entity.metadata.contains_key("zzz_bogus_field"),
            "conformance break must be repaired in the same update"
        );
        let post = engine.conformance_findings("specs", None).unwrap();
        assert!(
            post.iter().all(|f| f.id != drifted.to_string()),
            "post-repair entity must be conformant; got {post:?}"
        );
    }

    /// Repair widens accepted inputs, never admissible outputs: a
    /// repair write whose post-state would violate the schema refuses
    /// with the relevant write-time code and nothing lands.
    #[test]
    fn relations_unset_post_state_must_still_validate() {
        let (_tmp, mut engine) = repair_engine();
        let drifted = EntityId::new("specs", "drifted");
        let mut args = repair_args(drifted.clone(), None);
        // Post-state violation: an unknown section alongside the
        // repair input.
        args.sections = IndexMap::from_iter([("nonexistent_section".to_string(), "x".to_string())]);
        let err = engine
            .update_entity(args, Actor::Cli, None, None)
            .unwrap_err();
        assert_eq!(
            err.code(),
            "UNKNOWN_SECTION",
            "strict-write post-condition must hold during repair; got {err:?}"
        );
        // Nothing landed: the relation survives.
        let entity = engine.store().get(&drifted).unwrap();
        assert!(
            !entity.relationships.is_empty(),
            "refused repair must not partially apply"
        );
    }

    /// Absent `(rel_type, target)` pairs are silent no-ops — symmetric
    /// with `metadata_unset` — so a repair retry is idempotent.
    #[test]
    fn relations_unset_absent_pair_is_silent_noop() {
        let (_tmp, mut engine) = repair_engine();
        let drifted = EntityId::new("specs", "drifted");
        let mut args = repair_args(drifted.clone(), None);
        args.relations_unset[0].rel_type = "NEVER_DECLARED".to_string();
        // Also repair the field so the post-state is integral.
        args.metadata_unset = vec!["zzz_bogus_field".to_string()];
        engine
            .update_entity(args, Actor::Cli, None, None)
            .expect("absent pair no-ops, update lands");
        let entity = engine.store().get(&drifted).unwrap();
        assert_eq!(
            entity.relationships.len(),
            1,
            "the USES relation must survive an unmatched unset"
        );
    }

    // ---- anchors merge / unset -------------------------------------------

    fn anchor_input(artifact: &str, hash: &str) -> crate::anchor::AnchorInput {
        crate::anchor::AnchorInput {
            artifact: Some(artifact.to_string()),
            grain: Some("file".to_string()),
            class: Some("anchored".to_string()),
            hash: Some(hash.to_string()),
            hash_stability: Some("stable".to_string()),
            ..Default::default()
        }
    }

    fn anchor_unset(artifact: &str) -> crate::anchor::AnchorUnsetInput {
        crate::anchor::AnchorUnsetInput {
            artifact: Some(artifact.to_string()),
            grain: None,
            class: None,
        }
    }

    /// Bare update-args shell for anchor tests — no content mutation.
    fn anchor_args(id: EntityId, hash: Option<String>) -> UpdateEntityArgs {
        UpdateEntityArgs {
            anchors: Vec::new(),
            anchors_unset: Vec::new(),
            id,
            expected_hash: hash,
            sections: IndexMap::new(),
            append_sections: IndexMap::new(),
            patch_sections: IndexMap::new(),
            metadata: IndexMap::new(),
            metadata_unset: Vec::new(),
            declare_relations: Vec::new(),
            dry_run: false,
            relations_unset: Vec::new(),
        }
    }

    /// Engine over a folder mount, seeded with one entity carrying two
    /// anchors (a.rs, b.rs). Returns the engine, tempdir, id, and hash.
    fn anchored_engine() -> (Engine, TempDir, EntityId, String) {
        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        let (actor, client) = cli_actor();
        let mut args = empty_create_args("specs", "Anchored");
        args.anchors = vec![anchor_input("a.rs", "h-a"), anchor_input("b.rs", "h-b")];
        let created = engine
            .create_entity(args, actor, Some(&client), None)
            .unwrap();
        let id = EntityId::new("specs", "anchored");
        assert_eq!(engine.entity_anchors(&id).len(), 2);
        (engine, tmp, id, created.content_hash)
    }

    /// Merge acceptance: one new anchor on N existing yields N+1 with the
    /// others byte-identical; a same-`(artifact, grain, class)` write
    /// replaces exactly that one; the motivating regression is dead —
    /// "anchor batch A, later anchor batch B" leaves A ∪ B queryable.
    #[test]
    fn update_anchors_merge_appends_and_replaces_by_triple() {
        let (mut engine, _tmp, id, hash) = anchored_engine();
        let (actor, client) = cli_actor();

        // Batch B: one new artifact. A ∪ B must survive.
        let mut args = anchor_args(id.clone(), Some(hash));
        args.anchors = vec![anchor_input("c.rs", "h-c")];
        let out = engine
            .update_entity(args, actor, Some(&client), None)
            .unwrap();
        let anchors = engine.entity_anchors(&id);
        assert_eq!(anchors.len(), 3, "N existing + 1 new = N+1");
        assert_eq!(anchors[0].artifact, "a.rs");
        assert_eq!(anchors[0].hash.as_deref(), Some("h-a"));
        assert_eq!(anchors[1].artifact, "b.rs");
        assert_eq!(anchors[2].artifact, "c.rs");
        assert!(!engine.anchors_referencing_artifact("a.rs").is_empty());
        assert!(!engine.anchors_referencing_artifact("c.rs").is_empty());

        // Same-triple write replaces exactly that one, in place.
        let mut args = anchor_args(id.clone(), Some(out.content_hash));
        args.anchors = vec![anchor_input("a.rs", "h-a2")];
        engine
            .update_entity(args, actor, Some(&client), None)
            .unwrap();
        let anchors = engine.entity_anchors(&id);
        assert_eq!(anchors.len(), 3);
        assert_eq!(anchors[0].artifact, "a.rs");
        assert_eq!(anchors[0].hash.as_deref(), Some("h-a2"));
        assert_eq!(anchors[1].hash.as_deref(), Some("h-b"), "b untouched");
        assert_eq!(anchors[2].hash.as_deref(), Some("h-c"), "c untouched");
    }

    /// Re-sending the full current set is a no-op on the stored sidecar
    /// bytes, and an update with an empty/absent `anchors` list leaves the
    /// stored set untouched.
    #[test]
    fn update_anchors_full_resend_and_absent_are_noops_on_stored_set() {
        let (mut engine, tmp, id, hash) = anchored_engine();
        let (actor, client) = cli_actor();
        let sidecar_path = tmp.path().join(crate::anchor::ANCHOR_SIDECAR_PATH);
        let before = std::fs::read(&sidecar_path).unwrap();

        // Full re-send of the current set.
        let mut args = anchor_args(id.clone(), Some(hash));
        args.anchors = vec![anchor_input("a.rs", "h-a"), anchor_input("b.rs", "h-b")];
        let out = engine
            .update_entity(args, actor, Some(&client), None)
            .unwrap();
        assert_eq!(
            std::fs::read(&sidecar_path).unwrap(),
            before,
            "full re-send keeps the stored bytes"
        );

        // Absent anchors list + a real content change: set untouched.
        let mut args = anchor_args(id.clone(), Some(out.content_hash));
        args.sections
            .insert("identity".to_string(), "changed body".to_string());
        engine
            .update_entity(args, actor, Some(&client), None)
            .unwrap();
        assert_eq!(
            std::fs::read(&sidecar_path).unwrap(),
            before,
            "an anchorless update never touches the stored set"
        );
    }

    /// Unset acceptance: bare-artifact unset removes all of that
    /// artifact's anchors and nothing else; a grain-narrowed unset removes
    /// only the match; a nonexistent target succeeds and changes nothing;
    /// unset + merge in one call apply unset-first.
    #[test]
    fn update_anchors_unset_bare_narrowed_idempotent_and_unset_first() {
        let (mut engine, _tmp, id, hash) = anchored_engine();
        let (actor, client) = cli_actor();

        // Add a span-grain anchor on a.rs so a.rs carries two grains.
        let mut span = anchor_input("a.rs", "h-span");
        span.grain = Some("span".to_string());
        let mut args = anchor_args(id.clone(), Some(hash));
        args.anchors = vec![span];
        let out = engine
            .update_entity(args, actor, Some(&client), None)
            .unwrap();
        assert_eq!(engine.entity_anchors(&id).len(), 3);

        // Narrowed unset: only the span-grain anchor goes.
        let mut narrowed = anchor_unset("a.rs");
        narrowed.grain = Some("span".to_string());
        let mut args = anchor_args(id.clone(), Some(out.content_hash));
        args.anchors_unset = vec![narrowed];
        let out = engine
            .update_entity(args, actor, Some(&client), None)
            .unwrap();
        let anchors = engine.entity_anchors(&id);
        assert_eq!(anchors.len(), 2);
        assert!(
            anchors
                .iter()
                .all(|a| a.grain == crate::anchor::AnchorGrain::File)
        );

        // Nonexistent target: succeeds, changes nothing.
        let mut args = anchor_args(id.clone(), Some(out.content_hash.clone()));
        args.anchors_unset = vec![anchor_unset("never-there.rs")];
        engine
            .update_entity(args, actor, Some(&client), None)
            .expect("unset of a nonexistent target is a no-op, not an error");
        assert_eq!(engine.entity_anchors(&id).len(), 2);

        // Unset + merge in one call: bare unset of a.rs plus a fresh a.rs
        // anchor — unset applies first, so the fresh anchor lands.
        let mut args = anchor_args(id.clone(), Some(out.content_hash));
        args.anchors_unset = vec![anchor_unset("a.rs")];
        args.anchors = vec![anchor_input("a.rs", "h-a-fresh")];
        engine
            .update_entity(args, actor, Some(&client), None)
            .unwrap();
        let anchors = engine.entity_anchors(&id);
        assert_eq!(anchors.len(), 2);
        assert_eq!(anchors[0].artifact, "b.rs", "b.rs untouched throughout");
        assert_eq!(anchors[1].hash.as_deref(), Some("h-a-fresh"));
    }

    /// Criterion 9 (consistency-sweep 03/03): one payload naming the same
    /// `(artifact, grain, class)` triple twice is refused. That triple is the
    /// sidecar's merge identity, so the repeats used to collapse to the last
    /// occurrence and the caller was never told an anchor it sent had gone.
    /// Criterion 10's complement rides along: the same artifact at two grains
    /// is two rows and still writes.
    #[test]
    fn a_payload_naming_one_triple_twice_is_refused() {
        let (mut engine, _tmp, id, hash) = anchored_engine();
        let (actor, client) = cli_actor();

        let mut args = anchor_args(id.clone(), Some(hash.clone()));
        args.anchors = vec![
            anchor_input("a.rs", "h-first"),
            anchor_input("a.rs", "h-second"),
        ];
        let err = engine
            .update_entity(args, actor, Some(&client), None)
            .expect_err("the repeated triple must refuse");
        assert_eq!(err.code(), crate::anchor::INVALID_ANCHOR_CODE);
        assert!(
            format!("{err}").contains("more than once"),
            "the refusal names the collapse: {err}"
        );

        // Refused before any state change: the stored rows are untouched.
        assert_eq!(engine.entity_anchors(&id).len(), 2);

        // The same artifact at a different grain is a different row, and
        // still writes in one payload.
        let mut span = anchor_input("a.rs", "h-span");
        span.grain = Some("span".to_string());
        let mut args = anchor_args(id.clone(), Some(hash));
        args.anchors = vec![anchor_input("a.rs", "h-file"), span];
        engine
            .update_entity(args, actor, Some(&client), None)
            .expect("two grains on one artifact are two rows");
        assert_eq!(engine.entity_anchors(&id).len(), 3);
    }

    /// Criterion 5 through the real write surface: a later call carrying the
    /// same triple with no hash keeps the stored baseline, rather than
    /// dropping it for the next verify to silently re-establish.
    #[test]
    fn a_re_pin_without_a_hash_keeps_the_stored_baseline() {
        let (mut engine, _tmp, id, hash) = anchored_engine();
        let (actor, client) = cli_actor();

        let mut hashless = anchor_input("a.rs", "");
        hashless.hash = None;
        let mut args = anchor_args(id.clone(), Some(hash));
        args.anchors = vec![hashless];
        engine
            .update_entity(args, actor, Some(&client), None)
            .unwrap();

        let kept = engine
            .entity_anchors(&id)
            .into_iter()
            .find(|a| a.artifact == "a.rs")
            .expect("the row is still there");
        assert_eq!(
            kept.hash.as_deref(),
            Some("h-a"),
            "the baseline the re-pin did not mention survives it"
        );
    }

    /// An anchor-write update never moves `_hash`, and an unset-only
    /// update is a real commit (not a no-op) that leaves the entity bytes
    /// untouched.
    #[test]
    fn update_anchor_only_and_unset_only_commit_without_hash_movement() {
        let (mut engine, _tmp, id, hash) = anchored_engine();
        let (actor, client) = cli_actor();

        let mut args = anchor_args(id.clone(), Some(hash.clone()));
        args.anchors_unset = vec![anchor_unset("b.rs")];
        let out = engine
            .update_entity(args, actor, Some(&client), None)
            .unwrap();
        assert!(
            !out.write_id.is_empty(),
            "unset-only update commits the sidecar"
        );
        assert_eq!(out.content_hash, hash, "anchors never move `_hash`");
        assert_eq!(engine.entity_anchors(&id).len(), 1);

        // Refusal complement: with no anchors, no unsets, and no content,
        // the empty-update guard still fires.
        let err = engine
            .update_entity(
                anchor_args(id.clone(), Some(hash)),
                actor,
                Some(&client),
                None,
            )
            .unwrap_err();
        assert!(matches!(err, EngineError::EmptyUpdate { .. }));
    }

    /// The same contract across a wall-clock second boundary: an
    /// anchor-only update landing a full second after the create must
    /// not auto-stamp `last_modified` (which would change the entity
    /// bytes and move `_hash`). Pre-fix, the anchor-only leg fell past
    /// the no-op guard into the unconditional auto-stamp and this
    /// failed whenever create and update straddled a second — the
    /// pinned clock makes that straddle deterministic.
    #[test]
    fn anchor_only_update_across_second_boundary_never_moves_hash() {
        let (mut engine, _tmp, id, hash) = anchored_engine();
        let (actor, client) = cli_actor();

        let t0 = std::time::UNIX_EPOCH + std::time::Duration::from_secs(1_778_243_696);
        engine.set_mutation_clock(std::sync::Arc::new(move || t0));
        // Re-create baseline under the pinned clock so the stamped
        // `last_modified` is exactly t0's second.
        let mut args = anchor_args(id.clone(), Some(hash));
        args.metadata = [("level".to_string(), "M1".to_string())]
            .into_iter()
            .collect();
        let restamped = engine
            .update_entity(args, actor, Some(&client), None)
            .unwrap();

        // One second later: anchor-only update.
        let t1 = t0 + std::time::Duration::from_secs(1);
        engine.set_mutation_clock(std::sync::Arc::new(move || t1));
        let mut args = anchor_args(id.clone(), Some(restamped.content_hash.clone()));
        args.anchors = vec![anchor_input("c.rs", "h-c")];
        let out = engine
            .update_entity(args, actor, Some(&client), None)
            .unwrap();
        assert!(!out.write_id.is_empty(), "anchor-only update commits");
        assert_eq!(
            out.content_hash, restamped.content_hash,
            "anchors never move `_hash`, even across a second boundary"
        );
        // The preserved `last_modified` is observable on the entity too.
        let entity = engine.store().get(&id).unwrap();
        assert_eq!(
            entity
                .metadata
                .get("last_modified")
                .and_then(|v| v.as_str()),
            Some("2026-05-08T12:34:56Z"),
            "anchor-only update must not restamp last_modified"
        );
    }

    /// A malformed `anchors_unset[]` selector refuses the whole update
    /// with the typed `INVALID_ANCHOR` envelope and nothing is written —
    /// merge introduces no partial-apply.
    #[test]
    fn malformed_anchor_unset_refuses_and_nothing_is_written() {
        let (mut engine, tmp, id, hash) = anchored_engine();
        let (actor, client) = cli_actor();
        let sidecar_path = tmp.path().join(crate::anchor::ANCHOR_SIDECAR_PATH);
        let before = std::fs::read(&sidecar_path).unwrap();

        let mut bad = anchor_unset("a.rs");
        bad.grain = Some("paragraph".to_string()); // unknown grain
        let mut args = anchor_args(id.clone(), Some(hash));
        args.anchors_unset = vec![bad];
        // A valid incoming anchor rides the same call — it must not land.
        args.anchors = vec![anchor_input("c.rs", "h-c")];
        let err = engine
            .update_entity(args, actor, Some(&client), None)
            .unwrap_err();
        assert_eq!(err.code(), crate::anchor::INVALID_ANCHOR_CODE);
        assert_eq!(engine.entity_anchors(&id).len(), 2, "no partial apply");
        assert_eq!(std::fs::read(&sidecar_path).unwrap(), before);
    }

    // ---- reserved metadata keys: set refused, unset is the repair --------

    /// Bare update-args shell for the reserved-key tests.
    fn bare_args(id: EntityId, hash: Option<String>) -> UpdateEntityArgs {
        UpdateEntityArgs {
            anchors: Vec::new(),
            anchors_unset: Vec::new(),
            id,
            expected_hash: hash,
            sections: IndexMap::new(),
            append_sections: IndexMap::new(),
            patch_sections: IndexMap::new(),
            metadata: IndexMap::new(),
            metadata_unset: Vec::new(),
            declare_relations: Vec::new(),
            dry_run: false,
            relations_unset: Vec::new(),
        }
    }

    /// On an entity fixture carrying historically smuggled reserved
    /// keys (`mem` / `id` in its frontmatter, written before the write
    /// gates closed), `metadata_unset` naming them succeeds, removes
    /// them from the store and the on-disk file, and the entity
    /// round-trips cleanly thereafter. Refusal complement: `metadata`
    /// (set) with a reserved key still refuses on update — single and
    /// batch.
    #[test]
    fn reserved_key_unset_repairs_smuggled_entity_and_set_stays_refused() {
        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        // Pre-gate fixture: frontmatter smuggles `mem` and `id`.
        std::fs::write(
            mem_dir.join("smuggled.md"),
            "---\ntype: spec\nmem: wrong-mem\nid: bogus-id\n---\n# Smuggled\n\n## Identity\n\nsmuggled identity.\n\n## Purpose\n\nsmuggled purpose.\n",
        )
        .unwrap();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir.clone()),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        let (actor, client) = cli_actor();
        let id = EntityId::new("specs", "smuggled");
        let entity = engine.get_entity(&id).expect("fixture boots");
        assert!(
            entity.metadata.contains_key("mem") && entity.metadata.contains_key("id"),
            "fixture must carry the smuggled keys after boot"
        );
        let hash = entity.content_hash.clone();

        // Refusal complement, single: SET of a reserved key refuses.
        for reserved in ["type", "mem", "id"] {
            let mut args = bare_args(id.clone(), Some(hash.clone()));
            args.metadata
                .insert(reserved.to_string(), "resmuggled".to_string());
            let err = engine
                .update_entity(args, actor, Some(&client), None)
                .expect_err("reserved-key set must refuse on update");
            assert_eq!(err.code(), "READ_ONLY_FIELD", "key '{reserved}': {err:?}");
        }
        // Refusal complement, batch: same refusal through batch_update
        // (atomic — nothing lands).
        let mut batch_item = bare_args(id.clone(), Some(hash.clone()));
        batch_item
            .metadata
            .insert("id".to_string(), "resmuggled".to_string());
        let batch = engine
            .batch_update(vec![(batch_item, None)], actor, Some(&client), false)
            .expect("batch returns a result envelope");
        assert!(
            !batch.applied,
            "batch with a reserved-key set must not apply"
        );
        assert_eq!(batch.failed, 1);

        // The sanctioned repair: unset both smuggled keys in one call.
        let mut args = bare_args(id.clone(), Some(hash));
        args.metadata_unset = vec!["mem".to_string(), "id".to_string()];
        let out = engine
            .update_entity(args, actor, Some(&client), None)
            .expect("reserved-key unset is the sanctioned repair");
        assert!(!out.write_id.is_empty(), "repair is a real commit");
        assert_eq!(
            out.modified_metadata.unset,
            vec!["mem".to_string(), "id".to_string()]
        );

        // Invariant restored: store and disk are clean, and the entity
        // round-trips through a further ordinary update.
        let entity = engine.get_entity(&id).expect("entity survives repair");
        assert!(
            !entity.metadata.contains_key("mem") && !entity.metadata.contains_key("id"),
            "smuggled keys must be gone from the store"
        );
        let on_disk = std::fs::read_to_string(mem_dir.join("smuggled.md")).unwrap();
        assert!(
            !on_disk.contains("wrong-mem") && !on_disk.contains("bogus-id"),
            "smuggled keys must be gone from the file: {on_disk}"
        );
        let mut args = bare_args(id.clone(), Some(entity.content_hash.clone()));
        args.sections
            .insert("identity".to_string(), "repaired identity".to_string());
        engine
            .update_entity(args, actor, Some(&client), None)
            .expect("post-repair entity round-trips cleanly");
    }

    /// Unsetting `type` never leaves an entity typeless: the engine
    /// re-seeds the authoritative discriminator, so on a healthy entity
    /// the unset is a committed-nothing no-op (`UPDATE_NOOP`) and the
    /// type survives on disk. (A missing `type:` would silently re-type
    /// the entity to the mem's default on the next parse — the re-seed
    /// forecloses that.) Unset of a nonexistent reserved key is equally
    /// a no-op, not an error.
    #[test]
    fn reserved_type_unset_reseeds_and_is_a_noop_on_healthy_entities() {
        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir.clone()),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        let (actor, client) = cli_actor();
        let created = engine
            .create_entity(
                empty_create_args("specs", "Healthy"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        let id = EntityId::new("specs", "healthy");

        for key in ["type", "mem", "id"] {
            let mut args = bare_args(id.clone(), Some(created.content_hash.clone()));
            args.metadata_unset = vec![key.to_string()];
            let out = engine
                .update_entity(args, actor, Some(&client), None)
                .unwrap_or_else(|e| panic!("unset '{key}' on a healthy entity must no-op: {e:?}"));
            assert!(
                out.write_id.is_empty(),
                "unset '{key}' on a healthy entity is a no-op, not a commit"
            );
            assert!(
                out.warnings.iter().any(|w| w.code() == "UPDATE_NOOP"),
                "no-op must carry the UPDATE_NOOP warning for '{key}'"
            );
        }
        let entity = engine.get_entity(&id).unwrap();
        assert_eq!(entity.entity_type, "spec");
        assert_eq!(
            entity.metadata.get("type").and_then(|v| v.as_str()),
            Some("spec"),
            "the discriminator survives a type unset"
        );
    }

    // ---- cycle family on declare_relations -------------------------------

    /// `update.declare_relations` runs the same cycle family as
    /// `memstead_relate`: a cycle-closing edge on an acyclic rel-type
    /// refuses `RELATIONSHIP_CYCLE` with the relate path's recovery
    /// detail, a self-loop on a listed no-self-loop rel-type refuses
    /// identically, and — refusal complement — a non-cycle edge on the
    /// acyclic type is accepted exactly as today.
    #[test]
    fn declare_relations_refuses_cycle_and_self_loop_like_relate() {
        let tmp = TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let mut engine = Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        let (actor, client) = cli_actor();

        // alpha PART_OF beta lands via create + relate.
        let alpha = engine
            .create_entity(
                empty_create_args("specs", "Alpha"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        let beta = engine
            .create_entity(
                empty_create_args("specs", "Beta"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        engine
            .relate_entity(
                crate::engine::RelateEntityArgs {
                    source: alpha.id.clone(),
                    target: beta.id.clone(),
                    rel_type: "PART_OF".to_string(),
                    remove: false,
                    expected_hash: None,
                    description: None,
                    dry_run: false,
                },
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        let declare = |rel_type: &str, from: &EntityId, to: &EntityId, hash: String| {
            let mut args = bare_args(from.clone(), Some(hash));
            args.declare_relations = vec![crate::ops::RelateArg {
                target: to.clone(),
                rel_type: rel_type.to_string(),
                description: None,
            }];
            args
        };

        // beta declaring PART_OF→alpha closes beta→alpha→beta.
        let err = engine
            .update_entity(
                declare("PART_OF", &beta.id, &alpha.id, beta.content_hash.clone()),
                actor,
                Some(&client),
                None,
            )
            .expect_err("cycle-closing declare_relations must refuse");
        assert_eq!(err.code(), "RELATIONSHIP_CYCLE", "{err:?}");
        let details = err.details();
        assert_eq!(details["rel_type"], "PART_OF");
        assert!(details["existing_path"].is_array());
        assert!(
            engine
                .get_entity(&beta.id)
                .unwrap()
                .relationships
                .is_empty(),
            "the refused edge must not land"
        );

        // Self-loop on a listed no-self-loop rel-type (spec lists USES).
        // Alpha's hash moved with the relate above — read the live one.
        let alpha_hash = engine.get_entity(&alpha.id).unwrap().content_hash.clone();
        let err = engine
            .update_entity(
                declare("USES", &alpha.id, &alpha.id, alpha_hash),
                actor,
                Some(&client),
                None,
            )
            .expect_err("self-loop declare_relations must refuse");
        assert_eq!(err.code(), "RELATIONSHIP_CYCLE", "{err:?}");

        // Refusal complement: a non-cycle PART_OF edge is accepted.
        engine
            .update_entity(
                declare(
                    "PART_OF",
                    &beta.id,
                    &EntityId::new("specs", "gamma"),
                    beta.content_hash.clone(),
                ),
                actor,
                Some(&client),
                None,
            )
            .expect("a non-cycle PART_OF declare must land as today");
    }
}