Let the application supply the auto-merge policy; add Repo::merge_heads

readme.md

@@ -74,6 +74,34 @@ fn main() {
 }
 ```
 
+### Application-supplied merge policies
+
+`Repo::with_merge_policy` supplies an implementation of `MergePolicy` for a
+named custom policy. A locally created history records its selected policy in
+the genesis metadata. Subsequent automatic merges, including `merge_heads`,
+select the implementation named by that genesis; installing a different custom
+policy does not override an existing history's choice. Histories using `lww`
+continue to use the built-in LWW implementation. An unavailable custom policy
+causes a merge error rather than a fallback to a different rule.
+
+The application owns the custom policy's semantics and must deploy compatible,
+convergent implementations to its replicas. Policy names are identifiers, not
+proof that two implementations behave identically: use a new name when changing
+semantics. This API does not migrate existing histories to a new policy, nor does
+it prove that an imported merge payload was computed correctly.
+
+Parent-dependent policies require all immediate-parent payloads before a merge
+can be saved. If a parent has not synced, import it and retry. Policies independent
+of parents can opt out with `MergePolicy::requires_parent_payloads`.
+
+Replication must preserve `Operation::node_metadata` as well as the node
+timestamp. The metadata is the exact serialized value, not just a policy name;
+replacing it with local defaults can change the node's CID. Legacy operations
+without this field remain readable. Newly written LevelDB operations use a
+versioned storage format: this version reads old records, but older binaries
+cannot read the new records. Do not downgrade a database after writing with this
+version. Older replicas are not compatible with new custom-policy histories.
+
 ## 🖥️ CLI Tool
 
 CRSL includes a command-line interface for easy content management.

src/convergence/policies/lww.rs

@@ -7,6 +7,10 @@ use crate::convergence::policy::{MergePolicy, ResolveInput};
 pub struct LwwMergePolicy;
 
 impl<P: Clone> MergePolicy<P> for LwwMergePolicy {
+    fn requires_parent_payloads(&self) -> bool {
+        false
+    }
+
     fn resolve(&self, nodes: &[ResolveInput<P>]) -> P {
         let winner = nodes
             .iter()

src/convergence/policy.rs

@@ -6,6 +6,20 @@ pub struct ResolveInput<P> {
     pub cid: Cid,
     pub payload: P,
     pub timestamp: u64,
+    /// Payloads of this head's parents, in the node's parent order.
+    ///
+    /// A policy that treats the payload as more than an opaque value needs to
+    /// know what a head *changed*, not just what it holds: a node that copied
+    /// its parent's value for one field while updating another should not win
+    /// a last-writer race on the field it left alone. The library cannot make
+    /// that distinction — it does not know the payload's shape — so it hands
+    /// the parents over and lets the policy compare.
+    ///
+    /// The resolver supplies all immediate parents when the policy requires
+    /// them, or returns an error if any are missing. Empty for a genesis head,
+    /// policies that opt out of parent payloads (such as `LwwMergePolicy`),
+    /// and inputs constructed by callers that have no DAG at hand.
+    pub parent_payloads: Vec<P>,
 }
 
 impl<P> ResolveInput<P> {
@@ -14,15 +28,49 @@ impl<P> ResolveInput<P> {
             cid,
             payload,
             timestamp,
+            parent_payloads: Vec::new(),
+        }
+    }
+
+    pub fn with_parents(cid: Cid, payload: P, timestamp: u64, parent_payloads: Vec<P>) -> Self {
+        Self {
+            cid,
+            payload,
+            timestamp,
+            parent_payloads,
         }
     }
 }
 
 /// A merge strategy that produces a converged payload from candidate nodes.
+///
+/// The library ships [`LwwMergePolicy`](crate::convergence::policies::lww::LwwMergePolicy)
+/// and selects it by the `policy_type` recorded in the genesis metadata. An
+/// application whose payload is a composite — fields with different
+/// convergence rules — supplies its own implementation through
+/// [`Repo::with_merge_policy`](crate::repo::Repo::with_merge_policy); the
+/// library calls it only when its name matches the genesis metadata. The
+/// built-in `lww` name is reserved and always selects the library's LWW rule.
 pub trait MergePolicy<P>: Send + Sync {
+    /// Whether resolution needs the complete payloads of each head's immediate parents.
+    ///
+    /// Defaults to true: the resolver returns a missing-node error before
+    /// invoking `resolve` if any parent has not synced yet. The caller can
+    /// retry after importing the missing parents; no merge is persisted.
+    /// Return false only when resolution is independent of parent payloads.
+    /// In that case the resolver does not load them and supplies empty vectors.
+    fn requires_parent_payloads(&self) -> bool {
+        true
+    }
+
     /// Resolve competing nodes into a single payload.
     fn resolve(&self, nodes: &[ResolveInput<P>]) -> P;
 
-    /// Return a descriptive name of the policy (e.g. "lww").
+    /// Return the stable policy identifier recorded in genesis metadata.
+    ///
+    /// All implementations sharing a name must have identical deterministic
+    /// merge semantics across replicas. Use a new name when those semantics
+    /// change; installing it does not migrate existing content. `lww` is
+    /// reserved for the built-in rule, not an application override.
     fn name(&self) -> &str;
 }

src/convergence/resolver.rs

@@ -52,7 +52,7 @@ where
             ));
         }
 
-        let inputs = self.collect_inputs(heads, dag)?;
+        let inputs = self.collect_inputs(heads, dag, policy.requires_parent_payloads())?;
         let merged_payload = policy.resolve(&inputs);
         let metadata = self.merge_metadata(heads, dag)?;
         Ok(Node::new_child(
@@ -68,6 +68,7 @@ where
         &self,
         heads: &[Cid],
         dag: &DagGraph<S, P, M>,
+        requires_parent_payloads: bool,
     ) -> CrdtResult<Vec<ResolveInput<P>>>
     where
         S: NodeStorage<P, M>,
@@ -80,10 +81,24 @@ where
                 .get_node(&cid)
                 .map_err(CrdtError::Graph)?
                 .ok_or_else(|| CrdtError::Internal(format!("Head node not found: {cid}")))?;
-            inputs.push(ResolveInput::new(
+            let mut parent_payloads = Vec::new();
+            if requires_parent_payloads {
+                parent_payloads.reserve(node.parents().len());
+                for parent in node.parents() {
+                    let parent_node =
+                        dag.get_node(parent)
+                            .map_err(CrdtError::Graph)?
+                            .ok_or(CrdtError::Graph(
+                                crate::graph::error::GraphError::NodeNotFound(*parent),
+                            ))?;
+                    parent_payloads.push(parent_node.payload().clone());
+                }
+            }
+            inputs.push(ResolveInput::with_parents(
                 cid,
                 node.payload().clone(),
                 node.timestamp(),
+                parent_payloads,
             ));
         }
         Ok(inputs)
@@ -193,6 +208,63 @@ mod tests {
         Cid::new_v1(0x55, digest)
     }
 
+    /// Each `ResolveInput` handed to the policy carries its head's parent
+    /// payloads, so a policy can tell a head that changed the value from one
+    /// that only re-committed its parent's.
+    #[test]
+    fn create_merge_node_attaches_parent_payloads() {
+        struct Capture(std::sync::Mutex<Vec<Vec<String>>>);
+        impl MergePolicy<String> for Capture {
+            fn resolve(&self, nodes: &[ResolveInput<String>]) -> String {
+                let mut seen = self.0.lock().unwrap();
+                for n in nodes {
+                    seen.push(n.parent_payloads.clone());
+                }
+                nodes[0].payload.clone()
+            }
+            fn name(&self) -> &str {
+                "capture"
+            }
+        }
+
+        let storage = MemoryNodeStorage::<String, ContentMetadata>::default();
+        let dag = DagGraph::new(storage.clone());
+        let metadata = ContentMetadata::with_policy("capture");
+        let genesis_node = Node::new_genesis("genesis".to_string(), 1, metadata.clone());
+        let genesis_cid = genesis_node.content_id().unwrap();
+        dag.storage.put(&genesis_node).unwrap();
+        let head_a = Node::new_child(
+            "payload-a".to_string(),
+            vec![genesis_cid],
+            genesis_cid,
+            10,
+            metadata.clone(),
+        );
+        let head_a_cid = head_a.content_id().unwrap();
+        dag.storage.put(&head_a).unwrap();
+        let head_b = Node::new_child(
+            "payload-b".to_string(),
+            vec![genesis_cid],
+            genesis_cid,
+            11,
+            metadata,
+        );
+        let head_b_cid = head_b.content_id().unwrap();
+        dag.storage.put(&head_b).unwrap();
+
+        let policy = Capture(std::sync::Mutex::new(Vec::new()));
+        let resolver = ConflictResolver::<String, ContentMetadata>::new();
+        resolver
+            .create_merge_node(&[head_a_cid, head_b_cid], &dag, genesis_cid, 20, &policy)
+            .unwrap();
+
+        let seen = policy.0.into_inner().unwrap();
+        assert_eq!(
+            seen,
+            vec![vec!["genesis".to_string()], vec!["genesis".to_string()]]
+        );
+    }
+
     #[test]
     fn create_merge_node_merges_heads() {
         let storage = MemoryNodeStorage::<String, ContentMetadata>::default();

src/crdt/operation.rs

@@ -2,6 +2,7 @@ use serde::{Deserialize, Serialize};
 use std::fmt::Debug;
 use ulid::Ulid;
 
+use crate::convergence::metadata::ContentMetadata;
 use crate::crdt::timestamp::next_monotonic_timestamp;
 
 /// Unique identifier for operations (based on Ulid)
@@ -66,6 +67,17 @@ pub struct Operation<ContentId, T> {
     /// This ensures CID consistency across replicas.
     #[serde(default)]
     pub node_timestamp: Option<Timestamp>,
+    /// Exact DAG metadata for replication. Missing on legacy operations.
+    /// A local Create may explicitly select metadata here; otherwise the Repo's
+    /// installed policy is recorded. Imported Creates never infer it from the
+    /// receiver's policy: absence means the historical default metadata.
+    /// Repo commits populate this for every node. Updates and Merges carrying
+    /// their payload can therefore reconstruct metadata even before ancestry
+    /// arrives; Deletes still require existing payload history.
+    /// Local Updates/Deletes inherit metadata instead of using this field to
+    /// change policy. Keep the exact metadata representation, not just its name.
+    #[serde(default)]
+    pub node_metadata: Option<ContentMetadata>,
 }
 
 impl<ContentId, T> Operation<ContentId, T>
@@ -95,6 +107,7 @@ where
             author,
             parents: Vec::new(),
             node_timestamp: None,
+            node_metadata: None,
         }
     }
 

src/crdt/reducer.rs

@@ -53,6 +53,7 @@ mod tests {
             author: "test".into(),
             parents: Vec::new(),
             node_timestamp: None,
+            node_metadata: None,
         }
     }
 
@@ -71,6 +72,7 @@ mod tests {
             author: "test".into(),
             parents: Vec::new(),
             node_timestamp: None,
+            node_metadata: None,
         }
     }
 

src/crdt/storage.rs

@@ -1,5 +1,5 @@
 use crate::crdt::error::{CrdtError, Result};
-use crate::crdt::operation::Operation;
+use crate::crdt::operation::{Operation, OperationType};
 use crate::storage::{BatchError, LeveldbBatchGuard, SharedLeveldb, SharedLeveldbAccess};
 use bincode;
 use rusty_leveldb::LdbIterator;
@@ -8,6 +8,20 @@ use std::path::Path;
 use std::sync::Arc;
 use ulid::Ulid;
 
+// Legacy records start with the bincode string length (26) of a ULID.
+// A distinct versioned envelope prevents truncated new metadata from ever
+// being accepted as a legacy operation with default policy metadata.
+const OPERATION_V1: &[u8] = b"CRSLop\x01";
+type LegacyOperation<ContentId, T> = (
+    Ulid,
+    ContentId,
+    OperationType<T>,
+    u64,
+    String,
+    Vec<ContentId>,
+    Option<u64>,
+);
+
 /// Abstraction over the persistent storage used by `CrdtState`.
 pub trait OperationStorage<ContentId, T>: Send + Sync {
     fn save_operation(&self, op: &Operation<ContentId, T>) -> Result<()>;
@@ -20,6 +34,10 @@ pub trait OperationStorage<ContentId, T>: Send + Sync {
 }
 
 /// LevelDB-backed implementation of [`OperationStorage`].
+///
+/// Reads legacy unversioned bincode operations and versioned records carrying
+/// node metadata. New writes use the versioned format; older library versions
+/// cannot read them. Invalid records are errors, never silently skipped.
 #[derive(Clone)]
 pub struct LeveldbStorage<ContentId, T> {
     shared: Arc<SharedLeveldb>,
@@ -53,10 +71,56 @@ impl<ContentId, T> LeveldbStorage<ContentId, T> {
         ContentId: serde::Serialize,
         T: serde::Serialize,
     {
-        let value = bincode::serde::encode_to_vec(op, bincode::config::standard())?;
+        let mut value = OPERATION_V1.to_vec();
+        value.extend(bincode::serde::encode_to_vec(
+            op,
+            bincode::config::standard(),
+        )?);
         Ok(value)
     }
 
+    fn decode_operation(raw: &[u8]) -> Result<Operation<ContentId, T>>
+    where
+        ContentId: for<'de> serde::Deserialize<'de>,
+        T: for<'de> serde::Deserialize<'de>,
+    {
+        let (op, consumed, expected) = if let Some(body) = raw.strip_prefix(OPERATION_V1) {
+            let (op, consumed) =
+                bincode::serde::decode_from_slice(body, bincode::config::standard())?;
+            (op, consumed, body.len())
+        } else if raw.first() == Some(&26) {
+            let ((id, genesis, kind, timestamp, author, parents, node_timestamp), consumed) =
+                bincode::serde::decode_from_slice::<LegacyOperation<ContentId, T>, _>(
+                    raw,
+                    bincode::config::standard(),
+                )?;
+            (
+                Operation {
+                    id,
+                    genesis,
+                    kind,
+                    timestamp,
+                    author,
+                    parents,
+                    node_timestamp,
+                    node_metadata: None,
+                },
+                consumed,
+                raw.len(),
+            )
+        } else {
+            return Err(CrdtError::Internal(
+                "Unknown operation storage format".into(),
+            ));
+        };
+        if consumed != expected {
+            return Err(CrdtError::Internal(
+                "Trailing bytes in stored operation".into(),
+            ));
+        }
+        Ok(op)
+    }
+
     /// Writes value bytes either to the active batch or directly to the DB.
     fn put_bytes(&self, key: &[u8], value: &[u8]) -> Result<()> {
         if self
@@ -117,10 +181,8 @@ where
         let mut value = Vec::new();
         while iter.valid() {
             iter.current(&mut key, &mut value);
-            if let Ok((op, _)) = bincode::serde::decode_from_slice::<Operation<ContentId, T>, _>(
-                &value,
-                bincode::config::standard(),
-            ) {
+            if key.first() == Some(&0x01) {
+                let op = Self::decode_operation(&value)?;
                 if op.genesis == *genesis {
                     result.push(op);
                 }
@@ -134,13 +196,7 @@ where
     fn get_operation(&self, op_id: &Ulid) -> Result<Option<Operation<ContentId, T>>> {
         let key = Self::make_key(op_id);
         match self.shared.db().get(&key) {
-            Some(raw) => {
-                let (op, _) = bincode::serde::decode_from_slice::<Operation<ContentId, T>, _>(
-                    &raw,
-                    bincode::config::standard(),
-                )?;
-                Ok(Some(op))
-            }
+            Some(raw) => Self::decode_operation(&raw).map(Some),
             None => Ok(None),
         }
     }