use anyhow::{Context, Result};
use yrs::updates::decoder::Decode;
use yrs::{Doc, GetString, ReadTxn, Text, TextRef, Transact, Update};
const TEXT_KEY: &str = "content";
pub struct CrdtDoc {
doc: Doc,
}
impl CrdtDoc {
pub fn from_text(content: &str) -> Self {
let doc = Doc::new();
let text = doc.get_or_insert_text(TEXT_KEY);
let mut txn = doc.transact_mut();
text.insert(&mut txn, 0, content);
drop(txn);
CrdtDoc { doc }
}
pub fn to_text(&self) -> String {
let text = self.doc.get_or_insert_text(TEXT_KEY);
let txn = self.doc.transact();
text.get_string(&txn)
}
#[allow(dead_code)] pub fn apply_edit(&self, offset: u32, delete_len: u32, insert: &str) {
let text = self.doc.get_or_insert_text(TEXT_KEY);
let mut txn = self.doc.transact_mut();
if delete_len > 0 {
text.remove_range(&mut txn, offset, delete_len);
}
if !insert.is_empty() {
text.insert(&mut txn, offset, insert);
}
}
pub fn encode_state(&self) -> Vec<u8> {
let txn = self.doc.transact();
txn.encode_state_as_update_v1(&yrs::StateVector::default())
}
pub fn decode_state(bytes: &[u8]) -> Result<Self> {
let doc = Doc::new();
let update = Update::decode_v1(bytes)
.map_err(|e| anyhow::anyhow!("failed to decode CRDT state: {}", e))?;
let mut txn = doc.transact_mut();
txn.apply_update(update)
.map_err(|e| anyhow::anyhow!("failed to apply CRDT update: {}", e))?;
drop(txn);
Ok(CrdtDoc { doc })
}
}
pub fn merge(base_state: Option<&[u8]>, ours_text: &str, theirs_text: &str) -> Result<String> {
merge_inner(base_state, ours_text, theirs_text, None)
}
pub fn merge_with_editor_ops(
base_state: Option<&[u8]>,
ours_text: &str,
theirs_text: &str,
theirs_editor_ops: Option<&[EditorOp]>,
) -> Result<String> {
merge_inner(base_state, ours_text, theirs_text, theirs_editor_ops)
}
fn merge_inner(
base_state: Option<&[u8]>,
ours_text: &str,
theirs_text: &str,
theirs_editor_ops: Option<&[EditorOp]>,
) -> Result<String> {
if ours_text == theirs_text {
eprintln!("[crdt] ours == theirs, skipping merge");
return Ok(ours_text.to_string());
}
let base_doc = if let Some(bytes) = base_state {
CrdtDoc::decode_state(bytes).context("failed to decode base CRDT state")?
} else {
CrdtDoc::from_text("")
};
let mut base_text = base_doc.to_text();
eprintln!(
"[crdt] merge: base_len={} ours_len={} theirs_len={}",
base_text.len(),
ours_text.len(),
theirs_text.len()
);
let committed_response_headings = response_headings(&base_text);
let ours_prefix = common_prefix_len(&base_text, ours_text);
let ours_suffix = common_suffix_len(&base_text, ours_text);
let theirs_prefix = common_prefix_len(&base_text, theirs_text);
let theirs_suffix = common_suffix_len(&base_text, theirs_text);
let base_len = base_text.len();
let ours_shared = (ours_prefix + ours_suffix).min(base_len);
let theirs_shared = (theirs_prefix + theirs_suffix).min(base_len);
if base_len > 0
&& (ours_shared as f64 / base_len as f64) < 0.5
&& (theirs_shared as f64 / base_len as f64) < 0.5
{
eprintln!(
"[crdt] Stale CRDT base detected (shared: ours={}%, theirs={}%). Using ours as base.",
(ours_shared * 100) / base_len,
(theirs_shared * 100) / base_len
);
base_text = ours_text.to_string();
}
let mutual_prefix = common_prefix_len(ours_text, theirs_text);
let base_diverge =
common_prefix_len(&base_text, ours_text).max(common_prefix_len(&base_text, theirs_text));
if mutual_prefix > base_diverge {
let snap = &ours_text[..mutual_prefix];
let snapped = match snap.rfind('\n') {
Some(pos) if pos >= base_diverge => pos + 1,
_ => base_diverge, };
if snapped > base_diverge {
eprintln!(
"[crdt] Advancing base to shared prefix (diverge={} → {})",
base_diverge, snapped
);
base_text = ours_text[..snapped].to_string();
}
}
let theirs_replay_ops = match theirs_editor_ops {
Some(ops) if !ops.is_empty() && base_text.is_ascii() && theirs_text.is_ascii() => {
match replay_editor_ops(&base_text, ops) {
Some(replayed) if replayed == theirs_text => {
eprintln!(
"[crdt] editor_ops_replayed: {} captured op(s) reproduce theirs exactly (#qnodemerge4)",
ops.len()
);
Some(ops)
}
_ => {
eprintln!(
"[crdt] editor_ops_fallback: {} captured op(s) do not replay to theirs from the resolved base (stale/misaligned) — using diff-guess (#qnodemerge4)",
ops.len()
);
None
}
}
}
_ => None,
};
let ours_ops = compute_edit_ops(&base_text, ours_text);
let theirs_ops = compute_edit_ops(&base_text, theirs_text);
let base_encoded = if base_text == base_doc.to_text() {
base_doc.encode_state()
} else {
CrdtDoc::from_text(&base_text).encode_state()
};
let ours_doc = Doc::with_client_id(1);
{
let update =
Update::decode_v1(&base_encoded).map_err(|e| anyhow::anyhow!("decode error: {}", e))?;
let mut txn = ours_doc.transact_mut();
txn.apply_update(update)
.map_err(|e| anyhow::anyhow!("apply error: {}", e))?;
}
let theirs_doc = Doc::with_client_id(2);
{
let update =
Update::decode_v1(&base_encoded).map_err(|e| anyhow::anyhow!("decode error: {}", e))?;
let mut txn = theirs_doc.transact_mut();
txn.apply_update(update)
.map_err(|e| anyhow::anyhow!("apply error: {}", e))?;
}
{
let text = ours_doc.get_or_insert_text(TEXT_KEY);
let mut txn = ours_doc.transact_mut();
apply_ops(&text, &mut txn, &ours_ops);
}
{
let text = theirs_doc.get_or_insert_text(TEXT_KEY);
let mut txn = theirs_doc.transact_mut();
match theirs_replay_ops {
Some(ops) => apply_editor_ops(&text, &mut txn, ops),
None => apply_ops(&text, &mut txn, &theirs_ops),
}
}
let ours_sv = {
let txn = ours_doc.transact();
txn.state_vector()
};
let theirs_update = {
let txn = theirs_doc.transact();
txn.encode_state_as_update_v1(&ours_sv)
};
{
let update = Update::decode_v1(&theirs_update)
.map_err(|e| anyhow::anyhow!("decode error: {}", e))?;
let mut txn = ours_doc.transact_mut();
txn.apply_update(update)
.map_err(|e| anyhow::anyhow!("apply error: {}", e))?;
}
let merged = {
let text = ours_doc.get_or_insert_text(TEXT_KEY);
let txn = ours_doc.transact();
text.get_string(&txn)
};
let merged =
guard_committed_responses(merged, &committed_response_headings, ours_text, theirs_text);
Ok(dedup_adjacent_blocks(&merged))
}
#[derive(Debug, Clone)]
enum Node {
Interstitial(String),
Component { name: String, text: String },
}
impl Node {
fn component_name(&self) -> Option<&str> {
match self {
Node::Component { name, .. } => Some(name.as_str()),
Node::Interstitial(_) => None,
}
}
fn text(&self) -> &str {
match self {
Node::Interstitial(t) => t,
Node::Component { text, .. } => text,
}
}
}
fn segment_into_nodes(text: &str) -> Result<Vec<Node>> {
let comps = crate::component::parse(text)?;
let mut top: Vec<&crate::component::Component> = comps
.iter()
.filter(|c| {
!comps.iter().any(|o| {
!std::ptr::eq(o, *c) && o.open_start <= c.open_start && c.close_end <= o.close_end
})
})
.collect();
top.sort_by_key(|c| c.open_start);
let mut nodes = Vec::with_capacity(top.len() * 2 + 1);
let mut cursor = 0usize;
for c in top {
nodes.push(Node::Interstitial(text[cursor..c.open_start].to_string()));
nodes.push(Node::Component {
name: c.name.clone(),
text: text[c.open_start..c.close_end].to_string(),
});
cursor = c.close_end;
}
nodes.push(Node::Interstitial(text[cursor..].to_string()));
Ok(nodes)
}
pub fn merge_by_component(
base_state: Option<&[u8]>,
ours_text: &str,
theirs_text: &str,
) -> Result<String> {
if ours_text == theirs_text {
return Ok(ours_text.to_string());
}
let ours_nodes = match segment_into_nodes(ours_text) {
Ok(c) => c,
Err(e) => {
eprintln!(
"[crdt] merge_by_component: failed to segment ours ({e}); falling back to whole-doc merge"
);
return merge(base_state, ours_text, theirs_text);
}
};
let theirs_nodes = match segment_into_nodes(theirs_text) {
Ok(c) => c,
Err(e) => {
eprintln!(
"[crdt] merge_by_component: failed to segment theirs ({e}); falling back to whole-doc merge"
);
return merge(base_state, ours_text, theirs_text);
}
};
let ours_names: Vec<&str> = ours_nodes.iter().filter_map(Node::component_name).collect();
let theirs_names: Vec<&str> = theirs_nodes
.iter()
.filter_map(Node::component_name)
.collect();
if ours_names.is_empty() && theirs_names.is_empty() {
return merge(base_state, ours_text, theirs_text);
}
if ours_names != theirs_names {
eprintln!(
"[crdt] merge_by_component: structural divergence (ours components {ours_names:?} != theirs {theirs_names:?}); falling back to whole-doc merge"
);
return merge(base_state, ours_text, theirs_text);
}
let base_text = match base_state {
Some(bytes) => match CrdtDoc::decode_state(bytes) {
Ok(doc) => doc.to_text(),
Err(e) => {
eprintln!(
"[crdt] merge_by_component: failed to decode base state ({e}); merging nodes without a base"
);
String::new()
}
},
None => String::new(),
};
let base_nodes = segment_into_nodes(&base_text).unwrap_or_default();
let mut base_by_name: std::collections::HashMap<&str, &str> = std::collections::HashMap::new();
for node in &base_nodes {
if let Node::Component { name, text } = node {
base_by_name.entry(name.as_str()).or_insert(text.as_str());
}
}
let base_interstitials: Vec<&str> = base_nodes
.iter()
.filter_map(|c| match c {
Node::Interstitial(t) => Some(t.as_str()),
Node::Component { .. } => None,
})
.collect();
let merged_nodes = merge_aligned_nodes(&ours_nodes, &theirs_nodes, |name, idx| match name {
Some(n) => base_by_name.get(n).map(|s| s.to_string()),
None => base_interstitials.get(idx).map(|s| s.to_string()),
})?;
let merged: String = merged_nodes.into_iter().map(|(_, text)| text).collect();
let merged_names: Vec<String> = match segment_into_nodes(&merged) {
Ok(nodes) => nodes
.iter()
.filter_map(|n| n.component_name().map(str::to_string))
.collect(),
Err(e) => {
eprintln!(
"[crdt] merge_by_component: merged result failed to re-segment ({e}); falling back to whole-doc merge"
);
return merge(base_state, ours_text, theirs_text);
}
};
let names_match = merged_names.len() == ours_names.len()
&& merged_names
.iter()
.zip(ours_names.iter())
.all(|(m, o)| m.as_str() == *o);
if !names_match {
eprintln!(
"[crdt] merge_by_component: merged component framing diverged (ours {ours_names:?} != merged {merged_names:?}); falling back to whole-doc merge"
);
return merge(base_state, ours_text, theirs_text);
}
Ok(merged)
}
fn merge_aligned_nodes<F>(
ours_nodes: &[Node],
theirs_nodes: &[Node],
mut resolve_base: F,
) -> Result<Vec<(Option<String>, String)>>
where
F: FnMut(Option<&str>, usize) -> Option<String>,
{
let mut merged = Vec::with_capacity(ours_nodes.len());
let mut interstitial_idx = 0usize;
for (ours_node, theirs_node) in ours_nodes.iter().zip(theirs_nodes.iter()) {
let name = ours_node.component_name();
let node_base = resolve_base(name, interstitial_idx);
if name.is_none() {
interstitial_idx += 1;
}
let ours_slice = ours_node.text();
let theirs_slice = theirs_node.text();
let merged_text = if ours_slice == theirs_slice {
ours_slice.to_string()
} else if let Some(component_name) = name
&& let Some(merged) = reconcile_component(
component_name,
node_base.as_deref(),
ours_slice,
theirs_slice,
)
{
merged
} else {
let node_base_state = node_base
.as_deref()
.map(|t| CrdtDoc::from_text(t).encode_state());
merge(node_base_state.as_deref(), ours_slice, theirs_slice)?
};
merged.push((name.map(str::to_string), merged_text));
}
Ok(merged)
}
const PREAMBLE_KEY: &str = "\u{0}::preamble";
fn is_list_component(name: &str) -> bool {
matches!(name, "queue" | "backlog" | "review" | "done")
}
struct KeyedChild {
key: String,
text: String,
}
fn first_hash_id(s: &str) -> Option<String> {
let bytes = s.as_bytes();
let mut i = 0;
while i + 1 < bytes.len() {
if bytes[i] == b'[' && bytes[i + 1] == b'#' {
let rest = &s[i + 2..];
if let Some(close) = rest.find(']') {
let id = &rest[..close];
if !id.is_empty()
&& id
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
{
return Some(id.to_string());
}
}
}
i += 1;
}
None
}
fn normalize_item_text(item: &str) -> String {
let first = item.lines().next().unwrap_or("").trim();
let mut s = first.strip_prefix("- ").unwrap_or(first).trim_start();
for cb in ["[ ]", "[x]", "[X]", "[/]"] {
if let Some(r) = s.strip_prefix(cb) {
s = r.trim_start();
break;
}
}
s.replace("~~", "")
.replace(":pushpin:", "")
.replace('📌', "")
.split_whitespace()
.collect::<Vec<_>>()
.join(" ")
}
fn list_item_key(item: &str) -> String {
match first_hash_id(item) {
Some(id) => format!("id:{id}"),
None => format!("txt:{}", normalize_item_text(item)),
}
}
fn exchange_heading_key(block: &str) -> String {
let heading = block.lines().next().unwrap_or("").trim();
heading
.strip_suffix(" (HEAD)")
.unwrap_or(heading)
.trim_end()
.to_string()
}
fn split_list_children(body: &str) -> Option<Vec<KeyedChild>> {
let mut starts = Vec::new();
let mut pos = 0usize;
for line in body.split_inclusive('\n') {
let trimmed = line.trim_start();
if trimmed.starts_with("- ") || trimmed.trim_end() == "-" {
starts.push(pos);
}
pos += line.len();
}
if starts.is_empty() {
return None;
}
Some(segment_by_starts(body, &starts, list_item_key))
}
fn split_exchange_children(body: &str) -> Option<Vec<KeyedChild>> {
let mut starts = Vec::new();
let mut pos = 0usize;
for line in body.split_inclusive('\n') {
if line.trim_start().starts_with("### Re:") {
starts.push(pos);
}
pos += line.len();
}
if starts.is_empty() {
return None;
}
Some(segment_by_starts(body, &starts, exchange_heading_key))
}
fn segment_by_starts(body: &str, starts: &[usize], key_of: fn(&str) -> String) -> Vec<KeyedChild> {
let mut children = Vec::with_capacity(starts.len() + 1);
if starts[0] > 0 {
children.push(KeyedChild {
key: PREAMBLE_KEY.to_string(),
text: body[..starts[0]].to_string(),
});
}
for (i, &s) in starts.iter().enumerate() {
let e = starts.get(i + 1).copied().unwrap_or(body.len());
let slice = &body[s..e];
children.push(KeyedChild {
key: key_of(slice),
text: slice.to_string(),
});
}
children
}
fn keys_unique(children: &[KeyedChild]) -> bool {
let mut seen = std::collections::HashSet::new();
children.iter().all(|c| seen.insert(c.key.as_str()))
}
fn component_framing(text: &str) -> Option<(&str, &str, &str)> {
let comps = crate::component::parse(text).ok()?;
let comp = comps.iter().min_by_key(|c| c.open_start)?;
Some((
&text[..comp.open_end],
&text[comp.open_end..comp.close_start],
&text[comp.close_start..],
))
}
fn reconcile_component(
name: &str,
base_text: Option<&str>,
ours_text: &str,
theirs_text: &str,
) -> Option<String> {
if name != "exchange" && !is_list_component(name) {
return None;
}
let (ours_open, ours_body, ours_close) = component_framing(ours_text)?;
let (_, theirs_body, _) = component_framing(theirs_text)?;
let base_body = base_text.and_then(|t| component_framing(t).map(|(_, b, _)| b));
let merged_body = reconcile_component_body(name, base_body, ours_body, theirs_body)?;
Some(format!("{ours_open}{merged_body}{ours_close}"))
}
fn reconcile_component_body(
name: &str,
base_body: Option<&str>,
ours_body: &str,
theirs_body: &str,
) -> Option<String> {
let split = |b: &str| -> Option<Vec<KeyedChild>> {
if name == "exchange" {
split_exchange_children(b)
} else {
split_list_children(b)
}
};
let ours_children = split(ours_body)?;
let theirs_children = split(theirs_body)?;
let base_children = base_body.and_then(split).unwrap_or_default();
if !keys_unique(&ours_children)
|| !keys_unique(&theirs_children)
|| !keys_unique(&base_children)
{
return None;
}
let ours_map: std::collections::HashMap<&str, &KeyedChild> =
ours_children.iter().map(|c| (c.key.as_str(), c)).collect();
let theirs_map: std::collections::HashMap<&str, &KeyedChild> = theirs_children
.iter()
.map(|c| (c.key.as_str(), c))
.collect();
let base_map: std::collections::HashMap<&str, &KeyedChild> =
base_children.iter().map(|c| (c.key.as_str(), c)).collect();
let ours_keys: Vec<&str> = ours_children.iter().map(|c| c.key.as_str()).collect();
let theirs_keys: Vec<&str> = theirs_children.iter().map(|c| c.key.as_str()).collect();
let order = if name == "exchange" {
order_union(&ours_keys, &theirs_keys)
} else {
order_union(&theirs_keys, &ours_keys)
};
let protect_deletes = name == "exchange";
let mut out = String::new();
for key in &order {
let in_o = ours_map.get(key.as_str());
let in_t = theirs_map.get(key.as_str());
let in_b = base_map.get(key.as_str());
let resolved: Option<String> = match (in_o, in_t) {
(Some(o), Some(t)) => {
if o.text == t.text {
Some(o.text.clone())
} else {
let base_state = in_b.map(|b| CrdtDoc::from_text(&b.text).encode_state());
Some(merge(base_state.as_deref(), &o.text, &t.text).ok()?)
}
}
(Some(o), None) => match in_b {
None => Some(o.text.clone()),
Some(b) if protect_deletes || o.text != b.text => Some(o.text.clone()),
Some(_) => None,
},
(None, Some(t)) => match in_b {
None => Some(t.text.clone()),
Some(b) if protect_deletes || t.text != b.text => Some(t.text.clone()),
Some(_) => None,
},
(None, None) => None,
};
if let Some(text) = resolved {
out.push_str(&text);
}
}
Some(out)
}
fn order_union(ours_keys: &[&str], theirs_keys: &[&str]) -> Vec<String> {
let ours_set: std::collections::HashSet<&str> = ours_keys.iter().copied().collect();
let mut result: Vec<String> = ours_keys.iter().map(|s| s.to_string()).collect();
for (i, &k) in theirs_keys.iter().enumerate() {
if ours_set.contains(k) || result.iter().any(|r| r == k) {
continue;
}
let mut insert_pos = 0usize;
for j in (0..i).rev() {
if let Some(p) = result.iter().position(|r| r == theirs_keys[j]) {
insert_pos = p + 1;
break;
}
}
result.insert(insert_pos, k.to_string());
}
result
}
fn encode_text_deterministic(text: &str) -> Vec<u8> {
let doc = Doc::with_client_id(0);
let yrs_text = doc.get_or_insert_text(TEXT_KEY);
{
let mut txn = doc.transact_mut();
yrs_text.insert(&mut txn, 0, text);
}
let txn = doc.transact();
txn.encode_state_as_update_v1(&yrs::StateVector::default())
}
const MULTINODE_MAGIC: &[u8; 4] = b"ADN1";
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct MultiNodeState {
nodes: Vec<PersistedNode>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct PersistedNode {
name: Option<String>,
state: Vec<u8>,
}
impl MultiNodeState {
pub fn from_text(text: &str) -> Result<Self> {
let nodes = segment_into_nodes(text)?
.iter()
.map(|node| PersistedNode {
name: node.component_name().map(str::to_string),
state: encode_text_deterministic(node.text()),
})
.collect();
Ok(MultiNodeState { nodes })
}
fn from_merged_nodes(merged: &[(Option<String>, String)]) -> Self {
MultiNodeState {
nodes: merged
.iter()
.map(|(name, text)| PersistedNode {
name: name.clone(),
state: encode_text_deterministic(text),
})
.collect(),
}
}
fn node_text(node: &PersistedNode) -> Result<String> {
Ok(CrdtDoc::decode_state(&node.state)?.to_text())
}
pub fn to_text(&self) -> Result<String> {
let mut text = String::new();
for node in &self.nodes {
text.push_str(&Self::node_text(node)?);
}
Ok(text)
}
fn base_lookup(&self) -> Result<(std::collections::HashMap<String, String>, Vec<String>)> {
let mut by_name = std::collections::HashMap::new();
let mut interstitials = Vec::new();
for node in &self.nodes {
let text = Self::node_text(node)?;
match &node.name {
Some(name) => {
by_name.entry(name.clone()).or_insert(text);
}
None => interstitials.push(text),
}
}
Ok((by_name, interstitials))
}
pub fn encode(&self) -> Vec<u8> {
let mut out = Vec::new();
out.extend_from_slice(MULTINODE_MAGIC);
out.push(1u8); out.extend_from_slice(&(self.nodes.len() as u32).to_le_bytes());
for node in &self.nodes {
match &node.name {
Some(name) => {
let bytes = name.as_bytes();
out.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
out.extend_from_slice(bytes);
}
None => out.extend_from_slice(&u32::MAX.to_le_bytes()),
}
out.extend_from_slice(&(node.state.len() as u32).to_le_bytes());
out.extend_from_slice(&node.state);
}
out
}
pub fn decode(bytes: &[u8]) -> Result<Self> {
let mut cursor = 0usize;
let take = |cursor: &mut usize, n: usize| -> Result<&[u8]> {
let end = cursor
.checked_add(n)
.filter(|end| *end <= bytes.len())
.ok_or_else(|| anyhow::anyhow!("multinode state truncated"))?;
let slice = &bytes[*cursor..end];
*cursor = end;
Ok(slice)
};
let read_u32 = |cursor: &mut usize| -> Result<u32> {
let b = take(cursor, 4)?;
Ok(u32::from_le_bytes([b[0], b[1], b[2], b[3]]))
};
if take(&mut cursor, 4)? != MULTINODE_MAGIC {
anyhow::bail!("not a multinode state container (magic mismatch)");
}
let version = take(&mut cursor, 1)?[0];
if version != 1 {
anyhow::bail!("unsupported multinode state version {version}");
}
let count = read_u32(&mut cursor)? as usize;
let mut nodes = Vec::with_capacity(count);
for _ in 0..count {
let name_len = read_u32(&mut cursor)?;
let name = if name_len == u32::MAX {
None
} else {
let bytes = take(&mut cursor, name_len as usize)?;
Some(
std::str::from_utf8(bytes)
.map_err(|e| anyhow::anyhow!("invalid node name UTF-8: {e}"))?
.to_string(),
)
};
let state_len = read_u32(&mut cursor)? as usize;
let state = take(&mut cursor, state_len)?.to_vec();
nodes.push(PersistedNode { name, state });
}
Ok(MultiNodeState { nodes })
}
pub fn decode_or_migrate(bytes: &[u8], fallback_text: &str) -> Self {
match Self::decode(bytes) {
Ok(state) => return state,
Err(_) => {
if let Ok(doc) = CrdtDoc::decode_state(bytes)
&& let Ok(state) = Self::from_text(&doc.to_text())
{
eprintln!(
"[crdt] MultiNodeState: migrated legacy whole-doc state into per-node state"
);
return state;
}
}
}
eprintln!(
"[crdt] MultiNodeState: state unreadable; rebuilding per-node state from fallback text"
);
Self::from_text(fallback_text).unwrap_or_default()
}
pub fn merge(
base: Option<&MultiNodeState>,
ours_text: &str,
theirs_text: &str,
) -> Result<(String, MultiNodeState)> {
let whole_doc_fallback = |reason: &str| -> Result<(String, MultiNodeState)> {
let base_text = match base {
Some(b) => b.to_text().unwrap_or_default(),
None => String::new(),
};
let base_state = base.map(|_| CrdtDoc::from_text(&base_text).encode_state());
if !reason.is_empty() {
eprintln!(
"[crdt] MultiNodeState::merge: {reason}; falling back to whole-doc merge"
);
}
let merged = merge(base_state.as_deref(), ours_text, theirs_text)?;
let state = MultiNodeState::from_text(&merged)?;
Ok((merged, state))
};
if ours_text == theirs_text {
return Ok((ours_text.to_string(), MultiNodeState::from_text(ours_text)?));
}
let ours_nodes = match segment_into_nodes(ours_text) {
Ok(c) => c,
Err(e) => return whole_doc_fallback(&format!("failed to segment ours ({e})")),
};
let theirs_nodes = match segment_into_nodes(theirs_text) {
Ok(c) => c,
Err(e) => return whole_doc_fallback(&format!("failed to segment theirs ({e})")),
};
let ours_names: Vec<&str> = ours_nodes.iter().filter_map(Node::component_name).collect();
let theirs_names: Vec<&str> = theirs_nodes
.iter()
.filter_map(Node::component_name)
.collect();
if ours_names.is_empty() && theirs_names.is_empty() {
return whole_doc_fallback("");
}
if ours_names != theirs_names {
return whole_doc_fallback(&format!(
"structural divergence (ours {ours_names:?} != theirs {theirs_names:?})"
));
}
let (base_by_name, base_interstitials) = match base {
Some(b) => b.base_lookup()?,
None => (std::collections::HashMap::new(), Vec::new()),
};
let merged_nodes =
merge_aligned_nodes(&ours_nodes, &theirs_nodes, |name, idx| match name {
Some(n) => base_by_name.get(n).cloned(),
None => base_interstitials.get(idx).cloned(),
})?;
let merged_text: String = merged_nodes.iter().map(|(_, text)| text.as_str()).collect();
let new_state = MultiNodeState::from_merged_nodes(&merged_nodes);
Ok((merged_text, new_state))
}
}
fn is_transient_merge_line(line: &str) -> bool {
let t = line.trim();
t.starts_with("<!-- agent:boundary:")
}
fn response_headings(text: &str) -> Vec<String> {
text.lines()
.map(str::trim)
.filter(|l| l.starts_with("### Re:"))
.map(|l| {
l.strip_suffix(" (HEAD)")
.unwrap_or(l)
.trim_end()
.to_string()
})
.collect()
}
fn guard_committed_responses(
merged: String,
committed_response_headings: &[String],
ours_text: &str,
theirs_text: &str,
) -> String {
if committed_response_headings.is_empty() {
return merged;
}
let ours_headings = response_headings(ours_text);
let merged_headings = response_headings(&merged);
let lost: Vec<&String> = committed_response_headings
.iter()
.filter(|h| ours_headings.contains(h) && !merged_headings.contains(h))
.collect();
if lost.is_empty() {
return merged;
}
let ours_lines: std::collections::HashSet<&str> = ours_text
.lines()
.map(str::trim)
.filter(|l| !l.is_empty())
.collect();
let theirs_adds_new = theirs_text
.lines()
.map(str::trim)
.any(|l| !l.is_empty() && !is_transient_merge_line(l) && !ours_lines.contains(l));
if theirs_adds_new {
eprintln!(
"[crdt] WARNING: merge dropped committed response heading(s) {:?} while theirs added new content; keeping merge (#ipc-crdt-response-drift)",
lost
);
merged
} else {
eprintln!(
"[crdt] stale theirs dropped committed response heading(s) {:?}; preferring ours (#ipc-crdt-response-drift)",
lost
);
ours_text.to_string()
}
}
pub fn dedup_adjacent_blocks(text: &str) -> String {
let blocks: Vec<&str> = text.split("\n\n").collect();
if blocks.len() < 2 {
return text.to_string();
}
let mut result: Vec<&str> = Vec::with_capacity(blocks.len());
for block in &blocks {
let trimmed = block.trim();
let non_empty_lines = trimmed.lines().filter(|l| !l.trim().is_empty()).count();
if non_empty_lines >= 1
&& !is_structural_block(trimmed)
&& let Some(prev) = result.last()
&& prev.trim() == trimmed
{
eprintln!(
"[crdt] dedup: removed duplicate block ({} lines)",
non_empty_lines
);
continue;
}
result.push(*block);
}
result.join("\n\n")
}
fn is_structural_block(trimmed: &str) -> bool {
let mut lines = trimmed.lines();
let first = match lines.next() {
Some(l) => l.trim(),
None => return false,
};
if lines.next().is_some() {
return false;
}
first.len() >= 3 && first.chars().all(|c| c == '-' || c == '*' || c == '_')
}
pub fn compact(state: &[u8]) -> Result<Vec<u8>> {
let doc = CrdtDoc::decode_state(state)?;
Ok(doc.encode_state())
}
fn common_prefix_len(a: &str, b: &str) -> usize {
a.bytes().zip(b.bytes()).take_while(|(x, y)| x == y).count()
}
fn common_suffix_len(a: &str, b: &str) -> usize {
a.bytes()
.rev()
.zip(b.bytes().rev())
.take_while(|(x, y)| x == y)
.count()
}
#[derive(Debug)]
enum EditOp {
Retain(u32),
Delete(u32),
Insert(String),
}
fn compute_edit_ops(from: &str, to: &str) -> Vec<EditOp> {
use similar::{ChangeTag, TextDiff};
let diff = TextDiff::from_lines(from, to);
let mut ops = Vec::new();
for change in diff.iter_all_changes() {
match change.tag() {
ChangeTag::Equal => {
let len = change.value().len() as u32;
if let Some(EditOp::Retain(n)) = ops.last_mut() {
*n += len;
} else {
ops.push(EditOp::Retain(len));
}
}
ChangeTag::Delete => {
let len = change.value().len() as u32;
if let Some(EditOp::Delete(n)) = ops.last_mut() {
*n += len;
} else {
ops.push(EditOp::Delete(len));
}
}
ChangeTag::Insert => {
let s = change.value().to_string();
if let Some(EditOp::Insert(existing)) = ops.last_mut() {
existing.push_str(&s);
} else {
ops.push(EditOp::Insert(s));
}
}
}
}
ops
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(tag = "kind", rename_all = "lowercase")]
pub enum EditorOp {
Insert { offset: usize, text: String },
Delete { offset: usize, len: usize },
}
pub fn replay_editor_ops(base: &str, ops: &[EditorOp]) -> Option<String> {
let mut buf = base.to_string();
for op in ops {
match op {
EditorOp::Insert { offset, text } => {
if *offset > buf.len() || !buf.is_char_boundary(*offset) {
return None;
}
buf.insert_str(*offset, text);
}
EditorOp::Delete { offset, len } => {
let end = offset.checked_add(*len)?;
if end > buf.len() || !buf.is_char_boundary(*offset) || !buf.is_char_boundary(end) {
return None;
}
buf.replace_range(*offset..end, "");
}
}
}
Some(buf)
}
fn apply_editor_ops(text: &TextRef, txn: &mut yrs::TransactionMut<'_>, ops: &[EditorOp]) {
for op in ops {
match op {
EditorOp::Insert { offset, text: s } => {
text.insert(txn, *offset as u32, s);
}
EditorOp::Delete { offset, len } => {
text.remove_range(txn, *offset as u32, *len as u32);
}
}
}
}
fn apply_ops(text: &TextRef, txn: &mut yrs::TransactionMut<'_>, ops: &[EditOp]) {
let mut cursor: u32 = 0;
for op in ops {
match op {
EditOp::Retain(n) => cursor += n,
EditOp::Delete(n) => {
text.remove_range(txn, cursor, *n);
}
EditOp::Insert(s) => {
text.insert(txn, cursor, s);
cursor += s.len() as u32;
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn roundtrip_text() {
let content = "Hello, world!\nLine two.\n";
let doc = CrdtDoc::from_text(content);
assert_eq!(doc.to_text(), content);
}
#[test]
fn replay_editor_ops_reconstructs_insert() {
let base = "hello world\n";
let ops = vec![EditorOp::Insert {
offset: 5,
text: "!".to_string(),
}];
assert_eq!(
replay_editor_ops(base, &ops).as_deref(),
Some("hello! world\n")
);
}
#[test]
fn replay_editor_ops_reconstructs_delete_and_replace_sequence() {
let base = "- do [#foo]\n";
let ops = vec![
EditorOp::Delete { offset: 7, len: 3 },
EditorOp::Insert {
offset: 7,
text: "foobar".to_string(),
},
];
assert_eq!(
replay_editor_ops(base, &ops).as_deref(),
Some("- do [#foobar]\n")
);
}
#[test]
fn replay_editor_ops_sequential_offsets_are_post_prior_op() {
let base = "ab";
let ops = vec![
EditorOp::Insert {
offset: 0,
text: "X".to_string(),
},
EditorOp::Insert {
offset: 2,
text: "Y".to_string(),
},
];
assert_eq!(replay_editor_ops(base, &ops).as_deref(), Some("XaYb"));
}
#[test]
fn replay_editor_ops_out_of_bounds_is_none() {
let base = "short";
let insert = vec![EditorOp::Insert {
offset: 99,
text: "x".to_string(),
}];
assert_eq!(replay_editor_ops(base, &insert), None);
let delete = vec![EditorOp::Delete { offset: 3, len: 99 }];
assert_eq!(replay_editor_ops(base, &delete), None);
}
#[test]
fn replay_editor_ops_non_char_boundary_is_none() {
let base = "é";
let ops = vec![EditorOp::Insert {
offset: 1,
text: "x".to_string(),
}];
assert_eq!(replay_editor_ops(base, &ops), None);
}
#[test]
fn editor_op_json_roundtrip() {
let ops = vec![
EditorOp::Insert {
offset: 3,
text: "hi".to_string(),
},
EditorOp::Delete { offset: 0, len: 2 },
];
let json = serde_json::to_string(&ops).unwrap();
let back: Vec<EditorOp> = serde_json::from_str(&json).unwrap();
assert_eq!(ops, back);
}
#[test]
fn merge_with_editor_ops_gate_passes_and_preserves_user_edit() {
let base = "<!-- agent:queue -->\n- do [#foo]\n<!-- /agent:queue -->\n";
let ours = "<!-- agent:queue -->\n- do [#foo]\n<!-- /agent:queue -->\nAGENT RESPONSE\n";
let theirs = "<!-- agent:queue -->\n- do [#foobar]\n<!-- /agent:queue -->\n";
let ops = vec![
EditorOp::Delete { offset: 28, len: 3 },
EditorOp::Insert {
offset: 28,
text: "foobar".to_string(),
},
];
assert_eq!(replay_editor_ops(base, &ops).as_deref(), Some(theirs));
let base_state = CrdtDoc::from_text(base).encode_state();
let merged = merge_with_editor_ops(Some(&base_state), ours, theirs, Some(&ops)).unwrap();
assert!(merged.contains("- do [#foobar]"), "merged: {merged}");
assert!(merged.contains("AGENT RESPONSE"), "merged: {merged}");
assert!(
!merged.contains("- do [#foo]\n"),
"stale item leaked: {merged}"
);
assert_eq!(
merged.matches("do [#foobar]").count(),
1,
"duplicated: {merged}"
);
}
#[test]
fn merge_with_editor_ops_stale_ops_fall_back_to_diff() {
let base = "Line A\nLine B\n";
let ours = "Line A\nLine B\nAgent line\n";
let theirs = "Line A\nLine B edited\n";
let bogus = vec![EditorOp::Insert {
offset: 0,
text: "ZZZ".to_string(),
}];
assert_ne!(replay_editor_ops(base, &bogus).as_deref(), Some(theirs));
let base_state = CrdtDoc::from_text(base).encode_state();
let with_ops =
merge_with_editor_ops(Some(&base_state), ours, theirs, Some(&bogus)).unwrap();
let diff_only = merge(Some(&base_state), ours, theirs).unwrap();
assert_eq!(
with_ops, diff_only,
"stale ops must fall back to the diff-guess result"
);
assert!(!with_ops.contains("ZZZ"), "stale op leaked: {with_ops}");
}
#[test]
fn merge_with_editor_ops_none_matches_plain_merge() {
let base = "header\n\nbody\n";
let ours = "header\n\nbody\nagent\n";
let theirs = "header\n\nbody edited\n";
let base_state = CrdtDoc::from_text(base).encode_state();
assert_eq!(
merge_with_editor_ops(Some(&base_state), ours, theirs, None).unwrap(),
merge(Some(&base_state), ours, theirs).unwrap()
);
}
#[test]
fn merge_with_editor_ops_empty_ops_fall_back() {
let base = "x\ny\n";
let ours = "x\ny\nagent\n";
let theirs = "x\ny edited\n";
let base_state = CrdtDoc::from_text(base).encode_state();
let empty: Vec<EditorOp> = vec![];
assert_eq!(
merge_with_editor_ops(Some(&base_state), ours, theirs, Some(&empty)).unwrap(),
merge(Some(&base_state), ours, theirs).unwrap()
);
}
#[test]
fn roundtrip_encode_decode() {
let content = "Some document content.\n";
let doc = CrdtDoc::from_text(content);
let encoded = doc.encode_state();
let decoded = CrdtDoc::decode_state(&encoded).unwrap();
assert_eq!(decoded.to_text(), content);
}
#[test]
fn apply_edit_insert() {
let doc = CrdtDoc::from_text("Hello world");
doc.apply_edit(5, 0, ",");
assert_eq!(doc.to_text(), "Hello, world");
}
#[test]
fn apply_edit_delete() {
let doc = CrdtDoc::from_text("Hello, world");
doc.apply_edit(5, 1, "");
assert_eq!(doc.to_text(), "Hello world");
}
#[test]
fn apply_edit_replace() {
let doc = CrdtDoc::from_text("Hello world");
doc.apply_edit(6, 5, "Rust");
assert_eq!(doc.to_text(), "Hello Rust");
}
#[test]
fn concurrent_append_merge_no_conflict() {
let base = "# Document\n\nBase content.\n";
let base_doc = CrdtDoc::from_text(base);
let base_state = base_doc.encode_state();
let ours = format!("{base}## Agent\n\nAgent response.\n");
let theirs = format!("{base}## User\n\nUser addition.\n");
let merged = merge(Some(&base_state), &ours, &theirs).unwrap();
assert!(merged.contains("Agent response."), "missing agent text");
assert!(merged.contains("User addition."), "missing user text");
assert!(merged.contains("Base content."), "missing base text");
assert!(!merged.contains("<<<<<<<"));
assert!(!merged.contains(">>>>>>>"));
}
#[test]
fn concurrent_insert_same_position() {
let base = "Line 1\nLine 3\n";
let base_doc = CrdtDoc::from_text(base);
let base_state = base_doc.encode_state();
let ours = "Line 1\nAgent line\nLine 3\n";
let theirs = "Line 1\nUser line\nLine 3\n";
let merged = merge(Some(&base_state), ours, theirs).unwrap();
assert!(merged.contains("Agent line"), "missing agent insertion");
assert!(merged.contains("User line"), "missing user insertion");
assert!(merged.contains("Line 1"), "missing line 1");
assert!(merged.contains("Line 3"), "missing line 3");
}
#[test]
fn merge_no_base_state() {
let ours = "Agent wrote this.\n";
let theirs = "User wrote this.\n";
let merged = merge(None, ours, theirs).unwrap();
assert!(merged.contains("Agent wrote this."));
assert!(merged.contains("User wrote this."));
}
#[test]
fn compact_preserves_content() {
let doc = CrdtDoc::from_text("Hello");
doc.apply_edit(5, 0, " world");
doc.apply_edit(11, 0, "!");
let state = doc.encode_state();
let compacted = compact(&state).unwrap();
let restored = CrdtDoc::decode_state(&compacted).unwrap();
assert_eq!(restored.to_text(), "Hello world!");
assert!(compacted.len() <= state.len());
}
#[test]
fn compact_reduces_size_after_edits() {
let doc = CrdtDoc::from_text("aaaa");
for i in 0..20 {
let c = ((b'a' + (i % 26)) as char).to_string();
doc.apply_edit(0, 1, &c);
}
let state = doc.encode_state();
let compacted = compact(&state).unwrap();
let restored = CrdtDoc::decode_state(&compacted).unwrap();
assert_eq!(restored.to_text(), doc.to_text());
}
#[test]
fn empty_document() {
let doc = CrdtDoc::from_text("");
assert_eq!(doc.to_text(), "");
let encoded = doc.encode_state();
let decoded = CrdtDoc::decode_state(&encoded).unwrap();
assert_eq!(decoded.to_text(), "");
}
#[test]
fn decode_invalid_bytes_errors() {
let result = CrdtDoc::decode_state(&[0xff, 0xfe, 0xfd]);
assert!(result.is_err());
}
#[test]
fn merge_identical_texts() {
let base = "Same content.\n";
let base_doc = CrdtDoc::from_text(base);
let state = base_doc.encode_state();
let merged = merge(Some(&state), base, base).unwrap();
assert_eq!(merged, base);
}
#[test]
fn merge_one_side_unchanged() {
let base = "Original.\n";
let base_doc = CrdtDoc::from_text(base);
let state = base_doc.encode_state();
let ours = "Original.\nAgent added.\n";
let merged = merge(Some(&state), ours, base).unwrap();
assert_eq!(merged, ours);
}
#[test]
fn merge_stale_base_no_duplicate_user_prompt() {
let base_content = "\
## Assistant
Previous response content.
Committed and pushed.
";
let base_doc = CrdtDoc::from_text(base_content);
let base_state = base_doc.encode_state();
let user_prompt = "\
Opening a video a shows video a.
Closing video a then opening video b start video b but video b is hidden.
Closing video b then reopening video b starts and shows video b. video b is visible.
";
let ours = format!(
"\
{}{}### Re: Close A → Open B still hidden
Added explicit height and visibility reset.
Committed and pushed.
",
base_content, user_prompt
);
let theirs = format!(
"\
{}{}
",
base_content, user_prompt
);
let merged = merge(Some(&base_state), &ours, &theirs).unwrap();
let prompt_count = merged.matches("Opening a video a shows video a.").count();
assert_eq!(
prompt_count, 1,
"User prompt duplicated! Appeared {} times in:\n{}",
prompt_count, merged
);
assert!(
merged.contains("### Re: Close A → Open B still hidden"),
"Agent response missing from merge:\n{}",
merged
);
}
#[test]
fn merge_stale_base_same_insertion_both_sides() {
let base_content = "Line 1\nLine 2\n";
let base_doc = CrdtDoc::from_text(base_content);
let base_state = base_doc.encode_state();
let shared_addition = "User typed this.\n";
let ours = format!("{}{}Agent response.\n", base_content, shared_addition);
let theirs = format!("{}{}", base_content, shared_addition);
let merged = merge(Some(&base_state), &ours, &theirs).unwrap();
let count = merged.matches("User typed this.").count();
assert_eq!(
count, 1,
"Shared text duplicated! Appeared {} times in:\n{}",
count, merged
);
assert!(
merged.contains("Agent response."),
"Agent text missing:\n{}",
merged
);
}
#[test]
fn merge_no_character_interleaving() {
let base = "# Doc\n\nPrevious content.\n\n";
let base_doc = CrdtDoc::from_text(base);
let base_state = base_doc.encode_state();
let ours = "# Doc\n\nPrevious content.\n\n*Compacted. Content archived to*\n";
let theirs = "# Doc\n\nPrevious content.\n\n**Soft-bristle brush only**\n";
let merged = merge(Some(&base_state), ours, theirs).unwrap();
assert!(
merged.contains("*Compacted. Content archived to*"),
"Agent text should be contiguous (not interleaved). Got:\n{}",
merged
);
assert!(
merged.contains("**Soft-bristle brush only**"),
"User text should be contiguous (not interleaved). Got:\n{}",
merged
);
}
#[test]
fn merge_concurrent_same_line_no_garbling() {
let base = "Some base text\n";
let base_doc = CrdtDoc::from_text(base);
let base_state = base_doc.encode_state();
let ours = "Agent wrote this line\n";
let theirs = "User wrote different text\n";
let merged = merge(Some(&base_state), ours, theirs).unwrap();
let has_agent_contiguous = merged.contains("Agent wrote this line");
let has_user_contiguous = merged.contains("User wrote different text");
assert!(
has_agent_contiguous || has_user_contiguous,
"At least one side should have contiguous text (no char interleaving). Got:\n{}",
merged
);
}
#[test]
fn merge_replace_vs_append_no_interleaving() {
let header = "---\nagent_doc_format: template\n---\n\n# Title\n\n<!-- agent:exchange -->\n";
let footer = "\n<!-- /agent:exchange -->\n";
let old_exchange = "\
### Committed, Pushed & Released
**project (v0.1.0):**
- Committed initial implementation
- Tagged v0.1.0 and pushed
Add a README.md to the project.
Also add AGENTS.md with a symlink CLAUDE.md
**sub-project:**
- Committed fix + SPEC.md
- Pushed to remote
";
let stale_base = format!("{header}{old_exchange}{footer}");
let stale_state = CrdtDoc::from_text(&stale_base).encode_state();
let _baseline = stale_base.clone();
let agent_exchange = "\
### Done
Added to project and pushed:
- **README.md** — overview, usage, design notes
- **AGENTS.md** — architecture, key decisions, commands, related projects
- **CLAUDE.md** → symlink to AGENTS.md
All committed and pushed.
";
let ours = format!("{header}{agent_exchange}{footer}");
let theirs_exchange = "\
### Committed, Pushed & Released
**project (v0.1.0):**
- Committed initial implementation
- Tagged v0.1.0 and pushed
Add a README.md to the project.
Also add AGENTS.md with a symlink CLAUDE.md
Please add tests.
Please comprehensively test adherence to the spec.
**sub-project:**
- Committed fix + SPEC.md
- Pushed to remote
";
let theirs = format!("{header}{theirs_exchange}{footer}");
let merged = merge(Some(&stale_state), &ours, &theirs).unwrap();
assert!(
merged.contains(
"- **AGENTS.md** — architecture, key decisions, commands, related projects"
),
"Agent text garbled (mid-word split). Got:\n{}",
merged
);
assert!(
merged.contains("Please add tests."),
"User addition missing. Got:\n{}",
merged
);
assert!(
!merged.contains("key deAdd") && !merged.contains("key de\n"),
"Old content interleaved into agent text. Got:\n{}",
merged
);
}
#[test]
fn merge_replace_vs_append_with_baseline_base() {
let header = "---\nagent_doc_format: template\n---\n\n# Title\n\n<!-- agent:exchange -->\n";
let footer = "\n<!-- /agent:exchange -->\n";
let old_exchange = "\
### Previous Response
Old content here.
Add a README.md to the project.
Also add AGENTS.md with a symlink CLAUDE.md
";
let baseline = format!("{header}{old_exchange}{footer}");
let agent_exchange = "\
### Done
- **README.md** — overview, usage, design notes
- **AGENTS.md** — architecture, key decisions, commands, related projects
- **CLAUDE.md** → symlink to AGENTS.md
All committed and pushed.
";
let ours = format!("{header}{agent_exchange}{footer}");
let user_addition = "\nPlease add tests.\n";
let theirs = format!("{header}{old_exchange}{user_addition}{footer}");
let baseline_state = CrdtDoc::from_text(&baseline).encode_state();
let merged = merge(Some(&baseline_state), &ours, &theirs).unwrap();
assert!(
merged.contains("key decisions, commands, related projects"),
"Agent text garbled. Got:\n{}",
merged
);
assert!(
merged.contains("Please add tests."),
"User addition missing. Got:\n{}",
merged
);
}
#[test]
fn merge_streaming_concurrent_edit_preserves_formatting() {
let base = "commit and push all rappstack packages.\n\n";
let base_doc = CrdtDoc::from_text(base);
let base_state = base_doc.encode_state();
let ours = "\
commit and push all rappstack packages.
### Re: commit and push
*Compacted. Content archived to `docs/`*
Done — all packages pushed.
";
let theirs = "\
commit and push all rappstack packages.
**Soft-bristle brush only**
";
let merged = merge(Some(&base_state), ours, theirs).unwrap();
assert!(
merged.contains("*Compacted. Content archived to `docs/`*"),
"Agent formatting broken. Got:\n{}",
merged
);
assert!(
merged.contains("**Soft-bristle brush only**"),
"User formatting broken. Got:\n{}",
merged
);
assert!(
!merged.contains("*C*C") && !merged.contains("**Sot"),
"Character interleaving detected. Got:\n{}",
merged
);
}
#[test]
fn merge_replace_vs_insert_no_interleaving() {
let header = "---\nagent_doc_format: template\nagent_doc_write: crdt\n---\n\n# Document Title\n\nSome preamble text that both sides share.\nThis provides enough common prefix to avoid stale detection.\n\n<!-- agent:exchange -->\n";
let footer = "<!-- /agent:exchange -->\n";
let old_exchange =
"Line one of old content\nLine two of old content\nLine three of old content\n";
let baseline = format!("{header}{old_exchange}{footer}");
let baseline_doc = CrdtDoc::from_text(&baseline);
let baseline_state = baseline_doc.encode_state();
let agent_exchange = "Completely new line one\nCompletely new line two\nCompletely new line three\nCompletely new line four\n";
let ours = format!("{header}{agent_exchange}{footer}");
let theirs = format!(
"{header}Line one of old content\nUser inserted this line\nLine two of old content\nLine three of old content\n{footer}"
);
let merged = merge(Some(&baseline_state), &ours, &theirs).unwrap();
assert!(
merged.contains("Completely new line one"),
"Agent line 1 missing or garbled. Got:\n{}",
merged
);
assert!(
merged.contains("Completely new line two"),
"Agent line 2 missing or garbled. Got:\n{}",
merged
);
assert!(
merged.contains("User inserted this line"),
"User insertion missing. Got:\n{}",
merged
);
assert!(
!merged.contains("CompleteUser") && !merged.contains("Complete\nUser"),
"Character interleaving detected. Got:\n{}",
merged
);
}
#[test]
fn reorder_agent_before_human_at_append_boundary() {
let base = "# Document\n\nBase content.\n";
let base_doc = CrdtDoc::from_text(base);
let base_state = base_doc.encode_state();
let ours = format!("{base}### Agent Response\n\nAgent wrote this.\n");
let theirs = format!("{base}User added this line.\n");
let merged = merge(Some(&base_state), &ours, &theirs).unwrap();
assert!(merged.contains("Agent wrote this."), "missing agent text");
assert!(
merged.contains("User added this line."),
"missing user text"
);
assert!(merged.contains("Base content."), "missing base text");
let agent_pos = merged.find("Agent wrote this.").unwrap();
let human_pos = merged.find("User added this line.").unwrap();
assert!(
agent_pos < human_pos,
"Agent content should appear before human content.\nAgent pos: {}, Human pos: {}\nMerged:\n{}",
agent_pos,
human_pos,
merged
);
}
#[test]
fn dedup_removes_identical_adjacent_blocks() {
let text =
"### Re: Question\nAnswer here.\n\n### Re: Question\nAnswer here.\n\nDifferent block.";
let result = dedup_adjacent_blocks(text);
assert_eq!(result.matches("### Re: Question").count(), 1);
assert!(result.contains("Different block."));
}
#[test]
fn dedup_preserves_different_adjacent_blocks() {
let text = "### Re: First\nAnswer one.\n\n### Re: Second\nAnswer two.";
let result = dedup_adjacent_blocks(text);
assert!(result.contains("### Re: First"));
assert!(result.contains("### Re: Second"));
}
#[test]
fn dedup_preserves_structural_thematic_breaks() {
let text = "---\n\n---\n\nContent.";
let result = dedup_adjacent_blocks(text);
assert_eq!(result, text);
let text2 = "***\n\n***\n\nContent.";
assert_eq!(dedup_adjacent_blocks(text2), text2);
let text3 = "___\n\n___\n\nContent.";
assert_eq!(dedup_adjacent_blocks(text3), text3);
}
#[test]
fn dedup_removes_single_line_content_duplicates() {
let text = "### Re: topic — opus-4-6\n\n### Re: topic — opus-4-6\n\nDifferent.";
let result = dedup_adjacent_blocks(text);
assert_eq!(result.matches("### Re: topic").count(), 1);
assert!(result.contains("Different."));
}
#[test]
fn dedup_removes_single_line_text_duplicates() {
let text = "Some response text.\n\nSome response text.\n\nNext section.";
let result = dedup_adjacent_blocks(text);
assert_eq!(result.matches("Some response text.").count(), 1);
assert!(result.contains("Next section."));
}
#[test]
fn dedup_preserves_different_single_line_blocks() {
let text = "Line A.\n\nLine B.\n\nLine C.";
let result = dedup_adjacent_blocks(text);
assert_eq!(result, text);
}
#[test]
fn dedup_handles_empty_text() {
assert_eq!(dedup_adjacent_blocks(""), "");
}
#[test]
fn dedup_no_change_when_no_duplicates() {
let text = "Block A\nLine 2.\n\nBlock B\nLine 2.";
let result = dedup_adjacent_blocks(text);
assert_eq!(result, text);
}
#[test]
fn is_structural_block_detects_thematic_breaks() {
assert!(is_structural_block("---"));
assert!(is_structural_block("***"));
assert!(is_structural_block("___"));
assert!(is_structural_block("-----"));
assert!(!is_structural_block("--"));
assert!(!is_structural_block("### Heading"));
assert!(!is_structural_block("some text"));
assert!(!is_structural_block("---\nmore"));
}
#[test]
fn merge_empty_exchange_no_duplicate_user_line() {
let header = "---\nagent_doc_format: template\nagent_doc_write: crdt\n---\n\n## Exchange\n\n<!-- agent:exchange patch=append -->\n";
let footer = "<!-- /agent:exchange -->\n\n## Pending\n\n<!-- agent:pending -->\n<!-- /agent:pending -->\n";
let base = format!("{header}\n<!-- agent:boundary:base-id -->\n{footer}");
let ours = format!(
"{header}\n### Re: topic — opus-4-6\n\nResponse content here.\n\nMore response text.\n\n<!-- agent:boundary:new-id -->\n{footer}"
);
let theirs = format!(
"{header}User prompt line one.\nUser prompt line two.\nLast line of user prompt.\n<!-- agent:boundary:base-id -->\n{footer}"
);
let base_doc = CrdtDoc::from_text(&base);
let base_state = base_doc.encode_state();
let merged = merge(Some(&base_state), &ours, &theirs).unwrap();
let count1 = merged.matches("User prompt line one.").count();
assert_eq!(
count1, 1,
"Line one duplicated (count={}). Merged:\n{}",
count1, merged
);
let count2 = merged.matches("User prompt line two.").count();
assert_eq!(
count2, 1,
"Line two duplicated (count={}). Merged:\n{}",
count2, merged
);
let count3 = merged.matches("Last line of user prompt.").count();
assert_eq!(
count3, 1,
"Last line duplicated (count={}). Merged:\n{}",
count3, merged
);
assert!(
merged.contains("### Re: topic — opus-4-6"),
"Response heading missing:\n{}",
merged
);
assert!(
merged.contains("Response content here."),
"Response content missing:\n{}",
merged
);
let prompt_pos = merged.find("User prompt line one.").unwrap();
let response_pos = merged.find("### Re: topic").unwrap();
assert!(
prompt_pos < response_pos,
"User prompt should precede response. prompt={} response={}\nMerged:\n{}",
prompt_pos,
response_pos,
merged
);
}
#[test]
fn merge_empty_exchange_prefixed_prompts_no_duplicate() {
let header = "---\nagent_doc_format: template\nagent_doc_write: crdt\n---\n\n## Exchange\n\n<!-- agent:exchange patch=append -->\n";
let footer = "<!-- /agent:exchange -->\n";
let base = format!("{header}\n<!-- agent:boundary:base-id -->\n{footer}");
let ours = format!(
"{header}❯ Review the code\n❯ What does it do?\n\n### Re: code review — opus-4-6\n\nThe code does X.\n\n<!-- agent:boundary:new-id -->\n{footer}"
);
let theirs = format!(
"{header}❯ Review the code\n❯ What does it do?\n<!-- agent:boundary:base-id -->\n{footer}"
);
let base_doc = CrdtDoc::from_text(&base);
let base_state = base_doc.encode_state();
let merged = merge(Some(&base_state), &ours, &theirs).unwrap();
let count = merged.matches("❯ Review the code").count();
assert_eq!(
count, 1,
"Prompt duplicated (count={}). Merged:\n{}",
count, merged
);
assert!(
merged.contains("The code does X."),
"Response missing:\n{}",
merged
);
}
#[test]
fn merge_foreign_tail_append_preserves_response_order() {
let header = "---\nagent_doc_format: template\nagent_doc_write: crdt\n---\n\n## Exchange\n\n<!-- agent:exchange patch=append -->\n";
let exchange_committed = "\
❯ first question
### Re: first question — opus-4-8
First answer here. Already committed to HEAD.
❯ second question
";
let queue_open = "<!-- /agent:exchange -->\n\n## Queue\n\n<!-- agent:queue -->\n";
let queue_close = "<!-- /agent:queue -->\n";
let base = format!(
"{header}{exchange_committed}<!-- agent:boundary:base-id -->\n{queue_open}{queue_close}"
);
let agent_response = "\
### Re: second question — opus-4-8
Second answer line one.
Second answer line two.
Second answer line three.
";
let ours = format!(
"{header}{exchange_committed}{agent_response}<!-- agent:boundary:new-id -->\n{queue_open}{queue_close}"
);
let theirs = format!(
"{header}{exchange_committed}<!-- agent:boundary:base-id -->\n{queue_open}do [#foreign-task]\n{queue_close}"
);
let base_state = CrdtDoc::from_text(&base).encode_state();
let merged = merge(Some(&base_state), &ours, &theirs).unwrap();
assert!(
merged.contains("### Re: second question — opus-4-8"),
"new response heading dropped:\n{}",
merged
);
let l1 = merged
.find("Second answer line one.")
.expect("line one missing");
let l2 = merged
.find("Second answer line two.")
.expect("line two missing");
let l3 = merged
.find("Second answer line three.")
.expect("line three missing");
assert!(
l1 < l2 && l2 < l3,
"response lines reversed/reordered (l1={l1} l2={l2} l3={l3}):\n{merged}"
);
let prior = merged
.find("### Re: first question")
.expect("prior response missing");
let current = merged.find("### Re: second question").unwrap();
assert!(
prior < current,
"new response hoisted above prior HEAD response (prior={prior} current={current}):\n{merged}"
);
assert!(
merged.contains("do [#foreign-task]"),
"foreign queue append lost:\n{}",
merged
);
}
#[test]
fn merge_stale_theirs_cannot_delete_committed_response() {
let header = "---\nagent_doc_format: template\nagent_doc_write: crdt\n---\n\n## Exchange\n\n<!-- agent:exchange patch=append -->\n";
let committed = "❯ first question\n\n### Re: first question — opus-4-8\n\nFirst answer. Committed to HEAD.\n\n❯ second question\n";
let foot = "<!-- /agent:exchange -->\n";
let base = format!("{header}{committed}<!-- agent:boundary:base-id -->\n{foot}");
let resp = "### Re: second question — opus-4-8\n\nLine one.\nLine two.\nLine three.\n\n";
let ours = format!("{header}{committed}{resp}<!-- agent:boundary:new-id -->\n{foot}");
let theirs = format!("{header}❯ second question\n<!-- agent:boundary:base-id -->\n{foot}");
let base_state = CrdtDoc::from_text(&base).encode_state();
let merged = merge(Some(&base_state), &ours, &theirs).unwrap();
assert!(
merged.contains("### Re: first question — opus-4-8"),
"committed prior response heading deleted by stale theirs:\n{merged}"
);
assert!(
merged.contains("First answer. Committed to HEAD."),
"committed prior response body deleted by stale theirs:\n{merged}"
);
assert!(
merged.contains("### Re: second question — opus-4-8"),
"new response heading missing:\n{merged}"
);
let prior = merged.find("### Re: first question").unwrap();
let current = merged.find("### Re: second question").unwrap();
assert!(prior < current, "response order inverted:\n{merged}");
}
#[test]
fn merge_committed_guard_keeps_new_user_input() {
let header = "---\nagent_doc_format: template\nagent_doc_write: crdt\n---\n\n## Exchange\n\n<!-- agent:exchange patch=append -->\n";
let committed = "❯ first question\n\n### Re: first question — opus-4-8\n\nFirst answer. Committed to HEAD.\n\n❯ second question\n";
let foot = "<!-- /agent:exchange -->\n";
let base = format!("{header}{committed}<!-- agent:boundary:base-id -->\n{foot}");
let resp = "### Re: second question — opus-4-8\n\nAnswer body.\n\n";
let ours = format!("{header}{committed}{resp}<!-- agent:boundary:new-id -->\n{foot}");
let theirs = format!(
"{header}{committed}❯ a genuinely new third prompt\n<!-- agent:boundary:base-id -->\n{foot}"
);
let base_state = CrdtDoc::from_text(&base).encode_state();
let merged = merge(Some(&base_state), &ours, &theirs).unwrap();
assert!(
merged.contains("❯ a genuinely new third prompt"),
"new user prompt clobbered by committed-response guard:\n{merged}"
);
assert!(
merged.contains("### Re: first question"),
"committed prior response lost:\n{merged}"
);
assert!(
merged.contains("### Re: second question"),
"new response lost:\n{merged}"
);
}
fn doc_with_exchange_queue(exchange: &str, queue: &str) -> String {
format!(
"---\nagent_doc_format: template\n---\n\n## Exchange\n\n<!-- agent:exchange -->\n{exchange}\n<!-- /agent:exchange -->\n\n## Queue\n\n<!-- agent:queue -->\n{queue}\n<!-- /agent:queue -->\n"
)
}
#[test]
fn merge_by_component_isolates_concurrent_cross_component_edits() {
let base = doc_with_exchange_queue("Existing prompt.", "- do [#a1]");
let base_state = CrdtDoc::from_text(&base).encode_state();
let ours = doc_with_exchange_queue(
"Existing prompt.\n\n### Re: existing — opus\n\nAgent response body.",
"- do [#a1]",
);
let theirs = doc_with_exchange_queue("Existing prompt.", "- do [#a1]\n- do [#b2]");
let merged = merge_by_component(Some(&base_state), &ours, &theirs).unwrap();
assert!(
merged.contains("### Re: existing"),
"agent exchange edit lost:\n{merged}"
);
assert!(
merged.contains("[#b2]"),
"operator queue edit lost:\n{merged}"
);
let queue_body = merged
.split("<!-- agent:queue -->")
.nth(1)
.and_then(|s| s.split("<!-- /agent:queue -->").next())
.unwrap_or("");
let exchange_body = merged
.split("<!-- agent:exchange -->")
.nth(1)
.and_then(|s| s.split("<!-- /agent:exchange -->").next())
.unwrap_or("");
assert!(
!queue_body.contains("Agent response body"),
"exchange content spliced into queue:\n{queue_body}"
);
assert!(
!exchange_body.contains("[#b2]"),
"queue content spliced into exchange:\n{exchange_body}"
);
}
#[test]
fn merge_by_component_blocks_console_output_in_queue_corruption() {
let console = "```\n● Supervisor is now fresh (recycled to 0.34.35).\n```";
let base = doc_with_exchange_queue("Q.", "- :pushpin: typing here");
let base_state = CrdtDoc::from_text(&base).encode_state();
let ours = doc_with_exchange_queue(
&format!("Q.\n\n### Re: q\n\n{console}"),
"- :pushpin: typing here",
);
let theirs = doc_with_exchange_queue("Q.", "- :pushpin: typing here now extended");
let merged = merge_by_component(Some(&base_state), &ours, &theirs).unwrap();
let queue_body = merged
.split("<!-- agent:queue -->")
.nth(1)
.and_then(|s| s.split("<!-- /agent:queue -->").next())
.unwrap_or("");
assert!(
!queue_body.contains("Supervisor is now fresh"),
"console output corrupted the queue node:\n{queue_body}"
);
assert!(
queue_body.contains("extended"),
"operator queue edit lost:\n{queue_body}"
);
assert!(
merged.contains("Supervisor is now fresh"),
"console output must remain in its own (exchange) node:\n{merged}"
);
}
#[test]
fn merge_by_component_degenerates_to_whole_doc_without_components() {
let base = "# Document\n\nBase content.\n";
let base_state = CrdtDoc::from_text(base).encode_state();
let ours = format!("{base}## Agent\n\nAgent response.\n");
let theirs = format!("{base}## User\n\nUser addition.\n");
let by_component = merge_by_component(Some(&base_state), &ours, &theirs).unwrap();
let whole_doc = merge(Some(&base_state), &ours, &theirs).unwrap();
assert_eq!(
by_component, whole_doc,
"component merge must equal whole-doc merge for a component-less doc"
);
}
#[test]
fn merge_by_component_falls_back_on_structural_divergence() {
let base = doc_with_exchange_queue("Q.", "- do [#a1]");
let base_state = CrdtDoc::from_text(&base).encode_state();
let ours = doc_with_exchange_queue("Q.\n\n### Re: q\n\nResponse.", "- do [#a1]");
let theirs = "---\nagent_doc_format: template\n---\n\n## Exchange\n\n<!-- agent:exchange -->\nQ.\n<!-- /agent:exchange -->\n".to_string();
let by_component = merge_by_component(Some(&base_state), &ours, &theirs).unwrap();
let whole_doc = merge(Some(&base_state), &ours, &theirs).unwrap();
assert_eq!(
by_component, whole_doc,
"structural divergence must use whole-doc fallback"
);
}
#[test]
fn multinode_state_roundtrip_encode_decode() {
let doc =
doc_with_exchange_queue("Prompt.\n\n### Re: q\n\nBody.", "- do [#a1]\n- do [#b2]");
let state = MultiNodeState::from_text(&doc).unwrap();
let encoded = state.encode();
let decoded = MultiNodeState::decode(&encoded).unwrap();
assert_eq!(
decoded, state,
"decode must reproduce the encoded per-node state"
);
assert_eq!(
decoded.to_text().unwrap(),
doc,
"per-node state must reconstruct the original document text"
);
assert!(
decoded.nodes.len() >= 4,
"expected ≥4 nodes, got {}",
decoded.nodes.len()
);
assert!(
decoded
.nodes
.iter()
.any(|n| n.name.as_deref() == Some("exchange"))
);
assert!(
decoded
.nodes
.iter()
.any(|n| n.name.as_deref() == Some("queue"))
);
}
#[test]
fn multinode_state_decode_rejects_non_container() {
let legacy = CrdtDoc::from_text("# Doc\n\nlegacy whole-doc state\n").encode_state();
assert!(
MultiNodeState::decode(&legacy).is_err(),
"legacy whole-doc state must not decode as a per-node container"
);
}
#[test]
fn multinode_state_migrates_legacy_whole_doc() {
let doc = doc_with_exchange_queue("Q.", "- do [#a1]");
let legacy = CrdtDoc::from_text(&doc).encode_state();
let migrated = MultiNodeState::decode_or_migrate(&legacy, "");
assert_eq!(
migrated.to_text().unwrap(),
doc,
"legacy migration must preserve the full document text"
);
assert!(
migrated
.nodes
.iter()
.any(|n| n.name.as_deref() == Some("queue")),
"legacy migration must recover component nodes"
);
}
#[test]
fn multinode_merge_advances_node_base_independently() {
let base_doc = doc_with_exchange_queue("Q.", "- do [#a1]");
let base = MultiNodeState::from_text(&base_doc).unwrap();
let ours = doc_with_exchange_queue("Q.\n\n### Re: q\n\nAgent body.", "- do [#a1]");
let theirs = base_doc.clone();
let (merged, advanced) = MultiNodeState::merge(Some(&base), &ours, &theirs).unwrap();
assert!(
merged.contains("Agent body."),
"exchange edit lost:\n{merged}"
);
let queue_before = base
.nodes
.iter()
.find(|n| n.name.as_deref() == Some("queue"))
.map(|n| &n.state);
let queue_after = advanced
.nodes
.iter()
.find(|n| n.name.as_deref() == Some("queue"))
.map(|n| &n.state);
assert_eq!(
queue_before, queue_after,
"untouched queue node base must be byte-identical across the cycle"
);
let exchange_before = base
.nodes
.iter()
.find(|n| n.name.as_deref() == Some("exchange"))
.map(|n| &n.state);
let exchange_after = advanced
.nodes
.iter()
.find(|n| n.name.as_deref() == Some("exchange"))
.map(|n| &n.state);
assert_ne!(
exchange_before, exchange_after,
"changed exchange node base must advance"
);
}
#[test]
fn multinode_merge_isolates_concurrent_cross_component_edits() {
let base_doc = doc_with_exchange_queue("Existing prompt.", "- do [#a1]");
let base = MultiNodeState::from_text(&base_doc).unwrap();
let ours = doc_with_exchange_queue(
"Existing prompt.\n\n### Re: existing — opus\n\nAgent response body.",
"- do [#a1]",
);
let theirs = doc_with_exchange_queue("Existing prompt.", "- do [#a1]\n- do [#b2]");
let (merged, _advanced) = MultiNodeState::merge(Some(&base), &ours, &theirs).unwrap();
assert!(
merged.contains("### Re: existing"),
"exchange edit lost:\n{merged}"
);
assert!(merged.contains("[#b2]"), "queue edit lost:\n{merged}");
let queue_body = merged
.split("<!-- agent:queue -->")
.nth(1)
.and_then(|s| s.split("<!-- /agent:queue -->").next())
.unwrap_or("");
assert!(
!queue_body.contains("Agent response body"),
"exchange content spliced into queue:\n{queue_body}"
);
}
#[test]
fn multinode_merge_falls_back_on_structural_divergence() {
let base_doc = doc_with_exchange_queue("Q.", "- do [#a1]");
let base = MultiNodeState::from_text(&base_doc).unwrap();
let ours = doc_with_exchange_queue("Q.\n\n### Re: q\n\nResponse.", "- do [#a1]");
let theirs = "---\nagent_doc_format: template\n---\n\n## Exchange\n\n<!-- agent:exchange -->\nQ.\n<!-- /agent:exchange -->\n".to_string();
let (merged, advanced) = MultiNodeState::merge(Some(&base), &ours, &theirs).unwrap();
let whole_doc = {
let base_state = CrdtDoc::from_text(&base_doc).encode_state();
merge(Some(&base_state), &ours, &theirs).unwrap()
};
assert_eq!(
merged, whole_doc,
"structural divergence must use whole-doc fallback"
);
assert_eq!(
advanced.to_text().unwrap(),
merged,
"fallback must still yield a per-node state round-tripping the merged text"
);
}
#[test]
fn multinode_merge_bootstraps_without_base() {
let ours = doc_with_exchange_queue("Q.\n\n### Re: q\n\nBody.", "- do [#a1]");
let theirs = doc_with_exchange_queue("Q.", "- do [#a1]\n- do [#b2]");
let (merged, state) = MultiNodeState::merge(None, &ours, &theirs).unwrap();
assert!(merged.contains("Body."));
assert!(merged.contains("[#b2]"));
assert_eq!(state.to_text().unwrap(), merged);
}
#[test]
fn reconcile_queue_item_edit_does_not_touch_sibling_item() {
let base = doc_with_exchange_queue("Q.", "- :pushpin: do [#a1]\n- :pushpin: do [#b2]");
let base_state = CrdtDoc::from_text(&base).encode_state();
let ours = doc_with_exchange_queue(
"Q.\n\n### Re: q — opus\n\nBody.",
"- :pushpin: do [#a1]\n- :pushpin: do [#b2]",
);
let theirs = doc_with_exchange_queue(
"Q.",
"- :pushpin: do [#a1]\n- :pushpin: do [#b2] with operator note",
);
let merged = merge_by_component(Some(&base_state), &ours, &theirs).unwrap();
assert!(
merged.contains("### Re: q"),
"exchange edit lost:\n{merged}"
);
assert_eq!(
merged.matches("do [#a1]").count(),
1,
"sibling item A drifted/duplicated:\n{merged}"
);
assert!(
merged.contains("- :pushpin: do [#a1]\n"),
"sibling item A text drifted:\n{merged}"
);
assert!(
merged.contains("operator note"),
"item B operator edit lost:\n{merged}"
);
}
#[test]
fn reconcile_exchange_prompt_between_blocks_converges_with_new_block() {
let r1 = "### Re: first — opus\n\nFirst answer.\n\n";
let r2 = "### Re: second — opus\n\nSecond answer.\n\n";
let base = doc_with_exchange_queue(&format!("{r1}{r2}"), "- do [#a1]");
let base_state = CrdtDoc::from_text(&base).encode_state();
let r3 = "### Re: third — opus\n\nThird answer.\n\n";
let ours = doc_with_exchange_queue(&format!("{r1}{r2}{r3}"), "- do [#a1]");
let theirs = doc_with_exchange_queue(
&format!("{r1}❯ a prompt typed between blocks\n\n{r2}"),
"- do [#a1]",
);
let merged = merge_by_component(Some(&base_state), &ours, &theirs).unwrap();
assert!(merged.contains("### Re: first"), "block 1 lost:\n{merged}");
assert!(merged.contains("### Re: second"), "block 2 lost:\n{merged}");
assert!(
merged.contains("### Re: third"),
"appended block 3 lost:\n{merged}"
);
assert!(
merged.contains("a prompt typed between blocks"),
"interleaved operator prompt lost:\n{merged}"
);
assert_eq!(
merged.matches("### Re: first").count(),
1,
"block 1 cross-spliced/duplicated:\n{merged}"
);
assert_eq!(
merged.matches("### Re: third").count(),
1,
"block 3 cross-spliced/duplicated:\n{merged}"
);
}
#[test]
fn reconcile_queue_preserves_concurrent_user_item_addition() {
let base = doc_with_exchange_queue("Q.", "- do [#a1]");
let base_state = CrdtDoc::from_text(&base).encode_state();
let ours = doc_with_exchange_queue("Q.", "- do [#a1] agent touched");
let theirs = doc_with_exchange_queue("Q.", "- do [#a1]\n- do [#user-added]");
let merged = merge_by_component(Some(&base_state), &ours, &theirs).unwrap();
assert!(
merged.contains("[#user-added]"),
"concurrent user queue addition dropped:\n{merged}"
);
assert!(
merged.contains("agent touched"),
"concurrent agent item edit lost:\n{merged}"
);
}
#[test]
fn reconcile_queue_honors_clean_item_deletion() {
let base = doc_with_exchange_queue("Q.", "- do [#a1]\n- do [#b2]");
let base_state = CrdtDoc::from_text(&base).encode_state();
let ours = doc_with_exchange_queue("Q.\n\n### Re: q\n\nBody.", "- do [#a1]\n- do [#b2]");
let theirs = doc_with_exchange_queue("Q.", "- do [#a1]");
let merged = merge_by_component(Some(&base_state), &ours, &theirs).unwrap();
assert!(merged.contains("Body."), "exchange edit lost:\n{merged}");
let queue_body = merged
.split("<!-- agent:queue -->")
.nth(1)
.and_then(|s| s.split("<!-- /agent:queue -->").next())
.unwrap_or("");
assert!(queue_body.contains("[#a1]"), "item A lost:\n{queue_body}");
assert!(
!queue_body.contains("[#b2]"),
"clean deletion of item B not honored:\n{queue_body}"
);
}
#[test]
fn reconcile_exchange_never_drops_committed_block_on_stale_side() {
let r1 = "### Re: first — opus\n\nFirst answer.\n\n";
let r2 = "### Re: second — opus\n\nSecond answer.\n\n";
let base = doc_with_exchange_queue(&format!("{r1}{r2}"), "- do [#a1]");
let base_state = CrdtDoc::from_text(&base).encode_state();
let ours = doc_with_exchange_queue(&format!("{r1}{r2}"), "- do [#a1]\n- do [#new]");
let theirs = doc_with_exchange_queue(&r1.to_string(), "- do [#a1]");
let merged = merge_by_component(Some(&base_state), &ours, &theirs).unwrap();
assert!(
merged.contains("### Re: second"),
"stale side deleted a committed block:\n{merged}"
);
assert!(
merged.contains("[#new]"),
"concurrent queue add lost:\n{merged}"
);
}
#[test]
fn reconcile_falls_back_on_duplicate_item_keys() {
let base = doc_with_exchange_queue("Q.", "- repeated item\n- repeated item");
let base_state = CrdtDoc::from_text(&base).encode_state();
let ours = doc_with_exchange_queue("Q.", "- repeated item\n- repeated item\n- new one");
let theirs = doc_with_exchange_queue("Q.", "- repeated item\n- repeated item");
let merged = merge_by_component(Some(&base_state), &ours, &theirs).unwrap();
assert!(
merged.contains("new one"),
"duplicate-key fallback dropped the new item:\n{merged}"
);
}
#[test]
fn first_hash_id_extracts_durable_id() {
assert_eq!(
first_hash_id("- :pushpin: do [#qnodemerge3]"),
Some("qnodemerge3".to_string())
);
assert_eq!(
first_hash_id("- [ ] [#6b5hwire] some text"),
Some("6b5hwire".to_string())
);
assert_eq!(first_hash_id("- free text no id"), None);
}
#[test]
fn list_item_key_stable_across_strike_and_glyphs() {
assert_eq!(list_item_key("- do [#a1]"), list_item_key("- ~~do [#a1]~~"));
assert_eq!(
list_item_key("- :pushpin: hello world"),
list_item_key("- ~~hello world~~")
);
}
#[test]
fn order_union_weaves_theirs_only_inserts() {
let ours = ["a", "b", "c"];
let theirs = ["a", "x", "b", "c", "y"];
assert_eq!(
order_union(&ours, &theirs),
vec![
"a".to_string(),
"x".into(),
"b".into(),
"c".into(),
"y".into()
]
);
}
#[test]
fn split_list_children_is_lossless() {
let body = "\n- do [#a1]\n continuation line\n- :pushpin: do [#b2]\n\n";
let children = split_list_children(body).unwrap();
let joined: String = children.iter().map(|c| c.text.as_str()).collect();
assert_eq!(joined, body, "list segmentation is not lossless");
}
#[test]
fn split_exchange_children_is_lossless() {
let body = "\n### Re: a — opus\n\nbody a\n\n### Re: b — opus\n\nbody b\n";
let children = split_exchange_children(body).unwrap();
let joined: String = children.iter().map(|c| c.text.as_str()).collect();
assert_eq!(joined, body, "exchange segmentation is not lossless");
}
}