From eaaecc506277bc07992119a4c4c64b0c71e90359 Mon Sep 17 00:00:00 2001
From: David Spencer <1526975+DecisionNerd@users.noreply.github.com>
Date: Mon, 24 Aug 2026 01:35:59 -0600
Subject: [PATCH 1/6] fix(storage): make topology rewrites durably recoverable

---
 .../graphforge-storage/src/durable_rewrite.rs | 954 ++++++++++++++++++
 crates/graphforge-storage/src/generation.rs   | 146 ++-
 crates/graphforge-storage/src/lib.rs          |   5 +-
 crates/graphforge-storage/src/staging.rs      |  19 +-
 crates/graphforge-storage/src/writer.rs       |   4 +-
 5 files changed, 1036 insertions(+), 92 deletions(-)
 create mode 100644 crates/graphforge-storage/src/durable_rewrite.rs

diff --git a/crates/graphforge-storage/src/durable_rewrite.rs b/crates/graphforge-storage/src/durable_rewrite.rs
new file mode 100644
index 00000000..8d5ad31f
--- /dev/null
+++ b/crates/graphforge-storage/src/durable_rewrite.rs
@@ -0,0 +1,954 @@
+//! Durable recovery for mutable graph-file rewrite batches.
+//!
+//! Once the intent below is durable, recovery always rolls forward.  Data and
+//! auxiliary receipts are installed first; `topology/generation.json` is the
+//! final authority switch.  Journal paths are bounded, canonical relative
+//! paths and every recovery input is authenticated before it is used.
+
+use std::fs::File;
+use std::io::{Read, Seek, Write};
+use std::path::{Component, Path};
+
+use graphforge_core::GfError;
+use graphforge_filesystem::{FileIdentity, StableDirectory};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+use uuid::Uuid;
+
+use crate::filesystem_admission::{ProjectLifecycleMode, ProjectRootRequirement};
+use crate::staging::RewriteBatch;
+
+const JOURNAL: &str = ".graphforge-rewrite-v1.json";
+const LOCK: &str = ".graphforge-rewrite.lock";
+const MAX_ENTRIES: usize = 16_384;
+const MAX_JOURNAL_BYTES: u64 = 8 * 1024 * 1024;
+
+/// Authenticated control receipt committed atomically with a rewrite.
+///
+/// This generic hook lets another storage participant bind its own typed,
+/// durable receipt to the same generation-last transaction without coupling
+/// this layer to that participant's format.
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
+pub struct AuxiliaryReceipt {
+    /// Stable schema/type identifier understood by the auxiliary participant.
+    pub kind: String,
+    /// Participant receipt schema version.
+    pub schema_version: u32,
+    /// Canonical project-relative destination containing the receipt bytes.
+    pub path: String,
+    /// Lowercase SHA-256 digest of its exact durable receipt bytes.
+    pub digest: String,
+    /// Exact staged receipt length.
+    pub bytes: u64,
+}
+
+#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
+pub(crate) struct GenerationPair {
+    pub topology: u64,
+    pub search: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct Intent {
+    version: u8,
+    state: IntentState,
+    transaction: String,
+    root_volume: u64,
+    root_file: String,
+    prior: GenerationPair,
+    next: GenerationPair,
+    auxiliary: Option<AuxiliaryReceipt>,
+    entries: Vec<Entry>,
+    checksum: String,
+}
+
+#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+enum IntentState {
+    Preparing,
+    Durable,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct Entry {
+    class: EntryClass,
+    destination: String,
+    temporary: String,
+    parent_volume: u64,
+    parent_file: String,
+    bytes: u64,
+    sha256: String,
+    temporary_volume: u64,
+    temporary_file: String,
+}
+
+#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+enum EntryClass {
+    Data,
+    GenerationAuthority,
+}
+
+fn storage(error: impl std::fmt::Display) -> GfError {
+    GfError::Storage(error.to_string())
+}
+fn hex(bytes: &[u8]) -> String {
+    bytes.iter().map(|b| format!("{b:02x}")).collect()
+}
+
+fn canonical_relative(root: &Path, path: &Path) -> Result<String, GfError> {
+    let relative = path
+        .strip_prefix(root)
+        .map_err(|_| storage("rewrite destination escapes project root"))?;
+    if relative.as_os_str().is_empty()
+        || relative
+            .components()
+            .any(|c| !matches!(c, Component::Normal(_)))
+    {
+        return Err(storage(
+            "rewrite destination is not a canonical relative path",
+        ));
+    }
+    relative
+        .to_str()
+        .map(str::to_owned)
+        .ok_or_else(|| storage("rewrite destination is not UTF-8"))
+}
+
+fn identity(path: &Path) -> Result<(u64, String), GfError> {
+    let id = graphforge_filesystem::path_identity(path).map_err(storage)?;
+    Ok((id.volume_serial, hex(&id.file_id)))
+}
+
+fn hash_reader(mut file: File) -> Result<(u64, String), GfError> {
+    file.rewind().map_err(storage)?;
+    let mut hash = Sha256::new();
+    let mut bytes = 0_u64;
+    let mut buffer = [0_u8; 1024 * 1024];
+    loop {
+        let count = file.read(&mut buffer).map_err(storage)?;
+        if count == 0 {
+            break;
+        }
+        bytes = bytes
+            .checked_add(count as u64)
+            .ok_or_else(|| storage("rewrite byte count overflow"))?;
+        hash.update(&buffer[..count]);
+    }
+    Ok((bytes, hex(&hash.finalize())))
+}
+
+fn sync_dir(path: &Path) -> Result<(), GfError> {
+    #[cfg(unix)]
+    {
+        File::open(path).and_then(|f| f.sync_all()).map_err(storage)
+    }
+    #[cfg(not(unix))]
+    {
+        let _ = path;
+        Ok(())
+    }
+}
+
+struct RewriteGuard {
+    admission: crate::filesystem_admission::ProjectLifecycleAdmission,
+    directory: StableDirectory,
+    lifecycle: File,
+    lifecycle_identity: FileIdentity,
+}
+
+impl Drop for RewriteGuard {
+    fn drop(&mut self) {
+        let _ = crate::file_lock::unlock(&self.lifecycle);
+    }
+}
+
+fn acquire(root: &Path) -> Result<RewriteGuard, GfError> {
+    // The lifecycle guard binds the named project root. Ephemeral mode avoids
+    // repeating the expensive filesystem probe; durable projects have already
+    // passed it at facade admission.
+    let admission = crate::filesystem_admission::admit_project_lifecycle(
+        root,
+        ProjectLifecycleMode::Ephemeral,
+        ProjectRootRequirement::Existing,
+    )?;
+    let directory = StableDirectory::open(root).map_err(storage)?;
+    let lock = directory
+        .open_or_create_child_file(std::ffi::OsStr::new(LOCK))
+        .map_err(storage)?;
+    lock.sync_all().map_err(storage)?;
+    directory.sync().map_err(storage)?;
+    crate::file_lock::lock_exclusive(&lock).map_err(storage)?;
+    admission.revalidate_identity()?;
+    let lifecycle_identity = graphforge_filesystem::file_identity(&lock).map_err(storage)?;
+    Ok(RewriteGuard {
+        admission,
+        directory,
+        lifecycle: lock,
+        lifecycle_identity,
+    })
+}
+
+impl RewriteGuard {
+    fn revalidate(&self) -> Result<(), GfError> {
+        self.admission.revalidate_identity()?;
+        let named = self
+            .directory
+            .open_child_file(std::ffi::OsStr::new(LOCK))
+            .map_err(storage)?;
+        if graphforge_filesystem::file_identity(&named).map_err(storage)? != self.lifecycle_identity
+            || graphforge_filesystem::file_link_count(&named).map_err(storage)? != 1
+        {
+            return Err(storage("rewrite lifecycle lock identity changed"));
+        }
+        Ok(())
+    }
+}
+
+fn intent_bytes(intent: &Intent) -> Result<Vec<u8>, GfError> {
+    serde_json::to_vec(intent).map_err(storage)
+}
+
+fn checksum(intent: &Intent) -> Result<String, GfError> {
+    let unsigned = Intent {
+        checksum: String::new(),
+        version: intent.version,
+        state: intent.state,
+        transaction: intent.transaction.clone(),
+        root_volume: intent.root_volume,
+        root_file: intent.root_file.clone(),
+        prior: intent.prior,
+        next: intent.next,
+        auxiliary: intent.auxiliary.clone(),
+        entries: intent
+            .entries
+            .iter()
+            .map(|e| Entry {
+                destination: e.destination.clone(),
+                class: e.class,
+                temporary: e.temporary.clone(),
+                parent_volume: e.parent_volume,
+                parent_file: e.parent_file.clone(),
+                bytes: e.bytes,
+                sha256: e.sha256.clone(),
+                temporary_volume: e.temporary_volume,
+                temporary_file: e.temporary_file.clone(),
+            })
+            .collect(),
+    };
+    Ok(hex(&Sha256::digest(intent_bytes(&unsigned)?)))
+}
+
+fn publish_journal(root: &StableDirectory, intent: &Intent) -> Result<(), GfError> {
+    let bytes = intent_bytes(intent)?;
+    if bytes.len() as u64 > MAX_JOURNAL_BYTES {
+        return Err(storage("rewrite journal exceeds bound"));
+    }
+    let name = format!(".{JOURNAL}.{}.tmp", intent.transaction);
+    let mut temp = root
+        .create_replaceable_child_file(std::ffi::OsStr::new(&name))
+        .map_err(storage)?;
+    temp.write_all(&bytes)
+        .and_then(|_| temp.sync_all())
+        .map_err(storage)?;
+    let expected = graphforge_filesystem::file_identity(&temp).map_err(storage)?;
+    root.replace_child(
+        std::ffi::OsStr::new(&name),
+        expected,
+        std::ffi::OsStr::new(JOURNAL),
+    )
+    .map_err(storage)?;
+    root.sync().map_err(storage)
+}
+
+fn install(root_path: &Path, root: &StableDirectory, entry: &Entry) -> Result<(), GfError> {
+    let (parent, target) = retained_parent_at(root_path, root, &entry.destination)?;
+    let (_, temporary) = retained_parent_at(root_path, root, &entry.temporary)?;
+    let parent_id = parent.identity();
+    if (parent_id.volume_serial, hex(&parent_id.file_id))
+        != (entry.parent_volume, entry.parent_file.clone())
+    {
+        return Err(storage("rewrite parent identity changed"));
+    }
+    if let Ok(destination) = parent.open_child_file(&target) {
+        if hash_reader(destination)? == (entry.bytes, entry.sha256.clone()) {
+            if let Ok(temp) = parent.open_child_file(&temporary) {
+                let id = graphforge_filesystem::file_identity(&temp).map_err(storage)?;
+                drop(temp);
+                parent
+                    .unlink_child_if_identity(&temporary, id)
+                    .map_err(storage)?;
+                parent.sync().map_err(storage)?;
+            }
+            return Ok(());
+        }
+    }
+    let temp = parent.open_child_file(&temporary).map_err(storage)?;
+    let temp_id = graphforge_filesystem::file_identity(&temp).map_err(storage)?;
+    if (temp_id.volume_serial, hex(&temp_id.file_id))
+        != (entry.temporary_volume, entry.temporary_file.clone())
+    {
+        return Err(storage("rewrite temporary identity changed"));
+    }
+    if hash_reader(temp.try_clone().map_err(storage)?)? != (entry.bytes, entry.sha256.clone()) {
+        return Err(storage("rewrite recovery input is missing or corrupt"));
+    }
+    let expected = graphforge_filesystem::file_identity(&temp).map_err(storage)?;
+    drop(temp);
+    parent
+        .replace_child(&temporary, expected, &target)
+        .map_err(storage)?;
+    parent.sync().map_err(storage)?;
+    if hash_reader(parent.open_child_file(&target).map_err(storage)?)?
+        != (entry.bytes, entry.sha256.clone())
+    {
+        return Err(storage(
+            "installed rewrite destination failed authentication",
+        ));
+    }
+    Ok(())
+}
+
+fn retained_parent_at(
+    root_path: &Path,
+    root: &StableDirectory,
+    relative: &str,
+) -> Result<(StableDirectory, std::ffi::OsString), GfError> {
+    let path = Path::new(relative);
+    let mut components = path.components().peekable();
+    let mut directory = StableDirectory::open(root_path).map_err(storage)?;
+    if directory.identity() != root.identity() {
+        return Err(storage("rewrite root identity changed"));
+    }
+    while let Some(component) = components.next() {
+        let Component::Normal(name) = component else {
+            return Err(storage("non-canonical journal path"));
+        };
+        if components.peek().is_none() {
+            return Ok((directory, name.to_os_string()));
+        }
+        directory = directory.open_child_directory(name).map_err(storage)?;
+    }
+    Err(storage("empty journal path"))
+}
+
+fn remove_journal(root: &StableDirectory) -> Result<(), GfError> {
+    let file = root
+        .open_child_file(std::ffi::OsStr::new(JOURNAL))
+        .map_err(storage)?;
+    let id = graphforge_filesystem::file_identity(&file).map_err(storage)?;
+    drop(file);
+    root.unlink_child_if_identity(std::ffi::OsStr::new(JOURNAL), id)
+        .map_err(storage)?;
+    root.sync().map_err(storage)
+}
+
+fn read_journal(root: &StableDirectory) -> Result<Option<(Vec<u8>, FileIdentity)>, GfError> {
+    let file = match root.open_child_file(std::ffi::OsStr::new(JOURNAL)) {
+        Ok(file) => file,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+        Err(error) => return Err(storage(error)),
+    };
+    let id = graphforge_filesystem::file_identity(&file).map_err(storage)?;
+    let metadata = file.metadata().map_err(storage)?;
+    if metadata.len() > MAX_JOURNAL_BYTES {
+        return Err(storage("rewrite journal exceeds bound"));
+    }
+    let mut bytes = Vec::with_capacity(metadata.len() as usize);
+    file.take(MAX_JOURNAL_BYTES + 1)
+        .read_to_end(&mut bytes)
+        .map_err(storage)?;
+    Ok(Some((bytes, id)))
+}
+
+fn recover_locked(
+    root_path: &Path,
+    root: &StableDirectory,
+    current: GenerationPair,
+) -> Result<(), GfError> {
+    let Some((bytes, _journal_id)) = read_journal(root)? else {
+        return Ok(());
+    };
+    let intent: Intent = serde_json::from_slice(&bytes)
+        .map_err(|e| storage(format!("corrupt rewrite journal: {e}")))?;
+    if intent.version != 1
+        || intent.entries.len() > MAX_ENTRIES
+        || checksum(&intent)? != intent.checksum
+    {
+        return Err(storage("rewrite journal authentication failed"));
+    }
+    validate_intent(&intent)?;
+    let root_id = root.identity();
+    if (root_id.volume_serial, hex(&root_id.file_id))
+        != (intent.root_volume, intent.root_file.clone())
+    {
+        return Err(storage("rewrite project identity changed"));
+    }
+    if current != intent.prior && current != intent.next {
+        return Err(storage("rewrite generation authority diverged"));
+    }
+    if intent.state == IntentState::Preparing {
+        for entry in &intent.entries {
+            cleanup_preparing_input(root_path, root, entry)?;
+        }
+        return remove_journal(root);
+    }
+    let authority_count = intent
+        .entries
+        .iter()
+        .filter(|entry| entry.class == EntryClass::GenerationAuthority)
+        .count();
+    if authority_count != 1 {
+        return Err(storage(
+            "rewrite journal must contain exactly one generation authority",
+        ));
+    }
+    let data = intent
+        .entries
+        .iter()
+        .filter(|entry| entry.class == EntryClass::Data)
+        .collect::<Vec<_>>();
+    for (index, entry) in data.iter().enumerate() {
+        install(root_path, root, entry)?;
+        let boundary = if index == 0 {
+            "rewrite.after_first_data_install"
+        } else if index + 1 == data.len() {
+            "rewrite.after_last_data_install"
+        } else {
+            "rewrite.after_middle_data_install"
+        };
+        crate::project_failpoint::hit(boundary, None, None, "REWRITE_DATA", false)?;
+    }
+    crate::project_failpoint::hit(
+        "rewrite.before_generation_authority",
+        None,
+        None,
+        "REWRITE_GENERATION",
+        false,
+    )?;
+    root.revalidate_named().map_err(storage)?;
+    let authority = intent
+        .entries
+        .iter()
+        .find(|entry| entry.class == EntryClass::GenerationAuthority)
+        .expect("count checked");
+    verify_generation_authority(root_path, root, authority, intent.next)?;
+    root.revalidate_named().map_err(storage)?;
+    install(root_path, root, authority)?;
+    crate::project_failpoint::hit(
+        "rewrite.after_generation_authority",
+        None,
+        None,
+        "REWRITE_GENERATION",
+        false,
+    )?;
+    remove_journal(root)
+}
+
+fn cleanup_preparing_input(
+    root_path: &Path,
+    root: &StableDirectory,
+    entry: &Entry,
+) -> Result<(), GfError> {
+    let (parent, temporary) = retained_parent_at(root_path, root, &entry.temporary)?;
+    let file = match parent.open_child_file(&temporary) {
+        Ok(file) => file,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
+        Err(error) => return Err(storage(error)),
+    };
+    let identity = graphforge_filesystem::file_identity(&file).map_err(storage)?;
+    if (identity.volume_serial, hex(&identity.file_id))
+        != (entry.temporary_volume, entry.temporary_file.clone())
+    {
+        return Err(storage("preparing rewrite temporary identity changed"));
+    }
+    drop(file);
+    parent
+        .unlink_child_if_identity(&temporary, identity)
+        .map_err(storage)?;
+    parent.sync().map_err(storage)
+}
+
+fn verify_generation_authority(
+    root_path: &Path,
+    root: &StableDirectory,
+    entry: &Entry,
+    next: GenerationPair,
+) -> Result<(), GfError> {
+    let (parent, target) = retained_parent_at(root_path, root, &entry.destination)?;
+    let (_, temporary) = retained_parent_at(root_path, root, &entry.temporary)?;
+    let file = match parent.open_child_file(&temporary) {
+        Ok(file) => file,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            parent.open_child_file(&target).map_err(storage)?
+        }
+        Err(error) => return Err(storage(error)),
+    };
+    let mut bytes = Vec::new();
+    file.take(4097).read_to_end(&mut bytes).map_err(storage)?;
+    if bytes.len() > 4096 {
+        return Err(storage("generation authority exceeds bound"));
+    }
+    let value: serde_json::Value = serde_json::from_slice(&bytes).map_err(storage)?;
+    if value
+        .get("topology_generation")
+        .and_then(serde_json::Value::as_u64)
+        != Some(next.topology)
+        || value
+            .get("search_generation")
+            .and_then(serde_json::Value::as_u64)
+            != Some(next.search)
+    {
+        return Err(storage(
+            "generation authority bytes do not encode journal next state",
+        ));
+    }
+    Ok(())
+}
+
+fn validate_intent(intent: &Intent) -> Result<(), GfError> {
+    let mut destinations = std::collections::HashSet::new();
+    let mut temporaries = std::collections::HashSet::new();
+    for entry in &intent.entries {
+        if !destinations.insert(&entry.destination) || !temporaries.insert(&entry.temporary) {
+            return Err(storage("rewrite journal contains duplicate paths"));
+        }
+        canonical_journal_path(&entry.destination)?;
+        canonical_journal_path(&entry.temporary)?;
+        if entry.destination == JOURNAL
+            || entry.destination == LOCK
+            || entry.temporary == JOURNAL
+            || entry.temporary == LOCK
+        {
+            return Err(storage("rewrite journal targets a reserved control"));
+        }
+    }
+    let authority = intent
+        .entries
+        .iter()
+        .filter(|entry| entry.class == EntryClass::GenerationAuthority)
+        .collect::<Vec<_>>();
+    if authority.len() != 1 || authority[0].destination != "topology/generation.json" {
+        return Err(storage("rewrite journal has invalid generation authority"));
+    }
+    if intent.next.topology < intent.prior.topology
+        || intent.next.search < intent.prior.search
+        || intent.next.topology > intent.prior.topology.saturating_add(1)
+        || intent.next.search > intent.prior.search.saturating_add(1)
+    {
+        return Err(storage("rewrite generation transition is not monotonic"));
+    }
+    if let Some(receipt) = &intent.auxiliary {
+        let entry = intent
+            .entries
+            .iter()
+            .find(|entry| entry.destination == receipt.path)
+            .ok_or_else(|| storage("auxiliary receipt path is not staged"))?;
+        if entry.class != EntryClass::Data
+            || entry.sha256 != receipt.digest
+            || entry.bytes != receipt.bytes
+            || receipt.kind.is_empty()
+            || receipt.schema_version == 0
+        {
+            return Err(storage(
+                "auxiliary receipt digest is not bound to staged bytes",
+            ));
+        }
+    }
+    Ok(())
+}
+
+fn canonical_journal_path(value: &str) -> Result<(), GfError> {
+    let path = Path::new(value);
+    if path.as_os_str().is_empty()
+        || path.is_absolute()
+        || path
+            .components()
+            .any(|component| !matches!(component, Component::Normal(_)))
+    {
+        return Err(storage("rewrite journal contains non-canonical path"));
+    }
+    Ok(())
+}
+
+pub(crate) fn recover(root: &Path) -> Result<(), GfError> {
+    let guard = acquire(root)?;
+    let current = crate::generation::read_generation_state_raw(root)?;
+    recover_locked(
+        guard.admission.root(),
+        &guard.directory,
+        GenerationPair {
+            topology: current.topology,
+            search: current.search,
+        },
+    )
+}
+
+pub(crate) fn commit(
+    batch: RewriteBatch,
+    root: &Path,
+    bump_topology: bool,
+    bump_search: bool,
+    auxiliary: Option<AuxiliaryReceipt>,
+) -> Result<GenerationPair, GfError> {
+    let guard = acquire(root)?;
+    guard.revalidate()?;
+    let prior_state = crate::generation::read_generation_state_raw(root)?;
+    let prior = GenerationPair {
+        topology: prior_state.topology,
+        search: prior_state.search,
+    };
+    recover_locked(root, &guard.directory, prior)?;
+    let recovered = crate::generation::read_generation_state_raw(root)?;
+    let prior = GenerationPair {
+        topology: recovered.topology,
+        search: recovered.search,
+    };
+    let next = GenerationPair {
+        topology: if bump_topology {
+            prior
+                .topology
+                .checked_add(1)
+                .ok_or_else(|| storage("topology generation counter overflow"))?
+        } else {
+            prior.topology
+        },
+        search: if bump_search {
+            prior
+                .search
+                .checked_add(1)
+                .ok_or_else(|| storage("search generation counter overflow"))?
+        } else {
+            prior.search
+        },
+    };
+    let generation_bytes = crate::generation::encode_generation_state(next.topology, next.search)?;
+    guard.revalidate()?;
+    let transaction = Uuid::now_v7().simple().to_string();
+    let root_identity = identity(root)?;
+    let mut entries = Vec::new();
+    let mut staged = batch.into_staged();
+    let generation_path = root.join("topology/generation.json");
+    std::fs::create_dir_all(generation_path.parent().expect("generation has parent"))
+        .map_err(storage)?;
+    let mut generation = tempfile::Builder::new()
+        .prefix("generation.json.")
+        .suffix(".tmp")
+        .tempfile_in(generation_path.parent().unwrap())
+        .map_err(storage)?;
+    generation
+        .write_all(&generation_bytes)
+        .and_then(|_| generation.as_file().sync_all())
+        .map_err(storage)?;
+    staged.push((generation, generation_path));
+    if staged.len() > MAX_ENTRIES {
+        return Err(storage("rewrite batch exceeds entry bound"));
+    }
+    let last = staged.len().saturating_sub(1);
+    for (ordinal, (temp, destination)) in staged.iter().enumerate() {
+        let relative = canonical_relative(root, &destination)?;
+        let parent = destination
+            .parent()
+            .ok_or_else(|| storage("rewrite destination has no parent"))?;
+        let parent_identity = identity(parent)?;
+        temp.as_file().sync_all().map_err(storage)?;
+        let original = graphforge_filesystem::file_identity(temp.as_file()).map_err(storage)?;
+        let durable = temp.path().to_path_buf();
+        if graphforge_filesystem::path_identity(&durable).map_err(storage)? != original {
+            return Err(storage("rewrite temporary identity changed before intent"));
+        }
+        sync_dir(parent)?;
+        let (bytes, sha256) = hash_reader(temp.as_file().try_clone().map_err(storage)?)?;
+        let temp_relative = canonical_relative(root, &durable)?;
+        entries.push(Entry {
+            class: if ordinal == last {
+                EntryClass::GenerationAuthority
+            } else {
+                EntryClass::Data
+            },
+            destination: relative,
+            temporary: temp_relative,
+            parent_volume: parent_identity.0,
+            parent_file: parent_identity.1,
+            bytes,
+            sha256,
+            temporary_volume: original.volume_serial,
+            temporary_file: hex(&original.file_id),
+        });
+    }
+    let mut intent = Intent {
+        version: 1,
+        state: IntentState::Preparing,
+        transaction,
+        root_volume: root_identity.0,
+        root_file: root_identity.1,
+        prior,
+        next,
+        auxiliary,
+        entries,
+        checksum: String::new(),
+    };
+    validate_intent(&intent)?;
+    intent.checksum = checksum(&intent)?;
+    crate::project_failpoint::hit(
+        "rewrite.before_intent",
+        None,
+        None,
+        "REWRITE_BEFORE_INTENT",
+        false,
+    )?;
+    publish_journal(&guard.directory, &intent)?;
+    // Before intent, NamedTempFile owns cleanup. Preparing intent permits
+    // identity-safe abort if disarming any handle fails. Only after every temp
+    // is intentionally retained do we publish the durable roll-forward state.
+    for (index, (temp, _)) in staged.into_iter().enumerate() {
+        temp.into_temp_path()
+            .keep()
+            .map_err(|error| storage(error.error))?;
+        if index == 0 {
+            crate::project_failpoint::hit(
+                "rewrite.after_preparing_disarm",
+                None,
+                None,
+                "REWRITE_PREPARING",
+                false,
+            )?;
+        }
+    }
+    intent.state = IntentState::Durable;
+    intent.checksum = checksum(&intent)?;
+    publish_journal(&guard.directory, &intent)?;
+    #[cfg(test)]
+    if FAIL_AFTER_DURABLE_INTENT.swap(false, std::sync::atomic::Ordering::SeqCst) {
+        return Err(storage("injected ordinary error after durable intent"));
+    }
+    crate::project_failpoint::hit(
+        "rewrite.after_durable_intent",
+        None,
+        None,
+        "REWRITE_INTENT",
+        false,
+    )?;
+    // Use the same classified replay path as crash recovery so authority order
+    // cannot drift between the initial commit and roll-forward.
+    guard.revalidate()?;
+    recover_locked(root, &guard.directory, prior)?;
+    guard.revalidate()?;
+    crate::io_stats::record_rewrite_commit();
+    Ok(next)
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use arrow::array::{Int64Array, RecordBatch};
+    use arrow::datatypes::{DataType, Field, Schema};
+    use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+    use tempfile::TempDir;
+
+    use super::*;
+
+    fn entry(destination: &str, class: EntryClass) -> Entry {
+        Entry {
+            class,
+            destination: destination.to_owned(),
+            temporary: format!("topology/{destination}.abc.tmp"),
+            parent_volume: 1,
+            parent_file: "01".to_owned(),
+            bytes: 2,
+            sha256: "aa".repeat(32),
+            temporary_volume: 1,
+            temporary_file: "02".to_owned(),
+        }
+    }
+
+    fn intent() -> Intent {
+        Intent {
+            version: 1,
+            state: IntentState::Durable,
+            transaction: "tx".to_owned(),
+            root_volume: 1,
+            root_file: "01".to_owned(),
+            prior: GenerationPair {
+                topology: 4,
+                search: 3,
+            },
+            next: GenerationPair {
+                topology: 5,
+                search: 4,
+            },
+            auxiliary: None,
+            entries: vec![
+                entry("topology/nodes.parquet", EntryClass::Data),
+                entry("topology/generation.json", EntryClass::GenerationAuthority),
+            ],
+            checksum: String::new(),
+        }
+    }
+
+    #[test]
+    fn intent_validation_rejects_duplicate_reserved_and_nonmonotonic_authority() {
+        assert!(validate_intent(&intent()).is_ok());
+        let mut duplicate = intent();
+        duplicate.entries[0].temporary = duplicate.entries[1].temporary.clone();
+        assert!(validate_intent(&duplicate).is_err());
+        let mut reserved = intent();
+        reserved.entries[0].destination = JOURNAL.to_owned();
+        assert!(validate_intent(&reserved).is_err());
+        let mut backwards = intent();
+        backwards.next.topology = 3;
+        assert!(validate_intent(&backwards).is_err());
+        let mut wrong_authority = intent();
+        wrong_authority.entries[1].destination = "topology/other.json".to_owned();
+        assert!(validate_intent(&wrong_authority).is_err());
+    }
+
+    #[test]
+    fn auxiliary_receipt_must_name_and_digest_an_exact_staged_entry() {
+        let mut valid = intent();
+        valid.auxiliary = Some(AuxiliaryReceipt {
+            kind: "uuid-membership/v3".to_owned(),
+            schema_version: 3,
+            path: valid.entries[0].destination.clone(),
+            digest: valid.entries[0].sha256.clone(),
+            bytes: valid.entries[0].bytes,
+        });
+        assert!(validate_intent(&valid).is_ok());
+        valid.auxiliary.as_mut().unwrap().digest = "00".repeat(32);
+        assert!(validate_intent(&valid).is_err());
+    }
+
+    #[test]
+    fn checksum_authenticates_every_recovery_control() {
+        let mut value = intent();
+        value.checksum = checksum(&value).unwrap();
+        assert_eq!(checksum(&value).unwrap(), value.checksum);
+        value.entries[0].bytes += 1;
+        assert_ne!(checksum(&value).unwrap(), value.checksum);
+    }
+
+    #[test]
+    fn ordinary_error_after_durable_intent_rolls_forward_on_double_reopen() {
+        let root = TempDir::new().unwrap();
+        let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)]));
+        let batch = RecordBatch::try_new(
+            Arc::clone(&schema),
+            vec![Arc::new(Int64Array::from(vec![7_i64]))],
+        )
+        .unwrap();
+        let destination = root.path().join("topology/nodes.parquet");
+        let mut rewrite = RewriteBatch::new();
+        rewrite.stage(&destination, schema, &batch).unwrap();
+        FAIL_AFTER_DURABLE_INTENT.store(true, std::sync::atomic::Ordering::SeqCst);
+        assert!(commit(rewrite, root.path(), true, true, None).is_err());
+        assert!(root.path().join(JOURNAL).is_file());
+        assert_eq!(
+            crate::generation::read_topology_generation(root.path()).unwrap(),
+            1
+        );
+        assert_eq!(
+            crate::generation::read_topology_generation(root.path()).unwrap(),
+            1
+        );
+        assert!(destination.is_file());
+        assert!(!root.path().join(JOURNAL).exists());
+    }
+
+    #[test]
+    fn subprocess_crash_matrix_preserves_generation_last_and_reopens_idempotently() {
+        const CHILD_ROOT: &str = "GRAPHFORGE_REWRITE_CHILD_ROOT";
+        let destinations = [
+            "topology/nodes.parquet",
+            "properties/Person.parquet",
+            "edge_properties/KNOWS.parquet",
+        ];
+        if let Ok(root) = std::env::var(CHILD_ROOT) {
+            let root = Path::new(&root);
+            let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)]));
+            let mut rewrite = RewriteBatch::new();
+            for (index, relative) in destinations.iter().enumerate() {
+                let batch = RecordBatch::try_new(
+                    Arc::clone(&schema),
+                    vec![Arc::new(Int64Array::from(vec![index as i64 + 10]))],
+                )
+                .unwrap();
+                rewrite
+                    .stage(&root.join(relative), Arc::clone(&schema), &batch)
+                    .unwrap();
+            }
+            let _ = commit(rewrite, root, true, true, None);
+            panic!("child failpoint did not terminate the process");
+        }
+
+        let phases = [
+            ("rewrite.before_intent", false),
+            ("rewrite.after_preparing_disarm", false),
+            ("rewrite.after_durable_intent", true),
+            ("rewrite.after_first_data_install", true),
+            ("rewrite.after_middle_data_install", true),
+            ("rewrite.after_last_data_install", true),
+            ("rewrite.before_generation_authority", true),
+            ("rewrite.after_generation_authority", true),
+        ];
+        for (phase, committed) in phases {
+            let root = TempDir::new().unwrap();
+            let status = std::process::Command::new(std::env::current_exe().unwrap())
+                .arg("--exact")
+                .arg("durable_rewrite::tests::subprocess_crash_matrix_preserves_generation_last_and_reopens_idempotently")
+                .arg("--nocapture")
+                .env(CHILD_ROOT, root.path())
+                .env("GRAPHFORGE_PROJECT_FAILPOINTS", "graphforge-internal-subprocess-v1")
+                .env("GRAPHFORGE_PROJECT_FAILPOINT", phase)
+                .status()
+                .unwrap();
+            assert_eq!(
+                status.code(),
+                Some(crate::project_failpoint::exit_code()),
+                "{phase}"
+            );
+
+            let first = crate::generation::read_topology_generation(root.path()).unwrap();
+            crate::staging::remove_stale_temps(root.path()).unwrap();
+            let second = crate::generation::read_topology_generation(root.path()).unwrap();
+            assert_eq!(
+                (first, second),
+                if committed { (1, 1) } else { (0, 0) },
+                "{phase}"
+            );
+            for (index, relative) in destinations.iter().enumerate() {
+                let path = root.path().join(relative);
+                assert_eq!(path.exists(), committed, "{phase}: {relative}");
+                if committed {
+                    let mut reader =
+                        ParquetRecordBatchReaderBuilder::try_new(File::open(path).unwrap())
+                            .unwrap()
+                            .build()
+                            .unwrap();
+                    let batch = reader.next().unwrap().unwrap();
+                    let values = batch
+                        .column(0)
+                        .as_any()
+                        .downcast_ref::<Int64Array>()
+                        .unwrap();
+                    assert_eq!(values.value(0), index as i64 + 10, "{phase}: {relative}");
+                    assert!(reader.next().is_none(), "{phase}: duplicate payload");
+                }
+            }
+            assert!(!root.path().join(JOURNAL).exists(), "{phase}");
+            for relative in ["topology", "properties", "edge_properties"] {
+                assert!(
+                    std::fs::read_dir(root.path().join(relative))
+                        .unwrap()
+                        .filter_map(Result::ok)
+                        .all(|entry| !entry.file_name().to_string_lossy().ends_with(".tmp")),
+                    "{phase}: temp cleanup"
+                );
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+static FAIL_AFTER_DURABLE_INTENT: std::sync::atomic::AtomicBool =
+    std::sync::atomic::AtomicBool::new(false);
diff --git a/crates/graphforge-storage/src/generation.rs b/crates/graphforge-storage/src/generation.rs
index c564ad2e..97b53745 100644
--- a/crates/graphforge-storage/src/generation.rs
+++ b/crates/graphforge-storage/src/generation.rs
@@ -16,20 +16,13 @@
 //!
 //! # Crash-safety invariant
 //!
-//! [`commit_topology_aware`] bumps the counter **strictly before** the first
-//! rename of the staged batch. A crash after the bump but before (or during)
-//! the commit leaves the counter advanced over an unchanged or partially
-//! renamed topology — any existing index now merely *looks* stale and is
-//! rebuilt, costing one spurious rebuild. The reverse order would be unsound:
-//! a crash between commit and bump would leave new topology under the old
-//! counter, making a stale index look **fresh** and silently serving wrong
-//! traversals. Spurious bumps are safe; missed bumps are not.
-//!
-//! Multi-process writers can lose a bump (read-increment-rename is not
-//! cross-process atomic); this matches the consistency envelope of every
-//! Parquet rewrite in this embedded engine (see [`crate::staging`]).
+//! [`commit_topology_aware`] serializes writers, durably authenticates every
+//! retained replacement in a bounded intent journal, rolls data files forward,
+//! and publishes this counter last as the explicit authority switch. A crash at
+//! any barrier is replayed idempotently before the generation is read. Thus a
+//! generation can never describe a prefix of the intended topology and a
+//! completed topology can never remain authoritative under its prior counter.
 
-use std::io::Write;
 use std::path::{Path, PathBuf};
 
 use graphforge_core::GfError;
@@ -42,9 +35,9 @@ const GENERATION_KEY: &str = "topology_generation";
 const SEARCH_GENERATION_KEY: &str = "search_generation";
 
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
-struct GenerationState {
-    topology: u64,
-    search: u64,
+pub(crate) struct GenerationState {
+    pub(crate) topology: u64,
+    pub(crate) search: u64,
 }
 
 fn storage_err(e: impl std::fmt::Display) -> GfError {
@@ -87,7 +80,7 @@ pub fn read_search_generation(project_dir: &Path) -> Result<u64, GfError> {
     Ok(read_generation_state(project_dir)?.search)
 }
 
-fn read_generation_state(project_dir: &Path) -> Result<GenerationState, GfError> {
+pub(crate) fn read_generation_state_raw(project_dir: &Path) -> Result<GenerationState, GfError> {
     let path = generation_path(project_dir);
     let contents = match std::fs::read_to_string(&path) {
         Ok(c) => c,
@@ -124,6 +117,19 @@ fn read_generation_state(project_dir: &Path) -> Result<GenerationState, GfError>
     Ok(GenerationState { topology, search })
 }
 
+pub(crate) fn encode_generation_state(topology: u64, search: u64) -> Result<Vec<u8>, GfError> {
+    serde_json::to_vec(&serde_json::json!({
+        GENERATION_KEY: topology,
+        SEARCH_GENERATION_KEY: search,
+    }))
+    .map_err(storage_err)
+}
+
+fn read_generation_state(project_dir: &Path) -> Result<GenerationState, GfError> {
+    crate::durable_rewrite::recover(project_dir)?;
+    read_generation_state_raw(project_dir)
+}
+
 /// Atomically persist `current + 1` (sibling temp + rename) and return the
 /// new value. Creates `topology/` if needed.
 ///
@@ -131,7 +137,10 @@ fn read_generation_state(project_dir: &Path) -> Result<GenerationState, GfError>
 /// Returns [`GfError::Storage`] if the current value cannot be read (corrupt
 /// file) or on I/O failure; on failure the prior file is untouched.
 pub fn bump_topology_generation(project_dir: &Path) -> Result<u64, GfError> {
-    Ok(bump_generations(project_dir, true, false)?.topology)
+    Ok(
+        crate::durable_rewrite::commit(RewriteBatch::new(), project_dir, true, false, None)?
+            .topology,
+    )
 }
 
 /// Atomically advance and persist the graph-native search generation.
@@ -140,57 +149,7 @@ pub fn bump_topology_generation(project_dir: &Path) -> Result<u64, GfError> {
 /// Returns [`GfError::Storage`] if the existing generation is corrupt or the
 /// replacement cannot be persisted.
 pub fn bump_search_generation(project_dir: &Path) -> Result<u64, GfError> {
-    Ok(bump_generations(project_dir, false, true)?.search)
-}
-
-fn bump_generations(
-    project_dir: &Path,
-    bump_topology: bool,
-    bump_search: bool,
-) -> Result<GenerationState, GfError> {
-    let mut next = read_generation_state(project_dir)?;
-    if bump_topology {
-        next.topology = next
-            .topology
-            .checked_add(1)
-            .ok_or_else(|| GfError::Storage("topology generation counter overflow".to_owned()))?;
-    }
-    if bump_search {
-        next.search = next
-            .search
-            .checked_add(1)
-            .ok_or_else(|| GfError::Storage("search generation counter overflow".to_owned()))?;
-    }
-    let path = generation_path(project_dir);
-    let parent = path.parent().expect("generation path always has a parent");
-    std::fs::create_dir_all(parent).map_err(storage_err)?;
-    let mut tmp = tempfile::Builder::new()
-        .prefix("generation.json.")
-        .suffix(".tmp")
-        .tempfile_in(parent)
-        .map_err(storage_err)?;
-    let body = serde_json::json!({
-        GENERATION_KEY: next.topology,
-        SEARCH_GENERATION_KEY: next.search,
-    })
-    .to_string();
-    tmp.write_all(body.as_bytes()).map_err(storage_err)?;
-    tmp.as_file().sync_all().map_err(storage_err)?;
-    tmp.persist(&path).map_err(|e| storage_err(e.error))?;
-    sync_directory(parent)?;
-    Ok(next)
-}
-
-#[cfg(unix)]
-fn sync_directory(path: &Path) -> Result<(), GfError> {
-    std::fs::File::open(path)
-        .and_then(|directory| directory.sync_all())
-        .map_err(storage_err)
-}
-
-#[cfg(not(unix))]
-fn sync_directory(_path: &Path) -> Result<(), GfError> {
-    Ok(())
+    Ok(crate::durable_rewrite::commit(RewriteBatch::new(), project_dir, false, true, None)?.search)
 }
 
 /// Whether any staged destination in `staged` rewrites topology:
@@ -220,8 +179,8 @@ pub fn touches_search_source(staged: &RewriteBatch, project_dir: &Path) -> bool
         .any(|path| path == nodes || path.starts_with(&properties))
 }
 
-/// Commit `staged`, bumping each affected generation **first** (see the module
-/// docs for why bump-before-commit is the only sound order). Edge topology
+/// Durably commit `staged`, publishing each affected generation **last** (see
+/// the module-level crash invariant). Edge topology
 /// advances only topology; node topology advances topology and search; node
 /// properties advance only search.
 ///
@@ -230,23 +189,28 @@ pub fn touches_search_source(staged: &RewriteBatch, project_dir: &Path) -> bool
 /// than re-reading the counter (which a concurrent bump could have advanced).
 ///
 /// # Errors
-/// Returns [`GfError::Storage`] on bump or rename failure. A bump followed by
-/// a failed commit leaves the counter advanced — safe (the index reads as
-/// stale), see the crash-safety invariant.
+/// Returns [`GfError::Storage`] on admission, journal, authentication, replay,
+/// or namespace-durability failure. A durable intent is always rolled forward
+/// on retry/reopen; failures before intent preserve the prior authority.
 pub fn commit_topology_aware(
     staged: RewriteBatch,
     project_dir: &Path,
+) -> Result<Option<u64>, GfError> {
+    commit_topology_aware_with_auxiliary(staged, project_dir, None)
+}
+
+/// Commit a rewrite with an authenticated typed auxiliary receipt bound to a
+/// staged destination in the same generation-last transaction.
+pub fn commit_topology_aware_with_auxiliary(
+    staged: RewriteBatch,
+    project_dir: &Path,
+    auxiliary: Option<crate::AuxiliaryReceipt>,
 ) -> Result<Option<u64>, GfError> {
     let topology = touches_topology(&staged, project_dir);
     let search = touches_search_source(&staged, project_dir);
-    let bumped = if topology || search {
-        let generations = bump_generations(project_dir, topology, search)?;
-        topology.then_some(generations.topology)
-    } else {
-        None
-    };
-    staged.commit()?;
-    Ok(bumped)
+    let generations =
+        crate::durable_rewrite::commit(staged, project_dir, topology, search, auxiliary)?;
+    Ok(topology.then_some(generations.topology))
 }
 
 // ---------------------------------------------------------------------------
@@ -329,6 +293,20 @@ mod tests {
         }
     }
 
+    #[test]
+    fn corrupt_rewrite_journal_fails_closed_without_changing_authority() {
+        let dir = TempDir::new().unwrap();
+        assert_eq!(bump_topology_generation(dir.path()).unwrap(), 1);
+        std::fs::write(
+            dir.path().join(".graphforge-rewrite-v1.json"),
+            br#"{"version":1,"checksum":"forged"}"#,
+        )
+        .unwrap();
+        let before = std::fs::read(generation_path(dir.path())).unwrap();
+        assert!(read_topology_generation(dir.path()).is_err());
+        assert_eq!(std::fs::read(generation_path(dir.path())).unwrap(), before);
+    }
+
     #[test]
     fn touches_topology_matrix() {
         let dir = TempDir::new().unwrap();
@@ -366,6 +344,10 @@ mod tests {
         commit_topology_aware(staged, dir.path()).unwrap();
         assert_eq!(read_topology_generation(dir.path()).unwrap(), 0);
         assert_eq!(read_search_generation(dir.path()).unwrap(), 1);
+        // Repeated reopen/recovery is idempotent and consumes the intent.
+        assert_eq!(read_topology_generation(dir.path()).unwrap(), 0);
+        assert_eq!(read_search_generation(dir.path()).unwrap(), 1);
+        assert!(!dir.path().join(".graphforge-rewrite-v1.json").exists());
 
         // Mixed batch staging topology: exactly one bump.
         let staged = staged_for(
diff --git a/crates/graphforge-storage/src/lib.rs b/crates/graphforge-storage/src/lib.rs
index 246e9fb0..30e15475 100644
--- a/crates/graphforge-storage/src/lib.rs
+++ b/crates/graphforge-storage/src/lib.rs
@@ -10,7 +10,9 @@
 //! - [`search_manifest`] / [`search_publication`] — shared search search freshness and atomic publication
 #![forbid(unsafe_code)]
 
+mod durable_rewrite;
 mod file_lock;
+pub use durable_rewrite::AuxiliaryReceipt;
 #[doc(hidden)]
 pub mod filesystem_admission;
 
@@ -19,7 +21,8 @@ pub mod adjacency_delta;
 
 pub mod generation;
 pub use generation::{
-    commit_topology_aware, read_search_generation, read_topology_generation, touches_search_source,
+    commit_topology_aware, commit_topology_aware_with_auxiliary, read_search_generation,
+    read_topology_generation, touches_search_source,
 };
 
 pub mod graph_projection;
diff --git a/crates/graphforge-storage/src/staging.rs b/crates/graphforge-storage/src/staging.rs
index 682654dd..0a44db13 100644
--- a/crates/graphforge-storage/src/staging.rs
+++ b/crates/graphforge-storage/src/staging.rs
@@ -20,13 +20,11 @@
 //! `rename` is atomic) with a `.tmp` extension — invisible to every reader in
 //! this crate, which match on the `parquet` extension or exact file names.
 //!
-//! Atomicity envelope, stated honestly: failures during the **stage** phase
-//! (the realistic class — allocation, encode, ENOSPC while writing data) are
-//! all-or-nothing. A failure during the **commit** phase (rename — rare:
-//! permissions, exotic filesystems) can apply a prefix of the batch; the
-//! insertion-order rules bound that prefix to a consistent graph (at worst
-//! orphaned-but-unreferenced rows, never dangling references). Durability
-//! (fsync) is out of scope for this non-production engine.
+//! [`crate::generation::commit_topology_aware`] upgrades a batch into a durable
+//! transaction: it records authenticated, deterministic recovery inputs before
+//! replacing any destination and publishes generation authority last.  The
+//! plain [`commit`](RewriteBatch::commit) remains for tests and explicitly
+//! ephemeral callers; persistent graph mutations use the topology-aware path.
 
 use std::path::{Path, PathBuf};
 
@@ -59,6 +57,9 @@ fn pq_err(e: impl std::fmt::Display) -> GfError {
 /// Returns [`GfError::Storage`] when a graph-owned directory cannot be read or
 /// a recognized stale temp cannot be removed.
 pub fn remove_stale_temps(project_dir: &Path) -> Result<usize, GfError> {
+    // Recovery owns temp-looking durable inputs after intent. It must run
+    // before the stale-temp sweep can remove any graph-owned file.
+    let _ = crate::generation::read_topology_generation(project_dir)?;
     let mut removed = 0;
     for relative in STAGED_TEMP_DIRS {
         removed += remove_stale_temps_under(&project_dir.join(relative))?;
@@ -292,6 +293,10 @@ impl RewriteBatch {
     pub fn is_empty(&self) -> bool {
         self.staged.is_empty()
     }
+
+    pub(crate) fn into_staged(self) -> Vec<(NamedTempFile, PathBuf)> {
+        self.staged
+    }
 }
 
 /// Parquet row-group size for all staged files. Smaller than the 1 M-row
diff --git a/crates/graphforge-storage/src/writer.rs b/crates/graphforge-storage/src/writer.rs
index ea9b5050..5ed8efcc 100644
--- a/crates/graphforge-storage/src/writer.rs
+++ b/crates/graphforge-storage/src/writer.rs
@@ -3785,7 +3785,7 @@ pub fn set_edge_properties_rewrite(
 ) -> Result<u64, GfError> {
     let mut staged = RewriteBatch::new();
     let touched = stage_set_edge_properties(&mut staged, dir, rel_stem, updates)?;
-    staged.commit()?;
+    crate::generation::commit_topology_aware(staged, dir)?;
     Ok(touched)
 }
 
@@ -3802,7 +3802,7 @@ pub fn remove_edge_properties(
 ) -> Result<u64, GfError> {
     let mut staged = RewriteBatch::new();
     let touched = stage_remove_edge_properties(&mut staged, dir, rel_stem, removals)?;
-    staged.commit()?;
+    crate::generation::commit_topology_aware(staged, dir)?;
     Ok(touched)
 }
 

From f03864a26efb2a2fc62ae4a398d813143cef9d74 Mon Sep 17 00:00:00 2001
From: David Spencer <1526975+DecisionNerd@users.noreply.github.com>
Date: Tue, 25 Aug 2026 10:00:15 -0600
Subject: [PATCH 2/6] fix(storage): harden rewrite recovery identities

---
 crates/graphforge-filesystem/src/lib.rs       |  90 +++--
 .../graphforge-storage/src/durable_rewrite.rs | 372 ++++++++++++++----
 docs/adr/0013-project-generation-protocol.md  |  69 +++-
 .../book/architecture/concurrency-recovery.md |  33 ++
 docs/book/architecture/storage.md             |  32 ++
 5 files changed, 491 insertions(+), 105 deletions(-)

diff --git a/crates/graphforge-filesystem/src/lib.rs b/crates/graphforge-filesystem/src/lib.rs
index 2f127d72..0f86ac6c 100644
--- a/crates/graphforge-filesystem/src/lib.rs
+++ b/crates/graphforge-filesystem/src/lib.rs
@@ -365,16 +365,22 @@ fn stable_open_replaceable_child_file(
 }
 
 #[cfg(unix)]
-fn stable_open_or_create_child_file(parent: &File, _path: &Path, name: &OsStr) -> io::Result<File> {
-    use rustix::fs::{Mode, OFlags};
-    rustix::fs::openat(
-        parent,
-        name,
-        OFlags::RDWR | OFlags::CREATE | OFlags::NOFOLLOW | OFlags::NONBLOCK | OFlags::CLOEXEC,
-        Mode::from_bits_truncate(0o600),
-    )
-    .map(File::from)
-    .map_err(io::Error::from)
+fn stable_open_or_create_child_file(parent: &File, path: &Path, name: &OsStr) -> io::Result<File> {
+    match stable_open_child_file(parent, path, name, true) {
+        Ok(file) => Ok(file),
+        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
+            use rustix::fs::{Mode, OFlags};
+            rustix::fs::openat(
+                parent,
+                name,
+                OFlags::RDWR | OFlags::NOFOLLOW | OFlags::NONBLOCK | OFlags::CLOEXEC,
+                Mode::empty(),
+            )
+            .map(File::from)
+            .map_err(io::Error::from)
+        }
+        Err(error) => Err(error),
+    }
 }
 
 #[cfg(unix)]
@@ -535,22 +541,23 @@ fn stable_open_replaceable_child_file(
 }
 
 #[cfg(windows)]
-fn stable_open_or_create_child_file(
-    _parent: &File,
-    path: &Path,
-    _name: &OsStr,
-) -> io::Result<File> {
-    use std::os::windows::fs::OpenOptionsExt as _;
-    const FILE_SHARE_READ: u32 = 0x0000_0001;
-    const FILE_SHARE_WRITE: u32 = 0x0000_0002;
-    const FILE_FLAG_OPEN_REPARSE_POINT: u32 = 0x0020_0000;
-    std::fs::OpenOptions::new()
-        .read(true)
-        .write(true)
-        .create(true)
-        .share_mode(FILE_SHARE_READ | FILE_SHARE_WRITE)
-        .custom_flags(FILE_FLAG_OPEN_REPARSE_POINT)
-        .open(path)
+fn stable_open_or_create_child_file(parent: &File, path: &Path, name: &OsStr) -> io::Result<File> {
+    match stable_open_child_file(parent, path, name, true) {
+        Ok(file) => Ok(file),
+        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
+            use std::os::windows::fs::OpenOptionsExt as _;
+            const FILE_SHARE_READ: u32 = 0x0000_0001;
+            const FILE_SHARE_WRITE: u32 = 0x0000_0002;
+            const FILE_FLAG_OPEN_REPARSE_POINT: u32 = 0x0020_0000;
+            std::fs::OpenOptions::new()
+                .read(true)
+                .write(true)
+                .share_mode(FILE_SHARE_READ | FILE_SHARE_WRITE)
+                .custom_flags(FILE_FLAG_OPEN_REPARSE_POINT)
+                .open(path)
+        }
+        Err(error) => Err(error),
+    }
 }
 
 #[cfg(windows)]
@@ -2053,6 +2060,37 @@ mod tests {
         }
     }
 
+    #[test]
+    fn concurrent_open_or_create_has_one_stable_named_identity() {
+        let root = tempfile::tempdir().unwrap();
+        let root = std::sync::Arc::new(root);
+        let barrier = std::sync::Arc::new(std::sync::Barrier::new(9));
+        let workers = (0..8)
+            .map(|_| {
+                let root = std::sync::Arc::clone(&root);
+                let barrier = std::sync::Arc::clone(&barrier);
+                std::thread::spawn(move || {
+                    let directory = StableDirectory::open(root.path()).unwrap();
+                    barrier.wait();
+                    let file = directory
+                        .open_or_create_child_file(OsStr::new("lifecycle.lock"))
+                        .unwrap();
+                    file_identity(&file).unwrap()
+                })
+            })
+            .collect::<Vec<_>>();
+        barrier.wait();
+        let identities = workers
+            .into_iter()
+            .map(|worker| worker.join().unwrap())
+            .collect::<Vec<_>>();
+        assert!(identities.windows(2).all(|pair| pair[0] == pair[1]));
+        assert_eq!(
+            file_link_count(&File::open(root.path().join("lifecycle.lock")).unwrap()).unwrap(),
+            1
+        );
+    }
+
     #[cfg(unix)]
     #[test]
     fn peerless_fifo_child() {
diff --git a/crates/graphforge-storage/src/durable_rewrite.rs b/crates/graphforge-storage/src/durable_rewrite.rs
index 8d5ad31f..ba8e763e 100644
--- a/crates/graphforge-storage/src/durable_rewrite.rs
+++ b/crates/graphforge-storage/src/durable_rewrite.rs
@@ -5,6 +5,7 @@
 //! final authority switch.  Journal paths are bounded, canonical relative
 //! paths and every recovery input is authenticated before it is used.
 
+use std::fmt::Write as _;
 use std::fs::File;
 use std::io::{Read, Seek, Write};
 use std::path::{Component, Path};
@@ -80,6 +81,15 @@ struct Entry {
     sha256: String,
     temporary_volume: u64,
     temporary_file: String,
+    prior_destination: Option<AuthenticatedFile>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct AuthenticatedFile {
+    volume: u64,
+    file: String,
+    bytes: u64,
+    sha256: String,
 }
 
 #[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
@@ -93,7 +103,13 @@ fn storage(error: impl std::fmt::Display) -> GfError {
     GfError::Storage(error.to_string())
 }
 fn hex(bytes: &[u8]) -> String {
-    bytes.iter().map(|b| format!("{b:02x}")).collect()
+    bytes.iter().fold(
+        String::with_capacity(bytes.len().saturating_mul(2)),
+        |mut encoded, byte| {
+            write!(encoded, "{byte:02x}").expect("writing to a String cannot fail");
+            encoded
+        },
+    )
 }
 
 fn canonical_relative(root: &Path, path: &Path) -> Result<String, GfError> {
@@ -115,16 +131,11 @@ fn canonical_relative(root: &Path, path: &Path) -> Result<String, GfError> {
         .ok_or_else(|| storage("rewrite destination is not UTF-8"))
 }
 
-fn identity(path: &Path) -> Result<(u64, String), GfError> {
-    let id = graphforge_filesystem::path_identity(path).map_err(storage)?;
-    Ok((id.volume_serial, hex(&id.file_id)))
-}
-
 fn hash_reader(mut file: File) -> Result<(u64, String), GfError> {
     file.rewind().map_err(storage)?;
     let mut hash = Sha256::new();
     let mut bytes = 0_u64;
-    let mut buffer = [0_u8; 1024 * 1024];
+    let mut buffer = vec![0_u8; 1024 * 1024].into_boxed_slice();
     loop {
         let count = file.read(&mut buffer).map_err(storage)?;
         if count == 0 {
@@ -138,16 +149,15 @@ fn hash_reader(mut file: File) -> Result<(u64, String), GfError> {
     Ok((bytes, hex(&hash.finalize())))
 }
 
-fn sync_dir(path: &Path) -> Result<(), GfError> {
-    #[cfg(unix)]
-    {
-        File::open(path).and_then(|f| f.sync_all()).map_err(storage)
-    }
-    #[cfg(not(unix))]
-    {
-        let _ = path;
-        Ok(())
-    }
+fn authenticated_file(file: &File) -> Result<AuthenticatedFile, GfError> {
+    let identity = graphforge_filesystem::file_identity(file).map_err(storage)?;
+    let (bytes, sha256) = hash_reader(file.try_clone().map_err(storage)?)?;
+    Ok(AuthenticatedFile {
+        volume: identity.volume_serial,
+        file: hex(&identity.file_id),
+        bytes,
+        sha256,
+    })
 }
 
 struct RewriteGuard {
@@ -172,13 +182,18 @@ fn acquire(root: &Path) -> Result<RewriteGuard, GfError> {
         ProjectLifecycleMode::Ephemeral,
         ProjectRootRequirement::Existing,
     )?;
-    let directory = StableDirectory::open(root).map_err(storage)?;
+    let directory = StableDirectory::open(root)
+        .map_err(|error| storage(format!("rewrite root open failed: {error}")))?;
     let lock = directory
         .open_or_create_child_file(std::ffi::OsStr::new(LOCK))
-        .map_err(storage)?;
-    lock.sync_all().map_err(storage)?;
-    directory.sync().map_err(storage)?;
-    crate::file_lock::lock_exclusive(&lock).map_err(storage)?;
+        .map_err(|error| storage(format!("rewrite lock open failed: {error}")))?;
+    lock.sync_all()
+        .map_err(|error| storage(format!("rewrite lock sync failed: {error}")))?;
+    directory
+        .sync()
+        .map_err(|error| storage(format!("rewrite root sync failed: {error}")))?;
+    crate::file_lock::lock_exclusive(&lock)
+        .map_err(|error| storage(format!("rewrite lock acquisition failed: {error}")))?;
     admission.revalidate_identity()?;
     let lifecycle_identity = graphforge_filesystem::file_identity(&lock).map_err(storage)?;
     Ok(RewriteGuard {
@@ -233,6 +248,12 @@ fn checksum(intent: &Intent) -> Result<String, GfError> {
                 sha256: e.sha256.clone(),
                 temporary_volume: e.temporary_volume,
                 temporary_file: e.temporary_file.clone(),
+                prior_destination: e.prior_destination.as_ref().map(|prior| AuthenticatedFile {
+                    volume: prior.volume,
+                    file: prior.file.clone(),
+                    bytes: prior.bytes,
+                    sha256: prior.sha256.clone(),
+                }),
             })
             .collect(),
     };
@@ -249,7 +270,7 @@ fn publish_journal(root: &StableDirectory, intent: &Intent) -> Result<(), GfErro
         .create_replaceable_child_file(std::ffi::OsStr::new(&name))
         .map_err(storage)?;
     temp.write_all(&bytes)
-        .and_then(|_| temp.sync_all())
+        .and_then(|()| temp.sync_all())
         .map_err(storage)?;
     let expected = graphforge_filesystem::file_identity(&temp).map_err(storage)?;
     root.replace_child(
@@ -263,44 +284,87 @@ fn publish_journal(root: &StableDirectory, intent: &Intent) -> Result<(), GfErro
 
 fn install(root_path: &Path, root: &StableDirectory, entry: &Entry) -> Result<(), GfError> {
     let (parent, target) = retained_parent_at(root_path, root, &entry.destination)?;
-    let (_, temporary) = retained_parent_at(root_path, root, &entry.temporary)?;
+    let (temporary_parent, temporary) = retained_parent_at(root_path, root, &entry.temporary)?;
     let parent_id = parent.identity();
     if (parent_id.volume_serial, hex(&parent_id.file_id))
         != (entry.parent_volume, entry.parent_file.clone())
     {
         return Err(storage("rewrite parent identity changed"));
     }
-    if let Ok(destination) = parent.open_child_file(&target) {
-        if hash_reader(destination)? == (entry.bytes, entry.sha256.clone()) {
-            if let Ok(temp) = parent.open_child_file(&temporary) {
-                let id = graphforge_filesystem::file_identity(&temp).map_err(storage)?;
-                drop(temp);
-                parent
-                    .unlink_child_if_identity(&temporary, id)
-                    .map_err(storage)?;
-                parent.sync().map_err(storage)?;
-            }
-            return Ok(());
+    if temporary_parent.identity() != parent_id {
+        return Err(storage("rewrite temporary and destination parents differ"));
+    }
+
+    match parent.open_child_file(&temporary) {
+        Ok(temp) => {
+            authenticate_staged_file(&temp, entry)?;
+            authenticate_prior_destination(&parent, &target, entry.prior_destination.as_ref())?;
+            let expected = graphforge_filesystem::file_identity(&temp).map_err(storage)?;
+            drop(temp);
+            parent
+                .replace_child(&temporary, expected, &target)
+                .map_err(storage)?;
+            parent.sync().map_err(storage)?;
+            authenticate_installed_destination(&parent, &target, entry)?;
+        }
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            // A missing retained temporary is valid only after its exact file
+            // identity was installed at the destination by an earlier pass.
+            authenticate_installed_destination(&parent, &target, entry)?;
         }
+        Err(error) => return Err(storage(error)),
     }
-    let temp = parent.open_child_file(&temporary).map_err(storage)?;
-    let temp_id = graphforge_filesystem::file_identity(&temp).map_err(storage)?;
-    if (temp_id.volume_serial, hex(&temp_id.file_id))
+    Ok(())
+}
+
+fn authenticate_staged_file(file: &File, entry: &Entry) -> Result<(), GfError> {
+    let identity = graphforge_filesystem::file_identity(file).map_err(storage)?;
+    if (identity.volume_serial, hex(&identity.file_id))
         != (entry.temporary_volume, entry.temporary_file.clone())
     {
         return Err(storage("rewrite temporary identity changed"));
     }
-    if hash_reader(temp.try_clone().map_err(storage)?)? != (entry.bytes, entry.sha256.clone()) {
+    if hash_reader(file.try_clone().map_err(storage)?)? != (entry.bytes, entry.sha256.clone()) {
         return Err(storage("rewrite recovery input is missing or corrupt"));
     }
-    let expected = graphforge_filesystem::file_identity(&temp).map_err(storage)?;
-    drop(temp);
-    parent
-        .replace_child(&temporary, expected, &target)
-        .map_err(storage)?;
-    parent.sync().map_err(storage)?;
-    if hash_reader(parent.open_child_file(&target).map_err(storage)?)?
-        != (entry.bytes, entry.sha256.clone())
+    Ok(())
+}
+
+fn authenticate_prior_destination(
+    parent: &StableDirectory,
+    target: &std::ffi::OsStr,
+    prior: Option<&AuthenticatedFile>,
+) -> Result<(), GfError> {
+    match (parent.open_child_file(target), prior) {
+        (Err(error), None) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
+        (Ok(file), Some(prior)) => {
+            let identity = graphforge_filesystem::file_identity(&file).map_err(storage)?;
+            if (identity.volume_serial, hex(&identity.file_id))
+                != (prior.volume, prior.file.clone())
+                || hash_reader(file)? != (prior.bytes, prior.sha256.clone())
+            {
+                return Err(storage("rewrite destination changed after intent"));
+            }
+            Ok(())
+        }
+        (Err(error), Some(_)) if error.kind() == std::io::ErrorKind::NotFound => {
+            Err(storage("rewrite destination disappeared after intent"))
+        }
+        (Ok(_), None) => Err(storage("rewrite destination appeared after intent")),
+        (Err(error), _) => Err(storage(error)),
+    }
+}
+
+fn authenticate_installed_destination(
+    parent: &StableDirectory,
+    target: &std::ffi::OsStr,
+    entry: &Entry,
+) -> Result<(), GfError> {
+    let destination = parent.open_child_file(target).map_err(storage)?;
+    let identity = graphforge_filesystem::file_identity(&destination).map_err(storage)?;
+    if (identity.volume_serial, hex(&identity.file_id))
+        != (entry.temporary_volume, entry.temporary_file.clone())
+        || hash_reader(destination)? != (entry.bytes, entry.sha256.clone())
     {
         return Err(storage(
             "installed rewrite destination failed authentication",
@@ -354,7 +418,9 @@ fn read_journal(root: &StableDirectory) -> Result<Option<(Vec<u8>, FileIdentity)
     if metadata.len() > MAX_JOURNAL_BYTES {
         return Err(storage("rewrite journal exceeds bound"));
     }
-    let mut bytes = Vec::with_capacity(metadata.len() as usize);
+    let capacity = usize::try_from(metadata.len())
+        .map_err(|_| storage("rewrite journal length exceeds address space"))?;
+    let mut bytes = Vec::with_capacity(capacity);
     file.take(MAX_JOURNAL_BYTES + 1)
         .read_to_end(&mut bytes)
         .map_err(storage)?;
@@ -451,6 +517,12 @@ fn cleanup_preparing_input(
     entry: &Entry,
 ) -> Result<(), GfError> {
     let (parent, temporary) = retained_parent_at(root_path, root, &entry.temporary)?;
+    let parent_id = parent.identity();
+    if (parent_id.volume_serial, hex(&parent_id.file_id))
+        != (entry.parent_volume, entry.parent_file.clone())
+    {
+        return Err(storage("preparing rewrite parent identity changed"));
+    }
     let file = match parent.open_child_file(&temporary) {
         Ok(file) => file,
         Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
@@ -584,6 +656,11 @@ pub(crate) fn recover(root: &Path) -> Result<(), GfError> {
     )
 }
 
+#[cfg(test)]
+static FAIL_AFTER_DURABLE_INTENT: std::sync::atomic::AtomicBool =
+    std::sync::atomic::AtomicBool::new(false);
+
+#[allow(clippy::too_many_lines)]
 pub(crate) fn commit(
     batch: RewriteBatch,
     root: &Path,
@@ -604,6 +681,9 @@ pub(crate) fn commit(
         topology: recovered.topology,
         search: recovered.search,
     };
+    if batch.is_empty() && !bump_topology && !bump_search && auxiliary.is_none() {
+        return Ok(prior);
+    }
     let next = GenerationPair {
         topology: if bump_topology {
             prior
@@ -625,7 +705,7 @@ pub(crate) fn commit(
     let generation_bytes = crate::generation::encode_generation_state(next.topology, next.search)?;
     guard.revalidate()?;
     let transaction = Uuid::now_v7().simple().to_string();
-    let root_identity = identity(root)?;
+    let root_identity = guard.directory.identity();
     let mut entries = Vec::new();
     let mut staged = batch.into_staged();
     let generation_path = root.join("topology/generation.json");
@@ -638,7 +718,7 @@ pub(crate) fn commit(
         .map_err(storage)?;
     generation
         .write_all(&generation_bytes)
-        .and_then(|_| generation.as_file().sync_all())
+        .and_then(|()| generation.as_file().sync_all())
         .map_err(storage)?;
     staged.push((generation, generation_path));
     if staged.len() > MAX_ENTRIES {
@@ -646,20 +726,28 @@ pub(crate) fn commit(
     }
     let last = staged.len().saturating_sub(1);
     for (ordinal, (temp, destination)) in staged.iter().enumerate() {
-        let relative = canonical_relative(root, &destination)?;
-        let parent = destination
-            .parent()
-            .ok_or_else(|| storage("rewrite destination has no parent"))?;
-        let parent_identity = identity(parent)?;
+        let relative = canonical_relative(root, destination)?;
+        let (parent, target) = retained_parent_at(root, &guard.directory, &relative)?;
+        let parent_identity = parent.identity();
         temp.as_file().sync_all().map_err(storage)?;
         let original = graphforge_filesystem::file_identity(temp.as_file()).map_err(storage)?;
         let durable = temp.path().to_path_buf();
-        if graphforge_filesystem::path_identity(&durable).map_err(storage)? != original {
+        let temp_relative = canonical_relative(root, &durable)?;
+        let (temp_parent, temp_name) = retained_parent_at(root, &guard.directory, &temp_relative)?;
+        if temp_parent.identity() != parent_identity {
+            return Err(storage("rewrite temporary and destination parents differ"));
+        }
+        let named_temp = temp_parent.open_child_file(&temp_name).map_err(storage)?;
+        if graphforge_filesystem::file_identity(&named_temp).map_err(storage)? != original {
             return Err(storage("rewrite temporary identity changed before intent"));
         }
-        sync_dir(parent)?;
+        parent.sync().map_err(storage)?;
         let (bytes, sha256) = hash_reader(temp.as_file().try_clone().map_err(storage)?)?;
-        let temp_relative = canonical_relative(root, &durable)?;
+        let prior_destination = match parent.open_child_file(&target) {
+            Ok(file) => Some(authenticated_file(&file)?),
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
+            Err(error) => return Err(storage(error)),
+        };
         entries.push(Entry {
             class: if ordinal == last {
                 EntryClass::GenerationAuthority
@@ -668,20 +756,21 @@ pub(crate) fn commit(
             },
             destination: relative,
             temporary: temp_relative,
-            parent_volume: parent_identity.0,
-            parent_file: parent_identity.1,
+            parent_volume: parent_identity.volume_serial,
+            parent_file: hex(&parent_identity.file_id),
             bytes,
             sha256,
             temporary_volume: original.volume_serial,
             temporary_file: hex(&original.file_id),
+            prior_destination,
         });
     }
     let mut intent = Intent {
         version: 1,
         state: IntentState::Preparing,
         transaction,
-        root_volume: root_identity.0,
-        root_file: root_identity.1,
+        root_volume: root_identity.volume_serial,
+        root_file: hex(&root_identity.file_id),
         prior,
         next,
         auxiliary,
@@ -749,6 +838,8 @@ mod tests {
 
     use super::*;
 
+    static DURABLE_INTENT_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
     fn entry(destination: &str, class: EntryClass) -> Entry {
         Entry {
             class,
@@ -760,6 +851,7 @@ mod tests {
             sha256: "aa".repeat(32),
             temporary_volume: 1,
             temporary_file: "02".to_owned(),
+            prior_destination: None,
         }
     }
 
@@ -787,6 +879,42 @@ mod tests {
         }
     }
 
+    fn leave_durable_intent(root: &Path) -> Intent {
+        let _test_guard = DURABLE_INTENT_TEST_LOCK.lock().unwrap();
+        let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)]));
+        let batch = RecordBatch::try_new(
+            Arc::clone(&schema),
+            vec![Arc::new(Int64Array::from(vec![7_i64]))],
+        )
+        .unwrap();
+        let mut rewrite = RewriteBatch::new();
+        rewrite
+            .stage(&root.join("topology/nodes.parquet"), schema, &batch)
+            .unwrap();
+        FAIL_AFTER_DURABLE_INTENT.store(true, std::sync::atomic::Ordering::SeqCst);
+        assert!(commit(rewrite, root, true, true, None).is_err());
+        let stable = StableDirectory::open(root).unwrap();
+        let (bytes, _) = read_journal(&stable).unwrap().unwrap();
+        let value: Intent = serde_json::from_slice(&bytes).unwrap();
+        assert_eq!(value.state, IntentState::Durable);
+        value
+    }
+
+    fn republish_intent(root: &Path, intent: &mut Intent) {
+        intent.checksum = checksum(intent).unwrap();
+        publish_journal(&StableDirectory::open(root).unwrap(), intent).unwrap();
+    }
+
+    fn assert_recovery_fails_before_authority(root: &Path) {
+        assert!(recover(root).is_err());
+        assert_eq!(
+            crate::generation::read_generation_state_raw(root)
+                .unwrap()
+                .topology,
+            0
+        );
+    }
+
     #[test]
     fn intent_validation_rejects_duplicate_reserved_and_nonmonotonic_authority() {
         assert!(validate_intent(&intent()).is_ok());
@@ -829,19 +957,107 @@ mod tests {
     }
 
     #[test]
-    fn ordinary_error_after_durable_intent_rolls_forward_on_double_reopen() {
+    fn hostile_root_traversal_and_stale_intents_fail_closed() {
+        for mutation in ["root", "traversal", "stale"] {
+            let root = TempDir::new().unwrap();
+            let mut value = leave_durable_intent(root.path());
+            match mutation {
+                "root" => value.root_file = "00".repeat(16),
+                "traversal" => value.entries[0].destination = "../escape.parquet".to_owned(),
+                "stale" => {
+                    value.prior = GenerationPair {
+                        topology: 7,
+                        search: 7,
+                    };
+                    value.next = GenerationPair {
+                        topology: 8,
+                        search: 8,
+                    };
+                }
+                _ => unreachable!(),
+            }
+            republish_intent(root.path(), &mut value);
+            assert_recovery_fails_before_authority(root.path());
+        }
+    }
+
+    #[test]
+    fn substituted_or_truncated_temporary_fails_closed() {
+        for mutation in ["substitute", "truncate"] {
+            let root = TempDir::new().unwrap();
+            let value = leave_durable_intent(root.path());
+            let data = value
+                .entries
+                .iter()
+                .find(|entry| entry.class == EntryClass::Data)
+                .unwrap();
+            let temporary = root.path().join(&data.temporary);
+            let bytes = std::fs::read(&temporary).unwrap();
+            if mutation == "substitute" {
+                let replacement = temporary.with_extension("replacement");
+                std::fs::write(&replacement, bytes).unwrap();
+                std::fs::remove_file(&temporary).unwrap();
+                std::fs::rename(replacement, &temporary).unwrap();
+            } else {
+                std::fs::write(&temporary, &bytes[..bytes.len() / 2]).unwrap();
+            }
+            assert_recovery_fails_before_authority(root.path());
+        }
+    }
+
+    #[test]
+    fn byte_identical_final_substitution_fails_closed() {
         let root = TempDir::new().unwrap();
-        let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)]));
-        let batch = RecordBatch::try_new(
-            Arc::clone(&schema),
-            vec![Arc::new(Int64Array::from(vec![7_i64]))],
+        let value = leave_durable_intent(root.path());
+        let data = value
+            .entries
+            .iter()
+            .find(|entry| entry.class == EntryClass::Data)
+            .unwrap();
+        let stable = StableDirectory::open(root.path()).unwrap();
+        install(root.path(), &stable, data).unwrap();
+        let destination = root.path().join(&data.destination);
+        let bytes = std::fs::read(&destination).unwrap();
+        let replacement = destination.with_extension("replacement");
+        std::fs::write(&replacement, bytes).unwrap();
+        std::fs::remove_file(&destination).unwrap();
+        std::fs::rename(replacement, &destination).unwrap();
+        assert_recovery_fails_before_authority(root.path());
+    }
+
+    #[test]
+    fn parent_substitution_and_cross_root_temp_move_fail_closed() {
+        let root = TempDir::new().unwrap();
+        let _value = leave_durable_intent(root.path());
+        std::fs::rename(
+            root.path().join("topology"),
+            root.path().join("old-topology"),
         )
         .unwrap();
+        std::fs::create_dir(root.path().join("topology")).unwrap();
+        assert_recovery_fails_before_authority(root.path());
+
+        let root = TempDir::new().unwrap();
+        let value = leave_durable_intent(root.path());
+        let data = value
+            .entries
+            .iter()
+            .find(|entry| entry.class == EntryClass::Data)
+            .unwrap();
+        let outside = TempDir::new().unwrap();
+        std::fs::rename(
+            root.path().join(&data.temporary),
+            outside.path().join("moved-input"),
+        )
+        .unwrap();
+        assert_recovery_fails_before_authority(root.path());
+    }
+
+    #[test]
+    fn ordinary_error_after_durable_intent_rolls_forward_on_double_reopen() {
+        let root = TempDir::new().unwrap();
         let destination = root.path().join("topology/nodes.parquet");
-        let mut rewrite = RewriteBatch::new();
-        rewrite.stage(&destination, schema, &batch).unwrap();
-        FAIL_AFTER_DURABLE_INTENT.store(true, std::sync::atomic::Ordering::SeqCst);
-        assert!(commit(rewrite, root.path(), true, true, None).is_err());
+        let _intent = leave_durable_intent(root.path());
         assert!(root.path().join(JOURNAL).is_file());
         assert_eq!(
             crate::generation::read_topology_generation(root.path()).unwrap(),
@@ -870,7 +1086,9 @@ mod tests {
             for (index, relative) in destinations.iter().enumerate() {
                 let batch = RecordBatch::try_new(
                     Arc::clone(&schema),
-                    vec![Arc::new(Int64Array::from(vec![index as i64 + 10]))],
+                    vec![Arc::new(Int64Array::from(vec![
+                        i64::try_from(index).unwrap() + 10,
+                    ]))],
                 )
                 .unwrap();
                 rewrite
@@ -931,7 +1149,11 @@ mod tests {
                         .as_any()
                         .downcast_ref::<Int64Array>()
                         .unwrap();
-                    assert_eq!(values.value(0), index as i64 + 10, "{phase}: {relative}");
+                    assert_eq!(
+                        values.value(0),
+                        i64::try_from(index).unwrap() + 10,
+                        "{phase}: {relative}"
+                    );
                     assert!(reader.next().is_none(), "{phase}: duplicate payload");
                 }
             }
@@ -948,7 +1170,3 @@ mod tests {
         }
     }
 }
-
-#[cfg(test)]
-static FAIL_AFTER_DURABLE_INTENT: std::sync::atomic::AtomicBool =
-    std::sync::atomic::AtomicBool::new(false);
diff --git a/docs/adr/0013-project-generation-protocol.md b/docs/adr/0013-project-generation-protocol.md
index 875144df..7b385214 100644
--- a/docs/adr/0013-project-generation-protocol.md
+++ b/docs/adr/0013-project-generation-protocol.md
@@ -169,6 +169,70 @@ Derived caches may remain under root `cache/` only when they are keyed by the
 canonical source fingerprint and their absence or corruption cannot change
 results.
 
+### Durable mutable-topology rewrite transaction
+
+The graph workspace still contains mutable topology, property, and search files
+while a complete project generation is being constructed. A rewrite of more
+than one such file uses one authenticated, generation-last transaction. This
+transaction is subordinate to project-generation publication: its
+`topology/generation.json` record selects one internally consistent graph
+workspace state, but only `CURRENT` can publish that workspace as a project
+generation.
+
+The transaction retains the admitted project-root directory identity and an
+open, exclusively locked `.graphforge-rewrite.lock` handle. After locking it
+revalidates the named lock, its single-link invariant, the admitted root, and
+every opened descendant directory. The lock is held from recovery and
+prior-generation selection through installation, the generation-authority
+switch, namespace barriers, and journal removal. Prior and next topology/search
+generations are therefore derived inside the same critical section; standalone
+counter changes use this authority rather than racing a rewrite.
+
+Each destination is a canonical UTF-8 project-relative path containing only
+normal components. Caller paths, absolute paths, traversal, duplicate
+destinations or temporaries, and the rewrite journal/lock names are rejected.
+Before intent publication, each staged file is flushed and bound to:
+
+- its exact byte length and SHA-256 digest;
+- its retained parent-directory volume and file identity; and
+- its temporary basename, volume, and file identity.
+
+The authenticated journal also binds the project-root identity, transaction
+identifier, exact prior and next generation pair, entry class, and a checksum
+over every recovery control. It is limited to 16,384 entries and 8 MiB; the
+encoded generation authority is limited to 4 KiB. Work above those limits fails
+before authority changes.
+
+Intent publication has two durable states. `preparing` permits only
+identity-matched temporary cleanup if retaining all staged names was
+interrupted. Once every staged file has been deliberately retained, `durable`
+means recovery must roll forward. Recovery authenticates the journal and root,
+accepts only the exact prior or next generation pair, then installs and verifies
+all data destinations descriptor-relatively. A destination already containing
+the authenticated bytes is an idempotent completed install; otherwise the
+exact authenticated temporary must still exist. Missing, substituted,
+truncated, cross-root, or ambiguous state fails closed and preserves evidence;
+the named rewrite lock additionally requires one link.
+
+Exactly one entry is the generation authority and its destination is
+`topology/generation.json`. Its bounded JSON bytes must encode the journal's
+exact next topology/search pair. It is installed only after every data entry and
+after the retained root and rewrite-lock identities are revalidated. Directory
+namespace barriers follow each installation. The journal is removed and the
+root namespace made durable only after the authority switch. Thus a crash
+before durable intent leaves the prior generation; a crash at or after durable
+intent deterministically rolls forward; and repeated recovery neither
+duplicates rows nor elects state by scanning files.
+
+An auxiliary storage participant may join the transaction with one typed
+receipt. The receipt names one exact staged data destination and binds its
+schema kind/version, byte length, and SHA-256 digest; recovery verifies that
+binding before installation. The #931 authenticated UUID-to-surrogate index
+must use this hook so its manifest/run receipt and topology shards advance under
+the same generation-last authority. Auxiliary metadata alone, an unlisted
+receipt, or a digest that differs from the staged entry is not participation
+and fails closed.
+
 ### Container creation
 
 An absent path, or an explicitly supplied empty directory, may become a new
@@ -498,8 +562,9 @@ identity after locking to prevent path substitution.
 - Recovery is deterministic because it never elects a generation.
 - Windows NTFS and POSIX implementations must meet their documented
   platform-native barriers or fail before mutation; ReFS is unsupported.
-- Existing fixed-path `RewriteBatch` and standalone generation counters are
-  transitional internals to be replaced and related knowledge-layer issues.
+- Multi-file topology rewrites and standalone topology/search generation
+  changes share the authenticated generation-last rewrite transaction; they do
+  not expose a committed prefix or a counter that names partial bytes.
 
 The cost is duplicate immutable snapshot data until later content-addressed
 deduplication. Correctness and a finite recovery proof take precedence.
diff --git a/docs/book/architecture/concurrency-recovery.md b/docs/book/architecture/concurrency-recovery.md
index a3c3804d..97619646 100644
--- a/docs/book/architecture/concurrency-recovery.md
+++ b/docs/book/architecture/concurrency-recovery.md
@@ -60,6 +60,32 @@ Recovery resolves an exact valid `CURRENT` only. Journals and directory scans
 are advisory cleanup input. Corrupt or ambiguous pointers fail closed as
 `GF_PROJECT_CORRUPT` without electing a newest generation.
 
+Mutable graph-workspace rewrites have a second, subordinate recovery boundary.
+One exclusive `.graphforge-rewrite.lock` protects the exact admitted project
+root while the engine derives prior/next topology and search generations,
+installs a bounded batch, and switches `topology/generation.json` last. The
+authenticated intent binds the retained root and parent-directory identities,
+every canonical staged/final basename, temporary identity, byte length and
+digest, and exactly one bounded generation-authority record. It cannot publish
+a project generation; `CURRENT` remains the sole project authority.
+
+A `preparing` intent is safe only for identity-matched temporary cleanup. A
+`durable` intent is an unconditional roll-forward obligation: recovery accepts
+only the exact prior or next generation pair, authenticates an already-installed
+destination or its retained temporary, installs all data first, revalidates the
+root and named lock, and publishes generation authority last. Missing,
+substituted, truncated, non-canonical, cross-root, duplicate, or
+generation-divergent state fails closed without guessing or deleting ambiguous
+evidence; the named lock also has a single-link invariant. Reopen is idempotent
+before, during, and after the authority switch. The journal is bounded to
+16,384 entries and 8 MiB, with a 4 KiB generation authority.
+
+The same intent can bind one typed auxiliary receipt to one exact staged data
+entry by schema kind/version, length, and digest. #931 must use that
+participation point for its authenticated UUID-to-surrogate manifest/runs,
+ensuring topology and lookup authority recover together rather than through a
+later best-effort repair.
+
 ## Write modes and isolation
 
 Project mutation has three explicit embedded modes. Every mode gives readers
@@ -117,6 +143,13 @@ generation or the newly published complete generation according to the durable
 publication phase. Graph, provenance, knowledge-layer, and epistemic state move together; mixed
 generations are unsupported and treated as corruption.
 
+Within a private mutable graph workspace, the equivalent finite rule is:
+before durable rewrite intent the prior topology/search generation remains
+selected; at or after durable intent recovery rolls every staged destination
+forward and installs `topology/generation.json` last. That internal authority is
+fully recovered before the workspace can participate in a `CURRENT`
+publication.
+
 Exact retry after acknowledgement returns the prior receipt without restaging.
 Same-identity content changes return `GF_IDEMPOTENCY_CONFLICT`. Failures before
 `CURRENT` replacement report `committed: false`. Failures after replacement
diff --git a/docs/book/architecture/storage.md b/docs/book/architecture/storage.md
index fa50cf44..1ebb972c 100644
--- a/docs/book/architecture/storage.md
+++ b/docs/book/architecture/storage.md
@@ -209,6 +209,38 @@ knowledge, or epistemic tables. Semantic table ownership remains with the
 domain crates defined by
 [ADR 0012](../../adr/0012-knowledge-domain-ownership.md).
 
+#### Mutable topology rewrite recovery
+
+Before a graph workspace becomes an immutable project participant, topology,
+property, search, and index maintenance may replace several fixed-path files.
+Those files advance through one authenticated durable rewrite, never through a
+sequence of independently committed renames. The engine retains the admitted
+project-root identity and each destination parent directory, holds the named
+rewrite lock exclusively, and binds every staged/final relative path,
+temporary-file identity, exact length, and SHA-256 digest in a checksummed
+intent. Paths must be canonical descendants; substitution, traversal,
+duplicate names, and cross-root state fail closed, while the named rewrite lock
+also requires one link.
+
+The intent is bounded to 16,384 entries and 8 MiB. Its sole generation-authority
+entry is `topology/generation.json`, whose JSON is bounded to 4 KiB and must
+encode the exact next topology/search pair. Data files are installed and
+authenticated first, directory namespace barriers are completed, retained
+root/lock identities are revalidated, and generation authority is installed
+last. Only then is the intent removed durably. This makes an existing matching
+destination an idempotent completed step while refusing a missing or changed
+temporary instead of accepting a partial batch.
+
+An interrupted `preparing` intent cleans up only identity-matched retained
+temporaries. An interrupted `durable` intent always rolls forward from either
+the exact prior or exact next generation; any other generation state is
+corruption. The #931 UUID-to-surrogate index must participate through a typed
+auxiliary receipt that names and authenticates one exact staged receipt entry,
+so topology shards and index authority recover atomically. This internal
+topology/search generation is not project publication authority: a recovered
+workspace is still invisible to new project readers until the complete
+generation is selected by `CURRENT`.
+
 Unless a root is shown explicitly, graph paths in the sections below are
 relative to the pinned generation's `participants/graph/`; primary workbench
 paths are relative to `participants/workbench/`; derived index paths are

From 5c8f70fc9720133e2774d3e697ce47b47d59613b Mon Sep 17 00:00:00 2001
From: David Spencer <1526975+DecisionNerd@users.noreply.github.com>
Date: Tue, 25 Aug 2026 10:11:03 -0600
Subject: [PATCH 3/6] fix(storage): reconcile operational graph files exactly

---
 crates/graphforge-storage/src/graph_files.rs | 135 ++++++++++++++++++-
 1 file changed, 130 insertions(+), 5 deletions(-)

diff --git a/crates/graphforge-storage/src/graph_files.rs b/crates/graphforge-storage/src/graph_files.rs
index c978a5a0..d987eab2 100644
--- a/crates/graphforge-storage/src/graph_files.rs
+++ b/crates/graphforge-storage/src/graph_files.rs
@@ -606,6 +606,68 @@ fn validate_inventory_contract(inventory: &GraphFilesInventory) -> Result<(), Gf
 }
 
 fn collect_source_files(directory: &Path, paths: &mut Vec<PathBuf>) -> Result<(), GfError> {
+    collect_source_files_from(directory, directory, paths)
+}
+
+/// Operational files that may legitimately coexist with the graph workspace
+/// but are never generation data. Keep this list structural and exact: a
+/// basename suffix/prefix rule would let unauthenticated data disappear from
+/// inventory reconciliation while topology readers consume it.
+fn is_graph_operational_file(relative: &Path) -> bool {
+    let Some(components) = relative
+        .components()
+        .map(|component| match component {
+            std::path::Component::Normal(value) => value.to_str(),
+            _ => None,
+        })
+        .collect::<Option<Vec<_>>>()
+    else {
+        return false;
+    };
+    match components.as_slice() {
+        [".graphforge-rewrite.lock"]
+        | ["embeddings", ".catalog.lock" | ".refresh.lock"]
+        | ["graph-objects", "lifecycle.lock"]
+        | ["indexes", "search", .., ".writer.lock"] => true,
+        ["embeddings", name]
+            if name
+                .strip_prefix(".writer-")
+                .and_then(|value| value.strip_suffix(".lock"))
+                .is_some_and(|value| {
+                    value.len() == 64
+                        && value
+                            .bytes()
+                            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
+                }) =>
+        {
+            true
+        }
+        ["embeddings", "spaces", identity, ".writer.lock"]
+            if identity.len() == 64
+                && identity
+                    .bytes()
+                    .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) =>
+        {
+            true
+        }
+        ["graph-objects", "active", lease]
+            if lease.strip_suffix(".lock").is_some_and(|value| {
+                uuid::Uuid::parse_str(value).is_ok_and(|parsed| {
+                    value.len() == 36 && parsed.hyphenated().to_string() == value
+                })
+            }) =>
+        {
+            true
+        }
+        _ => false,
+    }
+}
+
+fn collect_source_files_from(
+    root: &Path,
+    directory: &Path,
+    paths: &mut Vec<PathBuf>,
+) -> Result<(), GfError> {
     let mut entries = directory
         .read_dir()
         .map_err(|error| storage("read graph workspace", directory, error))?
@@ -621,11 +683,12 @@ fn collect_source_files(directory: &Path, paths: &mut Vec<PathBuf>) -> Result<()
             return Err(validation("graph workspace contains a symbolic link"));
         }
         if file_type.is_dir() {
-            collect_source_files(&path, paths)?;
+            collect_source_files_from(root, &path, paths)?;
         } else if file_type.is_file() {
-            let name = entry.file_name();
-            let name = name.to_string_lossy();
-            if name.ends_with(".lock") || name.starts_with(".gf-stage-") {
+            let relative = path
+                .strip_prefix(root)
+                .map_err(|_| validation("graph workspace path escaped root"))?;
+            if is_graph_operational_file(relative) {
                 continue;
             }
             paths.push(path);
@@ -661,6 +724,9 @@ fn collect_regular_files(
             let relative = path
                 .strip_prefix(root)
                 .map_err(|_| corrupt("generation graph path escaped tree"))?;
+            if is_graph_operational_file(relative) {
+                continue;
+            }
             validate_relative_path(relative)?;
             let key = path_text(relative)?;
             if observed.insert(key, path).is_some() {
@@ -844,7 +910,7 @@ mod tests {
         fs::create_dir_all(source.path().join("topology/edges")).unwrap();
         fs::write(source.path().join("topology/nodes.parquet"), b"nodes").unwrap();
         fs::write(source.path().join("topology/edges/knows.parquet"), b"edges").unwrap();
-        fs::write(source.path().join("writer.lock"), b"ignored").unwrap();
+        fs::write(source.path().join(".graphforge-rewrite.lock"), b"ignored").unwrap();
 
         let (inventory, participant) = capture_graph_files(source.path()).unwrap();
         assert_eq!(inventory.file_count, 2);
@@ -858,6 +924,65 @@ mod tests {
         assert_eq!(decode_inventory(&participant.bytes).unwrap(), inventory);
     }
 
+    #[test]
+    fn staging_like_parquet_name_is_authenticated_and_tamper_detected() {
+        let source = tempfile::tempdir().unwrap();
+        let edge_dir = source.path().join("topology/edges/KNOWS");
+        fs::create_dir_all(&edge_dir).unwrap();
+        let injected = edge_dir.join(".gf-stage-injected.parquet");
+        fs::write(&injected, b"untrusted-edge-bytes").unwrap();
+
+        let (inventory, _) = capture_graph_files(source.path()).unwrap();
+        assert!(inventory.files.iter().any(|entry| {
+            entry.relative_path == "topology/edges/KNOWS/.gf-stage-injected.parquet"
+        }));
+        let generation = tempfile::tempdir().unwrap();
+        stage_graph_tree(source.path(), generation.path(), &inventory).unwrap();
+        let graph_root = graph_tree_root(generation.path());
+        fs::write(
+            graph_root.join("topology/edges/KNOWS/.gf-stage-late.parquet"),
+            b"late injection",
+        )
+        .unwrap();
+        assert!(verify_graph_tree(&graph_root, &inventory).is_err());
+    }
+
+    #[test]
+    fn operational_classifier_rejects_noncanonical_identities() {
+        assert!(!is_graph_operational_file(Path::new(&format!(
+            "embeddings/.writer-{}.lock",
+            "A".repeat(64)
+        ))));
+        assert!(!is_graph_operational_file(Path::new(&format!(
+            "embeddings/spaces/{}/.writer.lock",
+            "F".repeat(64)
+        ))));
+        let canonical = "018f1f39-7b2a-7ab0-8000-000000000001";
+        assert!(is_graph_operational_file(Path::new(&format!(
+            "graph-objects/active/{canonical}.lock"
+        ))));
+        assert!(!is_graph_operational_file(Path::new(&format!(
+            "graph-objects/active/{}.lock",
+            canonical.to_ascii_uppercase()
+        ))));
+        assert!(!is_graph_operational_file(Path::new(
+            "graph-objects/active/018f1f397b2a7ab08000000000000001.lock"
+        )));
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn operational_classifier_never_collapses_non_utf8_components() {
+        use std::ffi::OsString;
+        use std::os::unix::ffi::OsStringExt as _;
+
+        let path = PathBuf::from("indexes")
+            .join("search")
+            .join(OsString::from_vec(vec![0xff]))
+            .join(".writer.lock");
+        assert!(!is_graph_operational_file(&path));
+    }
+
     #[test]
     fn legacy_monolith_and_shards_sort_by_canonical_wire_path() {
         let source = tempfile::tempdir().unwrap();

From f66cebac2eea9d92ade13f5e92650b9e381c1daa Mon Sep 17 00:00:00 2001
From: David Spencer <1526975+DecisionNerd@users.noreply.github.com>
Date: Tue, 25 Aug 2026 10:30:17 -0600
Subject: [PATCH 4/6] fix(storage): exclude rewrite locks from participant
 inventories

---
 .../tests/knowledge_isolation.rs              | 13 +++++-----
 crates/graphforge-storage/src/graph_files.rs  |  2 +-
 .../src/project_generation.rs                 | 24 +++++++++++++++++++
 3 files changed, 32 insertions(+), 7 deletions(-)

diff --git a/crates/graphforge-api/tests/knowledge_isolation.rs b/crates/graphforge-api/tests/knowledge_isolation.rs
index 7100c82c..18fd25e1 100644
--- a/crates/graphforge-api/tests/knowledge_isolation.rs
+++ b/crates/graphforge-api/tests/knowledge_isolation.rs
@@ -1810,12 +1810,13 @@ fn snapshot_tree_excluding_reserved(root: &Path) -> HashMap<PathBuf, Vec<u8>> {
     snapshot_tree(root)
         .into_iter()
         .filter(|(path, _)| {
-            !matches!(
-                path.components()
-                    .next()
-                    .and_then(|part| part.as_os_str().to_str()),
-                Some("provenance" | "knowledge")
-            )
+            path != Path::new(".graphforge-rewrite.lock")
+                && !matches!(
+                    path.components()
+                        .next()
+                        .and_then(|part| part.as_os_str().to_str()),
+                    Some("provenance" | "knowledge")
+                )
         })
         .collect()
 }
diff --git a/crates/graphforge-storage/src/graph_files.rs b/crates/graphforge-storage/src/graph_files.rs
index d987eab2..54d1056c 100644
--- a/crates/graphforge-storage/src/graph_files.rs
+++ b/crates/graphforge-storage/src/graph_files.rs
@@ -613,7 +613,7 @@ fn collect_source_files(directory: &Path, paths: &mut Vec<PathBuf>) -> Result<()
 /// but are never generation data. Keep this list structural and exact: a
 /// basename suffix/prefix rule would let unauthenticated data disappear from
 /// inventory reconciliation while topology readers consume it.
-fn is_graph_operational_file(relative: &Path) -> bool {
+pub(crate) fn is_graph_operational_file(relative: &Path) -> bool {
     let Some(components) = relative
         .components()
         .map(|component| match component {
diff --git a/crates/graphforge-storage/src/project_generation.rs b/crates/graphforge-storage/src/project_generation.rs
index be780a9b..c1d4dfbd 100644
--- a/crates/graphforge-storage/src/project_generation.rs
+++ b/crates/graphforge-storage/src/project_generation.rs
@@ -486,6 +486,9 @@ impl ResolvedProjectGeneration {
                     let relative = path
                         .strip_prefix(&root)
                         .map_err(|_| transaction_failed("participant path is not contained"))?;
+                    if crate::graph_files::is_graph_operational_file(relative) {
+                        continue;
+                    }
                     validate_relative_path(relative)
                         .map_err(|_| transaction_failed("participant path is invalid"))?;
                     let relative = relative
@@ -1985,6 +1988,27 @@ mod tests {
         }
     }
 
+    #[test]
+    fn participant_inventory_ignores_only_canonical_operational_files() {
+        let root = tempfile::Builder::new()
+            .prefix("gf")
+            .tempdir_in("/tmp")
+            .unwrap();
+        let resolved = open_or_initialize_project(root.path()).unwrap();
+        let participants = resolved.participants_root();
+
+        fs::write(participants.join(".graphforge-rewrite.lock"), b"").unwrap();
+        resolved.validate_complete_participant_inventory().unwrap();
+
+        fs::write(participants.join("writer.lock"), b"").unwrap();
+        assert_code(
+            resolved
+                .validate_complete_participant_inventory()
+                .unwrap_err(),
+            "GF_TRANSACTION_FAILED",
+        );
+    }
+
     #[test]
     fn existing_resolution_remains_pinned_after_current_changes() {
         let (root, first) = project();

From 9124179f52cd4d49363b0ded7e11b22da6225df6 Mon Sep 17 00:00:00 2001
From: David Spencer <1526975+DecisionNerd@users.noreply.github.com>
Date: Tue, 25 Aug 2026 10:45:51 -0600
Subject: [PATCH 5/6] fix(ci): recognize persistent rewrite lock rendezvous

---
 .../graphforge-storage/src/project_generation.rs   |  5 +----
 scripts/ci/bulk-construction-conformance.py        | 10 +++++++++-
 scripts/ci/concurrency-short-gate.py               | 10 +++++++++-
 scripts/ci/test-bulk-construction-conformance.py   | 14 ++++++++++++++
 scripts/ci/test-concurrency-short-gate.py          | 14 ++++++++++++++
 5 files changed, 47 insertions(+), 6 deletions(-)

diff --git a/crates/graphforge-storage/src/project_generation.rs b/crates/graphforge-storage/src/project_generation.rs
index c1d4dfbd..c913d741 100644
--- a/crates/graphforge-storage/src/project_generation.rs
+++ b/crates/graphforge-storage/src/project_generation.rs
@@ -1990,10 +1990,7 @@ mod tests {
 
     #[test]
     fn participant_inventory_ignores_only_canonical_operational_files() {
-        let root = tempfile::Builder::new()
-            .prefix("gf")
-            .tempdir_in("/tmp")
-            .unwrap();
+        let root = tempfile::tempdir().unwrap();
         let resolved = open_or_initialize_project(root.path()).unwrap();
         let participants = resolved.participants_root();
 
diff --git a/scripts/ci/bulk-construction-conformance.py b/scripts/ci/bulk-construction-conformance.py
index 86a60a1e..ab65f911 100644
--- a/scripts/ci/bulk-construction-conformance.py
+++ b/scripts/ci/bulk-construction-conformance.py
@@ -27,6 +27,7 @@
 PARITY_TEST = ROOT / "scripts/ci/bulk-construction-parity.py"
 CASE_TIMEOUT_SECONDS = 900
 PERSISTENT_ADMISSION_LOCK_NAME = re.compile(r"\.graphforge-admission-[0-9a-f]{64}\.lock\Z")
+PERSISTENT_REWRITE_LOCK_NAME = ".graphforge-rewrite.lock"
 
 REQUIRED_CASES: dict[str, tuple[str, list[str]]] = {
     "rust-bulk-construction-lib": (
@@ -215,7 +216,14 @@ def unexpected_lock_artifacts(work: Path) -> list[str]:
         is_symlink = path.is_symlink()
         if not is_file and not is_symlink:
             continue
-        if is_file and not is_symlink and PERSISTENT_ADMISSION_LOCK_NAME.fullmatch(path.name):
+        if (
+            is_file
+            and not is_symlink
+            and (
+                PERSISTENT_ADMISSION_LOCK_NAME.fullmatch(path.name)
+                or path.name == PERSISTENT_REWRITE_LOCK_NAME
+            )
+        ):
             continue
         unexpected.append(str(path.relative_to(work)))
     return sorted(unexpected)
diff --git a/scripts/ci/concurrency-short-gate.py b/scripts/ci/concurrency-short-gate.py
index 22ded62c..d3fa5760 100644
--- a/scripts/ci/concurrency-short-gate.py
+++ b/scripts/ci/concurrency-short-gate.py
@@ -25,6 +25,7 @@
 CASE_TIMEOUT_SECONDS = 300
 FORBIDDEN_TIMING = re.compile(r"\b(?:time\.sleep|asyncio\.sleep|setTimeout)\s*\(")
 PERSISTENT_ADMISSION_LOCK_NAME = re.compile(r"\.graphforge-admission-[0-9a-f]{64}\.lock\Z")
+PERSISTENT_REWRITE_LOCK_NAME = ".graphforge-rewrite.lock"
 
 # Required id → (surface, argv). Additional cases are allowed; these must match exactly.
 REQUIRED_CASES: dict[str, tuple[str, list[str]]] = {
@@ -279,7 +280,14 @@ def unexpected_lock_artifacts(work: Path) -> list[str]:
         is_symlink = path.is_symlink()
         if not is_file and not is_symlink:
             continue
-        if is_file and not is_symlink and PERSISTENT_ADMISSION_LOCK_NAME.fullmatch(path.name):
+        if (
+            is_file
+            and not is_symlink
+            and (
+                PERSISTENT_ADMISSION_LOCK_NAME.fullmatch(path.name)
+                or path.name == PERSISTENT_REWRITE_LOCK_NAME
+            )
+        ):
             continue
         unexpected.append(str(path.relative_to(work)))
     return sorted(unexpected)
diff --git a/scripts/ci/test-bulk-construction-conformance.py b/scripts/ci/test-bulk-construction-conformance.py
index e861163d..9f0e35b4 100644
--- a/scripts/ci/test-bulk-construction-conformance.py
+++ b/scripts/ci/test-bulk-construction-conformance.py
@@ -41,6 +41,8 @@ def assert_persistent_admission_lock_contract() -> None:
         nested.mkdir()
         admission_lock = nested / f".graphforge-admission-{'a' * 64}.lock"
         admission_lock.touch()
+        rewrite_lock = nested / ".graphforge-rewrite.lock"
+        rewrite_lock.touch()
         assert GATE.unexpected_lock_artifacts(work) == []
 
         uppercase_dir = work / "uppercase"
@@ -50,6 +52,7 @@ def assert_persistent_admission_lock_contract() -> None:
             nested / f".graphforge-admission-{'a' * 63}.lock",
             uppercase_dir / f".graphforge-admission-{'A' * 64}.lock",
             nested / ".graphforge-admission-not-a-digest.lock",
+            nested / ".graphforge-rewrite-extra.lock",
         ]
         for path in unexpected:
             path.touch()
@@ -67,6 +70,17 @@ def assert_persistent_admission_lock_contract() -> None:
                 [*expected, str(symlink.relative_to(work))]
             )
 
+        rewrite_symlink = uppercase_dir / ".graphforge-rewrite.lock"
+        try:
+            rewrite_symlink.symlink_to(rewrite_lock)
+        except (NotImplementedError, OSError):
+            pass
+        else:
+            expected_symlinks = [str(rewrite_symlink.relative_to(work))]
+            if symlink.is_symlink():
+                expected_symlinks.append(str(symlink.relative_to(work)))
+            assert GATE.unexpected_lock_artifacts(work) == sorted([*expected, *expected_symlinks])
+
 
 def main() -> None:
     matrix = GATE.validate_matrix()
diff --git a/scripts/ci/test-concurrency-short-gate.py b/scripts/ci/test-concurrency-short-gate.py
index 8e30389e..329d9007 100644
--- a/scripts/ci/test-concurrency-short-gate.py
+++ b/scripts/ci/test-concurrency-short-gate.py
@@ -41,6 +41,8 @@ def assert_persistent_admission_lock_contract() -> None:
         nested.mkdir()
         admission_lock = nested / f".graphforge-admission-{'a' * 64}.lock"
         admission_lock.touch()
+        rewrite_lock = nested / ".graphforge-rewrite.lock"
+        rewrite_lock.touch()
         assert GATE.unexpected_lock_artifacts(work) == []
 
         uppercase_dir = work / "uppercase"
@@ -50,6 +52,7 @@ def assert_persistent_admission_lock_contract() -> None:
             nested / f".graphforge-admission-{'a' * 63}.lock",
             uppercase_dir / f".graphforge-admission-{'A' * 64}.lock",
             nested / ".graphforge-admission-not-a-digest.lock",
+            nested / ".graphforge-rewrite-extra.lock",
         ]
         for path in unexpected:
             path.touch()
@@ -67,6 +70,17 @@ def assert_persistent_admission_lock_contract() -> None:
                 [*expected, str(symlink.relative_to(work))]
             )
 
+        rewrite_symlink = uppercase_dir / ".graphforge-rewrite.lock"
+        try:
+            rewrite_symlink.symlink_to(rewrite_lock)
+        except (NotImplementedError, OSError):
+            pass
+        else:
+            expected_symlinks = [str(rewrite_symlink.relative_to(work))]
+            if symlink.is_symlink():
+                expected_symlinks.append(str(symlink.relative_to(work)))
+            assert GATE.unexpected_lock_artifacts(work) == sorted([*expected, *expected_symlinks])
+
 
 def main() -> None:
     matrix = GATE.validate_matrix()

From 658f98d99fe0e9049c4408106934554bc3c6dad2 Mon Sep 17 00:00:00 2001
From: David Spencer <1526975+DecisionNerd@users.noreply.github.com>
Date: Tue, 25 Aug 2026 11:07:22 -0600
Subject: [PATCH 6/6] fix(storage): keep clean generation reads side-effect
 free

---
 crates/graphforge-storage/src/durable_rewrite.rs |  8 ++++++++
 crates/graphforge-storage/src/generation.rs      | 10 +++++++++-
 2 files changed, 17 insertions(+), 1 deletion(-)

diff --git a/crates/graphforge-storage/src/durable_rewrite.rs b/crates/graphforge-storage/src/durable_rewrite.rs
index ba8e763e..adf17e06 100644
--- a/crates/graphforge-storage/src/durable_rewrite.rs
+++ b/crates/graphforge-storage/src/durable_rewrite.rs
@@ -656,6 +656,14 @@ pub(crate) fn recover(root: &Path) -> Result<(), GfError> {
     )
 }
 
+pub(crate) fn recovery_required(root: &Path) -> Result<bool, GfError> {
+    match std::fs::symlink_metadata(root.join(JOURNAL)) {
+        Ok(_) => Ok(true),
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
+        Err(error) => Err(storage(format!("rewrite journal probe failed: {error}"))),
+    }
+}
+
 #[cfg(test)]
 static FAIL_AFTER_DURABLE_INTENT: std::sync::atomic::AtomicBool =
     std::sync::atomic::AtomicBool::new(false);
diff --git a/crates/graphforge-storage/src/generation.rs b/crates/graphforge-storage/src/generation.rs
index 97b53745..58a71f7b 100644
--- a/crates/graphforge-storage/src/generation.rs
+++ b/crates/graphforge-storage/src/generation.rs
@@ -126,7 +126,9 @@ pub(crate) fn encode_generation_state(topology: u64, search: u64) -> Result<Vec<
 }
 
 fn read_generation_state(project_dir: &Path) -> Result<GenerationState, GfError> {
-    crate::durable_rewrite::recover(project_dir)?;
+    if crate::durable_rewrite::recovery_required(project_dir)? {
+        crate::durable_rewrite::recover(project_dir)?;
+    }
     read_generation_state_raw(project_dir)
 }
 
@@ -251,6 +253,12 @@ mod tests {
         let dir = TempDir::new().unwrap();
         assert_eq!(read_topology_generation(dir.path()).unwrap(), 0);
         assert_eq!(read_search_generation(dir.path()).unwrap(), 0);
+        assert!(!dir.path().join(".graphforge-rewrite.lock").exists());
+
+        let missing = dir.path().join("missing-project");
+        assert_eq!(read_topology_generation(&missing).unwrap(), 0);
+        assert_eq!(read_search_generation(&missing).unwrap(), 0);
+        assert!(!missing.exists());
     }
 
     #[test]