use serde_json::Value;
use std::path::{Path, PathBuf};
use std::process::{Command, Output};
use tempfile::TempDir;
const NEEDS_SERVER: &str =
"needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN, run with --include-ignored";
fn server() -> (String, String) {
let url = std::env::var("DECIDUOUS_TEST_SERVER").unwrap_or_else(|_| panic!("{NEEDS_SERVER}"));
let token = std::env::var("DECIDUOUS_TEST_TOKEN").unwrap_or_else(|_| panic!("{NEEDS_SERVER}"));
(url.trim_end_matches('/').to_string(), token)
}
fn unique(prefix: &str) -> String {
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
format!("{prefix}-{}-{nanos}", std::process::id())
}
struct Sandbox {
root: TempDir,
token: String,
}
impl Sandbox {
fn new(token: &str) -> Self {
let root = TempDir::new().unwrap();
std::fs::create_dir_all(root.path().join("home")).unwrap();
Sandbox {
root,
token: token.to_string(),
}
}
fn path(&self) -> &Path {
self.root.path()
}
fn git(&self, dir: &Path, args: &[&str]) -> Output {
let out = Command::new("git")
.args(args)
.current_dir(dir)
.env("HOME", self.path().join("home"))
.env("GIT_CONFIG_NOSYSTEM", "1")
.env("GIT_AUTHOR_NAME", "t")
.env("GIT_AUTHOR_EMAIL", "t@t")
.env("GIT_COMMITTER_NAME", "t")
.env("GIT_COMMITTER_EMAIL", "t@t")
.output()
.unwrap();
assert!(
out.status.success(),
"git {args:?}: {}",
String::from_utf8_lossy(&out.stderr)
);
out
}
fn dx(&self, dir: &Path, args: &[&str]) -> Output {
Command::new(env!("CARGO_BIN_EXE_deciduous"))
.args(args)
.current_dir(dir)
.env("HOME", self.path().join("home"))
.env("XDG_CONFIG_HOME", self.path().join("home").join(".config"))
.env("DECIDUOUS_MCP_TOKEN", &self.token)
.env("DECIDUOUS_NO_SERVER", "1")
.env_remove("DECIDUOUS_DB_PATH")
.env("NO_COLOR", "1")
.output()
.unwrap()
}
fn dx_ok(&self, dir: &Path, args: &[&str]) -> String {
let out = self.dx(dir, args);
assert!(
out.status.success(),
"deciduous {args:?} failed\nstdout: {}\nstderr: {}",
text(&out.stdout),
text(&out.stderr)
);
format!("{}{}", text(&out.stdout), text(&out.stderr))
}
fn repo(&self, rel: &str) -> PathBuf {
let dir = self.path().join(rel);
std::fs::create_dir_all(&dir).unwrap();
self.git(&dir, &["init", "-q", "-b", "main"]);
self.git(
&dir,
&[
"commit",
"-q",
"--allow-empty",
"-m",
&format!("init {rel}"),
],
);
self.dx_ok(&dir, &["init"]);
dir
}
fn remote_repo(&self, rel: &str, url: &str, workspace: &str) -> PathBuf {
let dir = self.repo(rel);
self.dx_ok(&dir, &["remote", "init", url, "--workspace", workspace]);
dir
}
}
fn text(b: &[u8]) -> String {
String::from_utf8_lossy(b).to_string()
}
fn export(url: &str, token: &str, workspace: &str) -> Value {
ureq::get(&format!("{url}/export?workspace={workspace}"))
.set("authorization", &format!("Bearer {token}"))
.call()
.unwrap()
.into_json()
.unwrap()
}
fn server_node<'a>(graph: &'a Value, title: &str) -> &'a Value {
graph["nodes"]
.as_array()
.unwrap()
.iter()
.find(|n| n["title"] == title)
.unwrap_or_else(|| panic!("no node titled {title:?} on the server: {graph}"))
}
fn local_change_id(sb: &Sandbox, dir: &Path, id: i64) -> String {
let g: Value = serde_json::from_str(&sb.dx_ok(dir, &["graph"])).unwrap();
g["nodes"]
.as_array()
.unwrap()
.iter()
.find(|n| n["id"] == id)
.unwrap()["change_id"]
.as_str()
.unwrap()
.to_string()
}
fn mcp(url: &str, token: &str, workspace: &str, calls: &[(&str, Value)]) {
let post = |sid: Option<&str>, body: Value| {
let mut req = ureq::post(&format!("{url}/mcp"))
.set("authorization", &format!("Bearer {token}"))
.set("accept", "application/json, text/event-stream")
.set("x-deciduous-workspace", workspace);
if let Some(s) = sid {
req = req.set("mcp-session-id", s);
}
req.send_json(body).unwrap()
};
let init = post(
None,
serde_json::json!({"jsonrpc":"2.0","id":0,"method":"initialize","params":{
"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"0"}}}),
);
let sid = init.header("mcp-session-id").unwrap().to_string();
post(
Some(&sid),
serde_json::json!({"jsonrpc":"2.0","method":"notifications/initialized"}),
);
for (i, (tool, args)) in calls.iter().enumerate() {
let resp = post(
Some(&sid),
serde_json::json!({"jsonrpc":"2.0","id":i + 1,"method":"tools/call","params":{"name":tool,"arguments":args}}),
)
.into_string()
.unwrap();
let body = if resp.trim_start().starts_with('{') {
resp
} else {
resp.lines()
.filter_map(|l| l.strip_prefix("data:"))
.collect::<Vec<_>>()
.join("\n")
};
let v: Value = serde_json::from_str(&body).unwrap();
assert!(
v.get("error").is_none() && v["result"]["isError"] != true,
"{tool} failed on the server: {v}"
);
}
}
fn set_remote_url(dir: &Path, url: &str) {
let path = dir.join(".deciduous").join("config.toml");
let text = std::fs::read_to_string(&path).unwrap();
let mut doc: toml_edit::DocumentMut = text.parse().unwrap();
doc["remote"]["url"] = toml_edit::value(url);
std::fs::write(&path, doc.to_string()).unwrap();
}
fn log_lines(dir: &Path) -> Vec<Value> {
let path = dir.join(".deciduous").join("remote-log.jsonl");
std::fs::read_to_string(&path)
.unwrap_or_default()
.lines()
.map(|l| serde_json::from_str(l).unwrap())
.collect()
}
fn dead_url() -> String {
let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = l.local_addr().unwrap().port();
drop(l);
format!("http://127.0.0.1:{port}")
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn status_and_link_do_not_overwrite_an_agents_edit() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-c1");
let dir = sb.remote_repo("c1", &url, &ws);
sb.dx_ok(&dir, &["add", "goal", "a1"]);
sb.dx_ok(&dir, &["add", "goal", "g1"]);
let g = export(&url, &token, &ws);
let a1 = server_node(&g, "a1")["id"].as_str().unwrap().to_string();
let g1 = server_node(&g, "g1")["id"].as_str().unwrap().to_string();
mcp(
&url,
&token,
&ws,
&[
(
"update_node",
serde_json::json!({"node_id": a1, "title": "a1 retitled by agent", "description": "agent detail"}),
),
(
"update_node",
serde_json::json!({"node_id": g1, "title": "g1 retitled by agent", "status": "completed"}),
),
],
);
sb.dx_ok(&dir, &["status", "1", "completed"]);
sb.dx_ok(&dir, &["link", "2", "1", "-r", "because"]);
let g = export(&url, &token, &ws);
let a1 = server_node(&g, "a1 retitled by agent");
assert_eq!(a1["status"], "completed", "the CLI's status edit arrived");
assert_eq!(
a1["description"], "agent detail",
"the agent's description survived"
);
let g1 = server_node(&g, "g1 retitled by agent");
assert_eq!(g1["status"], "completed", "link did not resend g1");
assert_eq!(g.get("edges").unwrap().as_array().unwrap().len(), 1);
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn an_offline_edit_reaches_the_server_on_push() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-c2");
let dir = sb.remote_repo("c2", &url, &ws);
sb.dx_ok(&dir, &["add", "goal", "offline goal"]);
set_remote_url(&dir, &dead_url());
let out = sb.dx_ok(&dir, &["status", "1", "completed"]);
assert!(
out.contains("queued"),
"the user is told the write waits: {out}"
);
set_remote_url(&dir, &url);
let out = sb.dx_ok(&dir, &["remote", "push"]);
assert!(
out.contains("1 applied"),
"push replays the queued op: {out}"
);
let g = export(&url, &token, &ws);
assert_eq!(server_node(&g, "offline goal")["status"], "completed");
let out = sb.dx_ok(&dir, &["remote", "push"]);
assert!(out.contains("Nothing to push"), "{out}");
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn archaeology_pivot_and_supersede_reach_the_server() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-c3");
let dir = sb.remote_repo("c3", &url, &ws);
sb.dx_ok(&dir, &["add", "decision", "old approach"]);
sb.dx_ok(&dir, &["add", "decision", "side approach"]);
sb.dx_ok(
&dir,
&["archaeology", "pivot", "1", "it was slow", "new approach"],
);
sb.dx_ok(&dir, &["archaeology", "supersede", "2"]);
let local: Value = serde_json::from_str(&sb.dx_ok(&dir, &["graph"])).unwrap();
let g = export(&url, &token, &ws);
assert_eq!(
g["nodes"].as_array().unwrap().len(),
local["nodes"].as_array().unwrap().len(),
"server: {g}"
);
assert_eq!(
g["edges"].as_array().unwrap().len(),
local["edges"].as_array().unwrap().len()
);
assert_eq!(server_node(&g, "old approach")["status"], "superseded");
assert_eq!(server_node(&g, "side approach")["status"], "superseded");
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn delete_and_unlink_reach_the_server() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-del");
let dir = sb.remote_repo("del", &url, &ws);
sb.dx_ok(&dir, &["add", "goal", "keep"]);
sb.dx_ok(&dir, &["add", "option", "doomed"]);
sb.dx_ok(&dir, &["add", "option", "unlinked"]);
sb.dx_ok(&dir, &["link", "1", "2"]);
sb.dx_ok(&dir, &["link", "1", "3"]);
let out = sb.dx_ok(&dir, &["unlink", "1", "3"]);
assert!(!out.contains("removed locally only"), "{out}");
sb.dx_ok(&dir, &["delete", "2"]);
let g = export(&url, &token, &ws);
let live: Vec<&str> = g["nodes"]
.as_array()
.unwrap()
.iter()
.filter(|n| n["deleted_at"].is_null())
.map(|n| n["title"].as_str().unwrap())
.collect();
assert!(
!live.contains(&"doomed"),
"server still has the deleted node: {g}"
);
assert!(g["edges"].as_array().unwrap().is_empty(), "{g}");
}
#[test]
fn writes_queue_in_an_ignored_log_while_the_server_is_unreachable() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = sb.repo("offline");
std::fs::write(
dir.join(".deciduous").join("config.toml"),
format!(
"[remote]\nurl = \"{}\"\nworkspace = \"offline\"\n",
dead_url()
),
)
.unwrap();
let out = sb.dx_ok(&dir, &["add", "goal", "on a plane"]);
assert!(out.contains("queued"), "{out}");
sb.dx_ok(&dir, &["status", "1", "completed"]);
sb.dx_ok(&dir, &["prompt", "1", "the exact words"]);
let lines = log_lines(&dir);
let kinds: Vec<&str> = lines.iter().filter_map(|l| l["kind"].as_str()).collect();
assert_eq!(
kinds,
["create_node", "update_node", "update_node"],
"{lines:?}"
);
let cid = local_change_id(&sb, &dir, 1);
assert!(lines.iter().all(|l| l["change_id"] == cid.as_str()));
assert_eq!(lines[1]["set"], serde_json::json!({"status": "completed"}));
assert_eq!(
lines[2]["metadata"],
serde_json::json!({"prompt": "the exact words"})
);
let ids: std::collections::HashSet<&str> =
lines.iter().map(|l| l["op_id"].as_str().unwrap()).collect();
assert_eq!(ids.len(), 3, "every op has its own id");
let ignored = sb.git(&dir, &["check-ignore", ".deciduous/remote-log.jsonl"]);
assert!(text(&ignored.stdout).contains("remote-log.jsonl"));
}
#[test]
fn a_project_without_a_remote_keeps_no_log() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = sb.repo("local-only");
sb.dx_ok(&dir, &["add", "goal", "local"]);
assert!(!dir.join(".deciduous").join("remote-log.jsonl").exists());
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn status_reports_waiting_writes_and_content_differences_when_counts_match() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-c9");
let dir = sb.remote_repo("c9", &url, &ws);
sb.dx_ok(&dir, &["add", "goal", "shared goal"]);
set_remote_url(&dir, &dead_url());
sb.dx_ok(&dir, &["add", "goal", "bobs unpushed goal"]);
set_remote_url(&dir, &url);
let g = export(&url, &token, &ws);
let shared = server_node(&g, "shared goal")["id"]
.as_str()
.unwrap()
.to_string();
mcp(
&url,
&token,
&ws,
&[
(
"add_node",
serde_json::json!({"node_type": "goal", "title": "agents goal"}),
),
(
"update_node",
serde_json::json!({"node_id": shared, "status": "completed"}),
),
],
);
let st = sb.dx(&dir, &["remote", "status"]);
assert_eq!(st.status.code(), Some(1), "drift exits 1");
let out = text(&st.stdout);
assert!(!out.contains("OK"), "counts match, content does not: {out}");
assert!(out.contains("1 write(s) waiting"), "{out}");
assert!(out.contains("bobs unpushed goal"), "{out}");
assert!(out.contains("agents goal"), "{out}");
assert!(
out.contains("status") && out.contains("completed"),
"the differing field is named: {out}"
);
sb.dx_ok(&dir, &["remote", "pull"]);
let out = sb.dx_ok(&dir, &["remote", "status"]);
assert!(out.contains("In sync"), "{out}");
}
fn config_workspace(dir: &Path) -> Option<String> {
let text = std::fs::read_to_string(dir.join(".deciduous").join("config.toml")).ok()?;
let doc: toml::Value = toml::from_str(&text).ok()?;
doc.get("remote")?
.get("workspace")?
.as_str()
.map(str::to_string)
}
fn live_titles(g: &Value) -> Vec<String> {
let mut t: Vec<String> = g["nodes"]
.as_array()
.map(|a| {
a.iter()
.filter(|n| n["deleted_at"].is_null())
.filter_map(|n| n["title"].as_str().map(str::to_string))
.collect()
})
.unwrap_or_default();
t.sort();
t
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn renaming_or_cloning_the_repository_keeps_its_workspace() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let name = unique("wal-c6");
let dir = sb.repo(&name);
sb.dx_ok(&dir, &["remote", "init", &url]);
sb.dx_ok(&dir, &["add", "goal", "before the move"]);
let moved = sb.path().join(format!("{name}-renamed"));
std::fs::rename(&dir, &moved).unwrap();
sb.dx_ok(&moved, &["add", "goal", "after the move"]);
sb.git(&moved, &["add", ".deciduous/config.toml"]);
sb.git(&moved, &["commit", "-q", "-m", "deciduous config"]);
let clone = sb.path().join(format!("{name}-clone"));
sb.git(
sb.path(),
&[
"clone",
"-q",
moved.to_str().unwrap(),
clone.to_str().unwrap(),
],
);
sb.dx_ok(&clone, &["add", "goal", "from the clone"]);
assert_eq!(
live_titles(&export(&url, &token, &name)),
["after the move", "before the move", "from the clone"]
);
assert!(live_titles(&export(&url, &token, &format!("{name}-renamed"))).is_empty());
assert!(live_titles(&export(&url, &token, &format!("{name}-clone"))).is_empty());
assert_eq!(config_workspace(&clone).as_deref(), Some(name.as_str()));
}
#[test]
fn a_worktree_uses_the_main_repositorys_workspace() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let main = sb.repo("wt-main");
let url = stub_server(serde_json::json!({"nodes": [], "edges": [], "documents": []}));
std::fs::write(
main.join(".deciduous").join("config.toml"),
format!("[remote]\nurl = \"{url}\"\n"),
)
.unwrap();
sb.git(&main, &["add", ".deciduous/config.toml"]);
sb.git(&main, &["commit", "-q", "-m", "config"]);
sb.git(
&main,
&[
"worktree",
"add",
"-q",
"../wt-main-feature",
"-b",
"feature",
],
);
let wt = sb.path().join("wt-main-feature");
let out = sb.dx_ok(&wt, &["add", "goal", "from the worktree"]);
assert_eq!(
config_workspace(&wt).as_deref(),
Some("wt-main"),
"worktree resolved to its own directory name: {out}"
);
assert!(
out.contains("wt-main"),
"the recorded name is announced: {out}"
);
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn same_named_repositories_do_not_share_a_workspace_unless_named() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let name = unique("wal-c5");
let a = sb.repo(&format!("a/{name}"));
sb.dx_ok(&a, &["remote", "init", &url]);
sb.dx_ok(&a, &["add", "goal", "a's goal"]);
let b = sb.repo(&format!("b/{name}"));
let out = sb.dx(&b, &["remote", "init", &url]);
let err = text(&out.stderr);
assert!(!out.status.success(), "{}{err}", text(&out.stdout));
assert!(err.contains("another repository"), "{err}");
assert!(err.contains("--workspace"), "the fix is named: {err}");
assert_eq!(config_workspace(&b), None);
let c = sb.repo(&format!("c/{}", name.to_uppercase()));
assert!(!sb.dx(&c, &["remote", "init", &url]).status.success());
std::fs::write(
b.join(".deciduous").join("config.toml"),
format!("[remote]\nurl = \"{url}\"\n"),
)
.unwrap();
let out = sb.dx_ok(&b, &["add", "goal", "b's goal"]);
assert!(out.contains("another repository"), "{out}");
assert!(out.contains("refused"), "{out}");
let pull = sb.dx(&b, &["remote", "pull"]);
assert!(!pull.status.success());
let local: Value = serde_json::from_str(&sb.dx_ok(&b, &["graph"])).unwrap();
assert_eq!(live_titles(&local), ["b's goal"]);
let d = sb.repo(&format!("d/{name}"));
sb.dx_ok(&d, &["remote", "init", &url, "--workspace", &name]);
sb.dx_ok(&d, &["add", "goal", "d's goal"]);
assert_eq!(
live_titles(&export(&url, &token, &name)),
["a's goal", "d's goal"]
);
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn a_non_ascii_repository_name_reaches_the_server() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let name = format!("{}-café-ünï", unique("wal-c8"));
let dir = sb.repo(&name);
let out = sb.dx_ok(&dir, &["remote", "init", &url]);
assert!(out.contains(&name), "{out}");
sb.dx_ok(&dir, &["add", "goal", "accented"]);
let out = sb.dx_ok(&dir, &["remote", "status"]);
assert!(out.contains("In sync"), "{out}");
let encoded: String = name
.bytes()
.map(|b| match b {
b'a'..=b'z' | b'0'..=b'9' | b'-' => (b as char).to_string(),
b => format!("%{b:02X}"),
})
.collect();
assert_eq!(live_titles(&export(&url, &token, &encoded)), ["accented"]);
}
fn stub_server(export: Value) -> String {
let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
let port = server.server_addr().to_ip().unwrap().port();
std::thread::spawn(move || {
for mut req in server.incoming_requests() {
let mut body = String::new();
let _ = req.as_reader().read_to_string(&mut body);
let path = req.url().split('?').next().unwrap_or("").to_string();
let reply = match path.as_str() {
"/health" => "ok".to_string(),
"/claim" => r#"{"workspace":"stub","claim":"unchecked"}"#.to_string(),
"/locate" => r#"{"workspaces":[]}"#.to_string(),
"/ops" => {
let v: Value = serde_json::from_str(&body).unwrap();
let results: Vec<Value> = v["ops"]
.as_array()
.unwrap()
.iter()
.map(|op| serde_json::json!({"op_id": op["op_id"], "result": "applied"}))
.collect();
serde_json::json!({"workspace": "stub", "results": results}).to_string()
}
"/export" => export.to_string(),
_ => {
let _ = req.respond(tiny_http::Response::empty(404));
continue;
}
};
let _ = req.respond(tiny_http::Response::from_string(reply));
}
});
format!("http://127.0.0.1:{port}")
}
#[test]
fn pull_removes_a_node_the_server_deleted() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = sb.repo("tomb");
std::fs::write(
dir.join(".deciduous").join("config.toml"),
format!("[remote]\nurl = \"{}\"\nworkspace = \"tomb\"\n", dead_url()),
)
.unwrap();
sb.dx_ok(&dir, &["add", "goal", "deleted by an agent"]);
sb.dx_ok(&dir, &["add", "goal", "still here"]);
let dead = local_change_id(&sb, &dir, 1);
let kept = local_change_id(&sb, &dir, 2);
let node = |cid: &str, title: &str, deleted_at: Value| {
serde_json::json!({
"id": format!("srv-{cid}"), "change_id": cid, "node_type": "goal", "title": title,
"description": null, "status": "pending", "metadata": {"branch": "main"},
"created_at": "2026-09-23T10:00:00Z", "updated_at": "2026-09-23T10:00:00Z",
"deleted_at": deleted_at
})
};
let mut tombstone = node(
&dead,
"deleted by an agent",
Value::String("2099-01-01T00:00:00Z".into()),
);
tombstone["updated_at"] = Value::String("2099-01-01T00:00:00Z".into());
let url = stub_server(serde_json::json!({
"nodes": [
tombstone,
node(&kept, "still here", Value::Null),
node("agent-node-0001", "an agent's goal", Value::Null),
],
"edges": [],
"documents": []
}));
set_remote_url(&dir, &url);
let out = sb.dx_ok(&dir, &["remote", "pull"]);
let local: Value = serde_json::from_str(&sb.dx_ok(&dir, &["graph"])).unwrap();
assert_eq!(
live_titles(&local),
["an agent's goal", "still here"],
"pull said: {out}"
);
assert!(out.contains("removed 1"), "the removal is reported: {out}");
assert!(out.contains("imported 1"), "{out}");
}
#[test]
fn pull_keeps_an_agents_retitle_when_a_later_local_write_touched_another_field() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = sb.repo("retitle");
std::fs::write(
dir.join(".deciduous").join("config.toml"),
format!("[remote]\nurl = \"{}\"\nworkspace = \"stub\"\n", dead_url()),
)
.unwrap();
sb.dx_ok(&dir, &["add", "goal", "as written here"]);
let cid = local_change_id(&sb, &dir, 1);
let node = |title: &str, updated_at: &str| {
serde_json::json!({"nodes": [{
"id": "srv-1", "change_id": cid, "node_type": "goal", "title": title,
"description": null, "status": "pending", "metadata": {"branch": "main"},
"created_at": "2026-01-01T00:00:00Z", "updated_at": updated_at,
"deleted_at": null
}], "edges": [], "documents": []})
};
set_remote_url(
&dir,
&stub_server(node("as written here", "2026-01-01T00:00:00Z")),
);
sb.dx_ok(&dir, &["remote", "pull"]);
let retitled_at = chrono::Utc::now().to_rfc3339();
set_remote_url(
&dir,
&stub_server(node("retitled by the agent", &retitled_at)),
);
std::thread::sleep(std::time::Duration::from_millis(20));
sb.dx_ok(&dir, &["status", "1", "pending"]);
let out = sb.dx_ok(&dir, &["remote", "pull"]);
let local: Value = serde_json::from_str(&sb.dx_ok(&dir, &["graph"])).unwrap();
let n = &local["nodes"][0];
assert_eq!(
(n["title"].as_str(), n["status"].as_str()),
(Some("retitled by the agent"), Some("pending")),
"pull said: {out}"
);
let file = std::fs::read_to_string(dir.join(".deciduous").join("graph.json")).unwrap();
assert!(
file.contains("retitled by the agent"),
"graph.json, which git carries to the other clones, has it too:\n{file}"
);
}
const AGENT_NODE: &str = "5a1e0000-0000-4000-8000-00000000a9e7";
fn agent_node_export(title: &str, updated_at: &str) -> Value {
serde_json::json!({"nodes": [{
"id": "srv-1", "change_id": AGENT_NODE, "node_type": "decision", "title": title,
"description": null, "status": "pending", "metadata": {"branch": "main"},
"created_at": "2026-01-01T00:00:00Z", "updated_at": updated_at,
"deleted_at": null
}], "edges": [], "documents": []})
}
fn merge_inputs(
dir: &Path,
doc: &Value,
ours: impl Fn(&mut Value),
theirs: impl Fn(&mut Value),
) -> (PathBuf, PathBuf, PathBuf) {
let mut base = doc.clone();
base["nodes"].as_object_mut().unwrap().remove(AGENT_NODE);
let side = |edit: &dyn Fn(&mut Value)| {
let mut d = doc.clone();
edit(&mut d["nodes"][AGENT_NODE]);
d
};
let paths = (
dir.join("merge-base.json"),
dir.join("merge-ours.json"),
dir.join("merge-theirs.json"),
);
for (p, v) in [
(&paths.0, base),
(&paths.1, side(&ours)),
(&paths.2, side(&theirs)),
] {
std::fs::write(p, serde_json::to_string_pretty(&v).unwrap()).unwrap();
}
paths
}
fn merge_record(sb: &Sandbox, dir: &Path, (b, o, t): &(PathBuf, PathBuf, PathBuf)) -> Value {
let args = [
"merge-record",
b.to_str().unwrap(),
o.to_str().unwrap(),
t.to_str().unwrap(),
];
sb.dx_ok(dir, &args);
let merged: Value = serde_json::from_str(&std::fs::read_to_string(o).unwrap()).unwrap();
merged["nodes"][AGENT_NODE].clone()
}
fn pulled_agent_node(sb: &Sandbox, rel: &str, title: &str, updated_at: &str) -> (PathBuf, Value) {
let dir = sb.repo(rel);
std::fs::write(
dir.join(".deciduous").join("config.toml"),
format!(
"[remote]\nurl = \"{}\"\nworkspace = \"stub\"\n",
stub_server(agent_node_export(title, updated_at))
),
)
.unwrap();
sb.dx_ok(&dir, &["remote", "pull"]);
let doc: Value = serde_json::from_str(
&std::fs::read_to_string(dir.join(".deciduous").join("graph.json")).unwrap(),
)
.unwrap();
assert!(
doc["nodes"][AGENT_NODE].is_object(),
"the pull brought the agent's node: {doc}"
);
(dir, doc)
}
fn set_status(status: &'static str, at: &'static str) -> impl Fn(&mut Value) {
move |n: &mut Value| {
n["status"] = status.into();
n["updated_at"] = at.into();
}
}
fn set_prompt(prompt: &'static str, at: &'static str) -> impl Fn(&mut Value) {
move |n: &mut Value| {
n["metadata"]["prompt"] = prompt.into();
n["updated_at"] = at.into();
}
}
#[test]
fn the_merge_driver_keeps_a_status_set_here_on_a_node_that_came_only_from_the_server() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let (dir, doc) = pulled_agent_node(&sb, "driver", "added by the agent", "2026-01-01T00:00:00Z");
let files = merge_inputs(
&dir,
&doc,
set_status("completed", "2026-01-02T00:00:00Z"),
set_prompt("from clone 2", "2026-01-03T00:00:00Z"),
);
let n = merge_record(&sb, &dir, &files);
assert_eq!(
(n["status"].as_str(), n["metadata"]["prompt"].as_str()),
(Some("completed"), Some("from clone 2")),
"{n}"
);
let files = merge_inputs(
&dir,
&doc,
set_prompt("from clone 2", "2026-01-03T00:00:00Z"),
set_status("completed", "2026-01-02T00:00:00Z"),
);
let n = merge_record(&sb, &dir, &files);
assert_eq!(
(n["status"].as_str(), n["metadata"]["prompt"].as_str()),
(Some("completed"), Some("from clone 2")),
"{n}"
);
}
#[test]
fn the_merge_driver_does_not_take_a_stale_copy_for_an_edit() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let (dir, doc) = pulled_agent_node(
&sb,
"stale",
"retitled by the agent",
"2026-01-05T00:00:00Z",
);
let files = merge_inputs(
&dir,
&doc,
|_| {},
|n| {
n["title"] = "added by the agent".into();
n["updated_at"] = "2026-01-02T00:00:00Z".into();
},
);
let n = merge_record(&sb, &dir, &files);
assert_eq!(n["title"].as_str(), Some("retitled by the agent"), "{n}");
}
#[test]
fn the_merge_driver_in_a_clone_without_a_remote_merges_by_timestamp_as_before() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let (_, doc) = pulled_agent_node(&sb, "source", "added by the agent", "2026-01-01T00:00:00Z");
let dir = sb.repo("plain");
let files = merge_inputs(
&dir,
&doc,
set_status("completed", "2026-01-02T00:00:00Z"),
set_prompt("from clone 2", "2026-01-03T00:00:00Z"),
);
let n = merge_record(&sb, &dir, &files);
assert_eq!(
(n["status"].as_str(), n["metadata"]["prompt"].as_str()),
(Some("pending"), Some("from clone 2")),
"{n}"
);
assert!(
!dir.join(".deciduous").join("remote-log.jsonl").exists(),
"the driver wrote nothing beside the merge"
);
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn a_node_deleted_on_the_server_is_not_resent_and_edits_to_it_are_refused_loudly() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-c4");
let dir = sb.remote_repo("c4", &url, &ws);
sb.dx_ok(&dir, &["add", "goal", "doomed by an agent"]);
let g = export(&url, &token, &ws);
let id = server_node(&g, "doomed by an agent")["id"]
.as_str()
.unwrap()
.to_string();
set_remote_url(&dir, &dead_url());
sb.dx_ok(&dir, &["status", "1", "completed"]);
set_remote_url(&dir, &url);
mcp(
&url,
&token,
&ws,
&[("delete_node", serde_json::json!({"node_id": id}))],
);
let out = sb.dx_ok(&dir, &["add", "goal", "unrelated"]);
assert!(!out.contains("Pushed"), "{out}");
assert!(out.contains("Rejected"), "{out}");
assert!(out.contains("deleted on the server"), "{out}");
assert_eq!(live_titles(&export(&url, &token, &ws)), ["unrelated"]);
let out = text(&sb.dx(&dir, &["remote", "status"]).stdout);
assert!(out.contains("1 rejected"), "{out}");
sb.dx_ok(&dir, &["remote", "push", "--drop-rejected"]);
let out = text(&sb.dx(&dir, &["remote", "status"]).stdout);
assert!(out.contains("0 rejected"), "{out}");
}
fn server_id(g: &Value, title: &str) -> String {
server_node(g, title)["id"].as_str().unwrap().to_string()
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn a_queued_edit_does_not_overwrite_a_newer_agent_edit_to_the_same_field() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-stale");
let dir = sb.remote_repo("stale", &url, &ws);
sb.dx_ok(&dir, &["add", "goal", "same field"]);
sb.dx_ok(&dir, &["add", "goal", "other field"]);
set_remote_url(&dir, &dead_url());
sb.dx_ok(&dir, &["status", "1", "completed"]);
sb.dx_ok(&dir, &["status", "2", "completed"]);
set_remote_url(&dir, &url);
let g = export(&url, &token, &ws);
mcp(
&url,
&token,
&ws,
&[
(
"update_node",
serde_json::json!({"node_id": server_id(&g, "same field"), "status": "rejected"}),
),
(
"update_node",
serde_json::json!({"node_id": server_id(&g, "other field"), "title": "other field, retitled"}),
),
],
);
let raw = sb.dx(&dir, &["remote", "push"]);
let out = format!("{}{}", text(&raw.stdout), text(&raw.stderr));
assert!(!raw.status.success(), "a refused push exited 0: {out}");
assert!(out.contains("1 applied"), "{out}");
assert!(out.contains("Rejected"), "the conflict is reported: {out}");
assert!(
out.contains("changed on the server"),
"the reason names the conflict: {out}"
);
let g = export(&url, &token, &ws);
assert_eq!(
server_node(&g, "same field")["status"],
"rejected",
"the agent's newer status survived"
);
assert_eq!(
server_node(&g, "other field, retitled")["status"],
"completed",
"an edit to a different field still lands"
);
}
fn export_with_tombstones(url: &str, token: &str, workspace: &str) -> Value {
ureq::get(&format!("{url}/export?workspace={workspace}&tombstones=1"))
.set("authorization", &format!("Bearer {token}"))
.call()
.unwrap()
.into_json()
.unwrap()
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn a_node_the_server_deleted_is_not_seeded_back_and_pull_removes_it() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-del3");
let dir = sb.remote_repo("del3", &url, &ws);
sb.dx_ok(&dir, &["add", "action", "a1"]);
set_remote_url(&dir, &dead_url());
sb.dx(&dir, &["status", "1", "completed"]);
set_remote_url(&dir, &url);
let g = export(&url, &token, &ws);
mcp(
&url,
&token,
&ws,
&[(
"delete_node",
serde_json::json!({"node_id": server_id(&g, "a1")}),
)],
);
sb.dx_ok(&dir, &["add", "goal", "g2"]);
sb.dx(&dir, &["link", "2", "1"]);
sb.dx_ok(&dir, &["remote", "push", "--drop-rejected"]);
let out = text(&sb.dx(&dir, &["remote", "status"]).stdout);
assert!(
!out.contains("push --seed"),
"status does not send a deleted node to --seed: {out}"
);
assert!(out.contains("Deleted on the server"), "{out}");
let before = export_with_tombstones(&url, &token, &ws);
let out = sb.dx_ok(&dir, &["remote", "push", "--seed"]);
let after = export_with_tombstones(&url, &token, &ws);
let row = |g: &Value| {
g["nodes"]
.as_array()
.unwrap()
.iter()
.find(|n| !n["deleted_at"].is_null())
.cloned()
.unwrap_or_else(|| panic!("no tombstone in {g}"))
};
assert_eq!(
row(&before),
row(&after),
"seed wrote into the deleted row: {out}"
);
assert!(
after["edges"].as_array().unwrap().is_empty(),
"an edge to the deleted node was written: {out}"
);
let out = sb.dx_ok(&dir, &["remote", "pull"]);
assert!(out.contains("removed 1"), "{out}");
let out = sb.dx_ok(&dir, &["remote", "status"]);
assert!(out.contains("In sync"), "{out}");
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn a_pull_that_changes_no_content_reports_no_updates() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-pullcount");
let dir = sb.remote_repo("pullcount", &url, &ws);
sb.dx_ok(&dir, &["add", "goal", "one"]);
sb.dx_ok(&dir, &["add", "goal", "two"]);
sb.dx_ok(&dir, &["status", "2", "completed"]);
sb.dx_ok(&dir, &["link", "1", "2"]);
let out = sb.dx_ok(&dir, &["remote", "status"]);
assert!(out.contains("In sync"), "{out}");
let out = sb.dx_ok(&dir, &["remote", "pull"]);
assert!(
out.contains("imported 0 new node(s), updated 0, removed 0"),
"{out}"
);
let g = export(&url, &token, &ws);
mcp(
&url,
&token,
&ws,
&[(
"update_node",
serde_json::json!({"node_id": server_id(&g, "one"), "title": "one, retitled"}),
)],
);
let out = sb.dx_ok(&dir, &["remote", "pull"]);
assert!(out.contains("updated 1,"), "{out}");
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn an_edit_the_log_never_got_is_sent_by_push_repair() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-repair");
let dir = sb.remote_repo("repair", &url, &ws);
sb.dx_ok(&dir, &["add", "goal", "shared goal"]);
sb.dx_ok(&dir, &["prompt", "1", "first words"]);
let cfg = dir.join(".deciduous").join("config.toml");
let saved = std::fs::read_to_string(&cfg).unwrap();
std::fs::write(&cfg, "").unwrap();
sb.dx_ok(&dir, &["status", "1", "completed"]);
sb.dx_ok(&dir, &["prompt", "1", "second words"]);
std::fs::write(&cfg, saved).unwrap();
let st = sb.dx(&dir, &["remote", "status"]);
let out = text(&st.stdout);
assert!(out.contains("Different"), "{out}");
assert!(
out.contains("push --repair"),
"status names the command that fixes it: {out}"
);
let out = sb.dx_ok(&dir, &["remote", "push", "--repair"]);
assert!(out.contains("applied"), "{out}");
let g = export(&url, &token, &ws);
let n = server_node(&g, "shared goal");
assert_eq!(n["status"], "completed");
assert_eq!(n["metadata"]["prompt"], "second words");
let out = sb.dx_ok(&dir, &["remote", "status"]);
assert!(out.contains("In sync"), "{out}");
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn remote_status_exits_nonzero_when_anything_differs() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-rc");
let dir = sb.remote_repo("rc", &url, &ws);
sb.dx_ok(&dir, &["add", "goal", "g"]);
assert!(sb.dx(&dir, &["remote", "status"]).status.success());
let g = export(&url, &token, &ws);
mcp(
&url,
&token,
&ws,
&[(
"update_node",
serde_json::json!({"node_id": server_id(&g, "g"), "status": "completed"}),
)],
);
let st = sb.dx(&dir, &["remote", "status"]);
assert_eq!(st.status.code(), Some(1), "{}", text(&st.stdout));
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn an_attached_document_is_reported_and_seeded_with_its_bytes() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-doc");
let dir = sb.remote_repo("doc", &url, &ws);
sb.dx_ok(&dir, &["add", "goal", "has a spec"]);
std::fs::write(dir.join("spec.txt"), "the spec, in full\n").unwrap();
let cfg = dir.join(".deciduous").join("config.toml");
let saved = std::fs::read_to_string(&cfg).unwrap();
std::fs::write(&cfg, "").unwrap();
sb.dx_ok(&dir, &["doc", "attach", "1", "spec.txt"]);
std::fs::write(&cfg, saved).unwrap();
let out = text(&sb.dx(&dir, &["remote", "status"]).stdout);
assert!(!out.contains("In sync"), "{out}");
assert!(out.contains("spec.txt"), "the document is named: {out}");
let out = sb.dx_ok(&dir, &["remote", "push", "--seed"]);
assert!(out.contains("documents 1"), "{out}");
let g = export(&url, &token, &ws);
let docs = g["documents"].as_array().unwrap();
assert_eq!(docs.len(), 1, "{g}");
let id = docs[0]["id"].as_str().unwrap();
let bytes = ureq::get(&format!("{url}/documents/{id}"))
.set("authorization", &format!("Bearer {token}"))
.call()
.unwrap()
.into_string()
.unwrap();
assert_eq!(bytes, "the spec, in full\n");
let out = sb.dx_ok(&dir, &["remote", "status"]);
assert!(out.contains("In sync"), "{out}");
assert!(
out.to_lowercase().contains("themes and tags"),
"what is not compared is said: {out}"
);
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn writes_through_the_local_mcp_server_reach_the_server_while_it_runs() {
use std::io::{BufRead, BufReader, Write};
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-localmcp");
let dir = sb.remote_repo("localmcp", &url, &ws);
let mut child = Command::new(env!("CARGO_BIN_EXE_deciduous"))
.arg("mcp")
.current_dir(&dir)
.env("HOME", sb.path().join("home"))
.env("XDG_CONFIG_HOME", sb.path().join("home").join(".config"))
.env("DECIDUOUS_MCP_TOKEN", &token)
.env("DECIDUOUS_NO_SERVER", "1")
.env_remove("DECIDUOUS_DB_PATH")
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::null())
.spawn()
.unwrap();
let mut stdin = child.stdin.take().unwrap();
let mut stdout = BufReader::new(child.stdout.take().unwrap());
let mut call = |id: u64, method: &str, params: Value| {
let msg = serde_json::json!({"jsonrpc":"2.0","id":id,"method":method,"params":params});
writeln!(stdin, "{msg}").unwrap();
stdin.flush().unwrap();
let mut line = String::new();
stdout.read_line(&mut line).unwrap();
line
};
call(
1,
"initialize",
serde_json::json!({"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"t","version":"0"}}),
);
call(
2,
"tools/call",
serde_json::json!({"name":"add_node","arguments":{"node_type":"goal","title":"via local mcp"}}),
);
call(
3,
"tools/call",
serde_json::json!({"name":"update_status","arguments":{"node_id":1,"status":"completed"}}),
);
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
let status = loop {
let g = export(&url, &token, &ws);
let status = g["nodes"]
.as_array()
.unwrap()
.iter()
.find(|n| n["title"] == "via local mcp")
.map(|n| n["status"].clone());
if status.as_ref().is_some_and(|s| s == "completed") || std::time::Instant::now() > deadline
{
break status;
}
std::thread::sleep(std::time::Duration::from_millis(100));
};
let _ = child.kill();
let _ = child.wait();
assert_eq!(status, Some(Value::String("completed".into())));
}
fn unborn_repo(sb: &Sandbox, rel: &str) -> PathBuf {
let dir = sb.path().join(rel);
std::fs::create_dir_all(&dir).unwrap();
sb.git(&dir, &["init", "-q", "-b", "main"]);
sb.dx_ok(&dir, &["init"]);
dir
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn a_claimed_workspace_refuses_unborn_and_unrelated_repositories_on_every_path() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let name = unique("wal-c5b");
let a = sb.repo(&format!("a/{name}"));
sb.dx_ok(&a, &["remote", "init", &url]);
sb.dx_ok(&a, &["add", "goal", "A's secret goal"]);
let b = unborn_repo(&sb, &format!("b/{name}"));
std::fs::write(
b.join(".deciduous").join("config.toml"),
format!("[remote]\nurl = \"{url}\"\nworkspace = \"{name}\"\n"),
)
.unwrap();
let out = sb.dx_ok(&b, &["add", "goal", "B goal before first commit"]);
assert!(out.contains("no commit yet"), "{out}");
assert!(
!out.contains("once the server is reachable"),
"a refusal is not an outage: {out}"
);
assert!(!sb.dx(&b, &["remote", "pull"]).status.success());
let local: Value = serde_json::from_str(&sb.dx_ok(&b, &["graph"])).unwrap();
assert_eq!(live_titles(&local), ["B goal before first commit"]);
let c = sb.repo(&format!("c/{name}"));
std::fs::write(c.join(".deciduous").join("config.toml"), "").unwrap();
sb.dx_ok(&c, &["add", "goal", "C's private goal"]);
std::fs::write(
c.join(".deciduous").join("config.toml"),
format!("[remote]\nurl = \"{url}\"\n"),
)
.unwrap();
let seed = sb.dx(&c, &["remote", "push", "--seed"]);
assert!(!seed.status.success(), "{}", text(&seed.stdout));
assert!(text(&seed.stderr).contains("another repository"));
let over = sb.dx(&c, &["remote", "push", "--overwrite"]);
assert!(!over.status.success(), "{}", text(&over.stdout));
let st = sb.dx(&c, &["remote", "status"]);
assert!(!st.status.success());
assert!(
!text(&st.stdout).contains("A's secret goal"),
"status showed another repository's graph: {}",
text(&st.stdout)
);
assert_eq!(
live_titles(&export(&url, &token, &name)),
["A's secret goal"]
);
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn a_shallow_clone_writes_to_its_repositorys_workspace() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let name = unique("wal-shal");
let dir = sb.repo(&name);
for i in 0..2 {
sb.git(
&dir,
&["commit", "-q", "--allow-empty", "-m", &format!("c{i}")],
);
}
sb.dx_ok(&dir, &["remote", "init", &url]);
sb.dx_ok(&dir, &["add", "goal", "from the full clone"]);
sb.git(&dir, &["add", ".deciduous/config.toml"]);
sb.git(&dir, &["commit", "-q", "-m", "deciduous config"]);
let ci = sb.path().join("ci").join(&name);
std::fs::create_dir_all(ci.parent().unwrap()).unwrap();
sb.git(
sb.path(),
&[
"clone",
"-q",
"--depth",
"1",
&format!("file://{}", dir.display()),
ci.to_str().unwrap(),
],
);
sb.dx_ok(&ci, &["init"]);
let out = sb.dx_ok(&ci, &["add", "goal", "from ci"]);
assert!(!out.contains("another repository"), "{out}");
assert_eq!(
live_titles(&export(&url, &token, &name)),
["from ci", "from the full clone"]
);
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn a_renamed_1_0_7_repository_keeps_writing_to_the_workspace_it_wrote_before() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let name = unique("wal-old");
let dir = sb.repo(&name);
std::fs::write(dir.join(".deciduous").join("config.toml"), "").unwrap();
sb.dx_ok(&dir, &["add", "goal", "old1"]);
let graph: Value = serde_json::from_str(&sb.dx_ok(&dir, &["graph"])).unwrap();
ureq::post(&format!("{url}/import"))
.set("authorization", &format!("Bearer {token}"))
.send_json(serde_json::json!({"workspace": name, "graph": graph}))
.unwrap();
std::fs::write(
dir.join(".deciduous").join("config.toml"),
format!("[remote]\nurl = \"{url}\"\n"),
)
.unwrap();
let moved = sb.path().join(format!("{name}-moved"));
std::fs::rename(&dir, &moved).unwrap();
let out = sb.dx_ok(&moved, &["add", "goal", "after the move"]);
assert_eq!(
config_workspace(&moved).as_deref(),
Some(name.as_str()),
"{out}"
);
assert_eq!(
live_titles(&export(&url, &token, &name)),
["after the move", "old1"]
);
assert!(live_titles(&export(&url, &token, &format!("{name}-moved"))).is_empty());
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn concurrent_metadata_ops_on_one_node_keep_every_key() {
let (url, token) = server();
let ws = unique("wal-race");
let post = {
let (url, token, ws) = (url.clone(), token.clone(), ws.clone());
std::sync::Arc::new(move |ops: Value| -> Value {
ureq::post(&format!("{url}/ops"))
.set("authorization", &format!("Bearer {token}"))
.send_json(serde_json::json!({"workspace": ws, "ops": ops}))
.unwrap()
.into_json()
.unwrap()
})
};
post(serde_json::json!([{
"op_id": uuid::Uuid::new_v4().to_string(), "at": "2026-01-01T00:00:00Z",
"kind": "create_node", "change_id": "race-node", "node_type": "goal",
"title": "race", "status": "pending", "metadata": {},
"created_at": "2026-01-01T00:00:00Z", "updated_at": "2026-01-01T00:00:00Z"
}]));
let n = 24;
let handles: Vec<_> = (0..n)
.map(|i| {
let post = post.clone();
std::thread::spawn(move || {
post(serde_json::json!([{
"op_id": uuid::Uuid::new_v4().to_string(), "at": "2026-01-01T00:00:01Z",
"kind": "update_node", "change_id": "race-node",
"metadata": {format!("k{i}"): i},
"was_metadata": {format!("k{i}"): null}
}]))
})
})
.collect();
for h in handles {
let r = h.join().unwrap();
assert_eq!(r["results"][0]["result"], "applied", "{r}");
}
let g = export(&url, &token, &ws);
let meta = server_node(&g, "race")["metadata"]
.as_object()
.unwrap()
.clone();
let missing: Vec<usize> = (0..n)
.filter(|i| !meta.contains_key(&format!("k{i}")))
.collect();
assert!(missing.is_empty(), "lost keys {missing:?}: {meta:?}");
}
fn offline_repo(sb: &Sandbox, rel: &str) -> PathBuf {
let dir = sb.repo(rel);
std::fs::write(
dir.join(".deciduous").join("config.toml"),
format!(
"[remote]\nurl = \"{}\"\nworkspace = \"{rel}\"\n",
dead_url()
),
)
.unwrap();
dir
}
fn log_path(dir: &Path) -> PathBuf {
dir.join(".deciduous").join("remote-log.jsonl")
}
fn queued_titles(dir: &Path) -> Vec<String> {
log_lines(dir)
.iter()
.filter(|l| l["kind"] == "create_node")
.map(|l| l["title"].as_str().unwrap().to_string())
.collect()
}
#[test]
fn rust_n6_a_torn_last_line_does_not_swallow_the_next_write() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = offline_repo(&sb, "torn");
sb.dx_ok(&dir, &["add", "goal", "before-torn"]);
let mut f = std::fs::OpenOptions::new()
.append(true)
.open(log_path(&dir))
.unwrap();
std::io::Write::write_all(&mut f, br#"{"entry":"op","op_id":"torn"#).unwrap();
drop(f);
let out = sb.dx_ok(&dir, &["add", "goal", "after-torn"]);
assert_eq!(queued_titles(&dir), ["before-torn", "after-torn"], "{out}");
let aside = std::fs::read_to_string(dir.join(".deciduous").join("remote-log.unreadable"))
.unwrap_or_default();
assert!(
aside.contains(r#"{"entry":"op","op_id":"torn"#),
"{aside:?}"
);
assert!(out.contains("remote-log.unreadable"), "{out}");
assert!(!out.contains("server refused"), "{out}");
assert!(out.contains("2 write(s)"), "{out}");
}
#[test]
fn bridge_n5_an_op_glued_to_a_torn_fragment_is_still_sent() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = offline_repo(&sb, "glued");
sb.dx_ok(&dir, &["add", "goal", "glued-op"]);
let whole = std::fs::read_to_string(log_path(&dir)).unwrap();
std::fs::write(
log_path(&dir),
format!("{{\"entry\":\"op\",\"op_id\":\"torn{whole}"),
)
.unwrap();
let out = sb.dx(&dir, &["remote", "push"]);
let all = format!("{}{}", text(&out.stdout), text(&out.stderr));
assert!(!all.contains("is not a log entry"), "{all}");
assert!(all.contains("1 write(s) still waiting"), "{all}");
assert!(
all.contains("remote-log.unreadable") || all.contains("could not be read"),
"{all}"
);
}
#[test]
fn gitsync_n6_a_corrupt_line_is_blamed_on_the_file_and_the_writes_are_counted() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = offline_repo(&sb, "corrupt");
sb.dx_ok(&dir, &["add", "goal", "first"]);
let mut f = std::fs::OpenOptions::new()
.append(true)
.open(log_path(&dir))
.unwrap();
std::io::Write::write_all(&mut f, b"{\"entry\":\"op\",\"op_id\":\"x\"\n").unwrap();
drop(f);
sb.dx_ok(&dir, &["add", "goal", "corrupt1"]);
let out = sb.dx_ok(&dir, &["add", "goal", "corrupt2"]);
assert!(!out.contains("server refused"), "{out}");
assert!(out.contains("3 write(s)"), "{out}");
assert!(
out.contains("line 2"),
"the unreadable line is named: {out}"
);
assert!(!out.contains("delete that line"), "{out}");
}
#[test]
fn rust_n4_a_lock_left_by_a_killed_process_neither_stalls_nor_drops_a_write() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = offline_repo(&sb, "stalelock");
sb.dx_ok(&dir, &["add", "goal", "before"]);
std::fs::write(dir.join(".deciduous").join("remote-log.lock"), b"").unwrap();
let t = std::time::Instant::now();
let out = sb.dx_ok(&dir, &["add", "goal", "after-kill"]);
let took = t.elapsed();
assert!(
!out.contains("could not be queued") && !out.contains("not queued"),
"{out}"
);
assert_eq!(queued_titles(&dir), ["before", "after-kill"], "{out}");
assert!(
took < std::time::Duration::from_secs(5),
"took {took:?}: {out}"
);
}
#[test]
fn rust_n5_concurrent_local_writers_queue_a_true_was_in_commit_order() {
let sb = std::sync::Arc::new(Sandbox::new("0123456789abcdef0123456789abcdef"));
let dir = offline_repo(&sb, "stale-was");
sb.dx_ok(&dir, &["add", "goal", "raced"]);
let statuses = ["active", "completed", "rejected", "pending", "superseded"];
let writers: Vec<_> = (0..4)
.map(|w| {
let (sb, dir) = (sb.clone(), dir.clone());
std::thread::spawn(move || {
for i in 0..12 {
let s = statuses[(w + i) % statuses.len()];
sb.dx_ok(&dir, &["status", "1", s]);
}
})
})
.collect();
for w in writers {
w.join().unwrap();
}
let updates: Vec<Value> = log_lines(&dir)
.into_iter()
.filter(|l| l["kind"] == "update_node")
.collect();
assert_eq!(updates.len(), 48);
let mut expect = Value::String("pending".into());
let mut broken = Vec::new();
for (i, op) in updates.iter().enumerate() {
if op["was"]["status"] != expect {
broken.push(format!(
"op {i}: was {} after {}",
op["was"]["status"], expect
));
}
expect = op["set"]["status"].clone();
}
let g: Value = serde_json::from_str(&sb.dx_ok(&dir, &["graph"])).unwrap();
assert_eq!(
g["nodes"][0]["status"], expect,
"the last op is not the last write"
);
assert!(broken.is_empty(), "{} stale: {broken:?}", broken.len());
}
fn poison_stub() -> (String, std::sync::Arc<std::sync::Mutex<Vec<String>>>) {
let applied = std::sync::Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
let port = server.server_addr().to_ip().unwrap().port();
let seen = applied.clone();
std::thread::spawn(move || {
for mut req in server.incoming_requests() {
let mut body = String::new();
let _ = req.as_reader().read_to_string(&mut body);
let path = req.url().split('?').next().unwrap_or("").to_string();
let reply = match path.as_str() {
"/health" => "ok".to_string(),
"/claim" => r#"{"workspace":"stub","claim":"unchecked"}"#.to_string(),
"/locate" => r#"{"workspaces":[]}"#.to_string(),
"/ops" => {
let v: Value = serde_json::from_str(&body).unwrap();
let ops = v["ops"].as_array().unwrap();
if ops.iter().any(|op| op["title"] == "poison") {
let _ = req.respond(tiny_http::Response::empty(500));
continue;
}
let mut seen = seen.lock().unwrap();
let results: Vec<Value> = ops
.iter()
.map(|op| {
if let Some(t) = op["title"].as_str() {
seen.push(t.to_string());
}
serde_json::json!({"op_id": op["op_id"], "result": "applied"})
})
.collect();
serde_json::json!({"workspace": "stub", "results": results}).to_string()
}
_ => {
let _ = req.respond(tiny_http::Response::empty(404));
continue;
}
};
let _ = req.respond(tiny_http::Response::from_string(reply));
}
});
(format!("http://127.0.0.1:{port}"), applied)
}
#[test]
fn server_n1_an_op_the_server_fails_on_is_set_aside_and_the_rest_delivered() {
let (url, applied) = poison_stub();
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = sb.repo("poisoned");
std::fs::write(
dir.join(".deciduous").join("config.toml"),
format!("[remote]\nurl = \"{url}\"\nworkspace = \"poisoned\"\n"),
)
.unwrap();
sb.dx_ok(&dir, &["add", "goal", "before"]);
sb.dx_ok(&dir, &["add", "goal", "poison"]);
let out = sb.dx_ok(&dir, &["add", "goal", "after"]);
sb.dx_ok(&dir, &["add", "goal", "later"]);
assert_eq!(
*applied.lock().unwrap(),
["before", "after", "later"],
"{out}"
);
assert!(out.contains("poison") && out.contains("500"), "{out}");
let lines = log_lines(&dir);
let poison = lines
.iter()
.find(|l| l["title"] == "poison")
.expect("the op is kept");
let id = poison["op_id"].as_str().unwrap().to_string();
let ack = lines
.iter()
.find(|l| l["entry"] == "ack" && l["op_id"] == id.as_str())
.expect("and answered");
assert_eq!(ack["result"], "rejected");
assert!(ack["reason"].as_str().unwrap().contains("500"), "{ack}");
let retry = sb.dx(&dir, &["remote", "push", "--retry-rejected"]);
let retry = format!("{}{}", text(&retry.stdout), text(&retry.stderr));
assert!(retry.contains("1 rejected op(s)"), "{retry}");
assert!(log_lines(&dir).iter().any(|l| l["op_id"] == id.as_str()));
sb.dx_ok(&dir, &["add", "goal", "keep-waiting-offline"]);
let dropped = sb.dx_ok(&dir, &["remote", "push", "--drop", &id[..8]]);
assert!(dropped.contains("create goal"), "{dropped}");
assert!(!log_lines(&dir).iter().any(|l| l["op_id"] == id.as_str()));
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn server_n1_a_nul_in_one_write_does_not_stop_the_writes_after_it() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-poison");
let dir = sb.remote_repo("poison", &url, &ws);
let now = "2026-09-23T00:00:00Z";
let poison = serde_json::json!({"entry":"op","op_id":"poisoned-op-1","at":now,
"kind":"create_node","change_id":"0badc0de-0000-4000-8000-000000000001",
"node_type":"goal","title":"poison","status":"pending",
"metadata":{"prompt":"has\u{0}nul"},"created_at":now,"updated_at":now});
std::fs::write(log_path(&dir), format!("{poison}\n")).unwrap();
let said = sb.dx_ok(&dir, &["add", "goal", "after"]);
assert!(said.contains("NUL"), "the refusal names the cause: {said}");
assert_eq!(live_titles(&export(&url, &token, &ws)), ["after"]);
let st = sb.dx(&dir, &["remote", "status"]);
let st = text(&st.stdout);
assert!(st.contains("0 write(s) waiting, 0 rejected"), "{st}");
assert!(st.contains("settled"), "{st}");
}
impl Sandbox {
fn dx_db(&self, dir: &Path, db: &Path, args: &[&str]) -> Output {
Command::new(env!("CARGO_BIN_EXE_deciduous"))
.args(args)
.current_dir(dir)
.env("HOME", self.path().join("home"))
.env("XDG_CONFIG_HOME", self.path().join("home").join(".config"))
.env("DECIDUOUS_MCP_TOKEN", &self.token)
.env("DECIDUOUS_NO_SERVER", "1")
.env("DECIDUOUS_DB_PATH", db)
.env("NO_COLOR", "1")
.output()
.unwrap()
}
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn rust_n1_a_write_goes_to_its_databases_workspace_not_the_cwds() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let (ws1, ws3) = (unique("wal-p1"), unique("wal-p3"));
let p1 = sb.remote_repo("p1", &url, &ws1);
let p3 = sb.remote_repo("p3", &url, &ws3);
let db1 = p1.join(".deciduous").join("deciduous.db");
let out = sb.dx_db(&p3, &db1, &["add", "goal", "P1-SECRET-cli"]);
assert!(out.status.success(), "{}", text(&out.stderr));
sb.dx_db(&p3, &db1, &["status", "1", "completed"]);
assert_eq!(
live_titles(&export(&url, &token, &ws3)),
Vec::<String>::new()
);
let g1 = export(&url, &token, &ws1);
assert_eq!(server_node(&g1, "P1-SECRET-cli")["status"], "completed");
let st = sb.dx(&p1, &["remote", "status"]);
assert!(st.status.success(), "{}", text(&st.stdout));
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn bridge_n3_a_stdio_server_started_elsewhere_sends_to_its_databases_workspace() {
use std::io::{BufRead, BufReader, Write};
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-desktop");
let proj = sb.remote_repo("desktop", &url, &ws);
let elsewhere = sb.path().join("elsewhere");
std::fs::create_dir_all(&elsewhere).unwrap();
let mut child = Command::new(env!("CARGO_BIN_EXE_deciduous"))
.arg("mcp")
.current_dir(&elsewhere)
.env("HOME", sb.path().join("home"))
.env("XDG_CONFIG_HOME", sb.path().join("home").join(".config"))
.env("DECIDUOUS_MCP_TOKEN", &token)
.env("DECIDUOUS_NO_SERVER", "1")
.env(
"DECIDUOUS_DB_PATH",
proj.join(".deciduous").join("deciduous.db"),
)
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::null())
.spawn()
.unwrap();
let mut stdin = child.stdin.take().unwrap();
let mut stdout = BufReader::new(child.stdout.take().unwrap());
for (id, method, params) in [
(
1,
"initialize",
serde_json::json!({"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"t","version":"0"}}),
),
(
2,
"tools/call",
serde_json::json!({"name":"add_node","arguments":{"node_type":"goal","title":"desktop-style"}}),
),
] {
let msg = serde_json::json!({"jsonrpc":"2.0","id":id,"method":method,"params":params});
writeln!(stdin, "{msg}").unwrap();
stdin.flush().unwrap();
let mut line = String::new();
stdout.read_line(&mut line).unwrap();
}
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
let titles = loop {
let t = live_titles(&export(&url, &token, &ws));
if !t.is_empty() || std::time::Instant::now() > deadline {
break t;
}
std::thread::sleep(std::time::Duration::from_millis(100));
};
let _ = child.kill();
let _ = child.wait();
assert_eq!(titles, ["desktop-style"]);
}
fn black_hole() -> String {
let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = l.local_addr().unwrap().port();
std::thread::spawn(move || {
let mut held = Vec::new();
for s in l.incoming() {
held.push(s);
}
});
format!("http://127.0.0.1:{port}")
}
#[test]
fn rust_n2_a_black_hole_server_does_not_delay_stdio_replies() {
use std::io::{BufRead, BufReader, Write};
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = sb.repo("blackhole");
std::fs::write(
dir.join(".deciduous").join("config.toml"),
format!("[remote]\nurl = \"{}\"\nworkspace = \"bh\"\n", black_hole()),
)
.unwrap();
let mut child = Command::new(env!("CARGO_BIN_EXE_deciduous"))
.arg("mcp")
.current_dir(&dir)
.env("HOME", sb.path().join("home"))
.env("XDG_CONFIG_HOME", sb.path().join("home").join(".config"))
.env("DECIDUOUS_MCP_TOKEN", "0123456789abcdef0123456789abcdef")
.env("DECIDUOUS_NO_SERVER", "1")
.env_remove("DECIDUOUS_DB_PATH")
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::null())
.spawn()
.unwrap();
let mut stdin = child.stdin.take().unwrap();
let stdout = child.stdout.take().unwrap();
let (tx, rx) = std::sync::mpsc::channel::<String>();
std::thread::spawn(move || {
for line in BufReader::new(stdout).lines() {
if tx.send(line.unwrap()).is_err() {
break;
}
}
});
let mut calls = vec![(
"initialize",
serde_json::json!({"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"t","version":"0"}}),
)];
for i in 0..3 {
calls.push((
"tools/call",
serde_json::json!({"name":"add_node","arguments":{"node_type":"goal","title":format!("bh{i}")}}),
));
}
calls.push(("ping", serde_json::json!({})));
calls.push((
"tools/call",
serde_json::json!({"name":"list_nodes","arguments":{}}),
));
let mut slow = None;
for (id, (method, params)) in calls.iter().enumerate() {
let msg = serde_json::json!({"jsonrpc":"2.0","id":id,"method":method,"params":params});
writeln!(stdin, "{msg}").unwrap();
stdin.flush().unwrap();
let t = std::time::Instant::now();
if rx.recv_timeout(std::time::Duration::from_secs(5)).is_err() {
slow = Some(format!(
"call {id} ({method}) had no answer after {:?}",
t.elapsed()
));
break;
}
}
let _ = child.kill();
let _ = child.wait();
assert_eq!(slow, None);
assert_eq!(queued_titles(&dir), ["bh0", "bh1", "bh2"]);
}
#[test]
fn rust_n2_a_cli_write_to_a_black_hole_server_returns_promptly() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = sb.repo("blackhole-cli");
std::fs::write(
dir.join(".deciduous").join("config.toml"),
format!("[remote]\nurl = \"{}\"\nworkspace = \"bh\"\n", black_hole()),
)
.unwrap();
let t = std::time::Instant::now();
let (tx, rx) = std::sync::mpsc::channel();
let d = dir.clone();
std::thread::spawn(move || {
let _ = tx.send(sb.dx(&d, &["add", "goal", "bh-cli"]));
});
let out = rx
.recv_timeout(std::time::Duration::from_secs(20))
.unwrap_or_else(|_| panic!("deciduous add still running after {:?}", t.elapsed()));
assert!(out.status.success(), "{}", text(&out.stderr));
let err = text(&out.stderr);
assert!(
err.contains("did not get it") && err.contains("1 write(s) queued"),
"{err}"
);
assert_eq!(queued_titles(&dir), ["bh-cli"]);
}
fn refusing_stub() -> String {
let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
let port = server.server_addr().to_ip().unwrap().port();
std::thread::spawn(move || {
for mut req in server.incoming_requests() {
let mut body = String::new();
let _ = req.as_reader().read_to_string(&mut body);
let path = req.url().split('?').next().unwrap_or("").to_string();
let reply = match path.as_str() {
"/health" => "ok".to_string(),
"/claim" => r#"{"workspace":"stub","claim":"unchecked"}"#.to_string(),
"/locate" => r#"{"workspaces":[]}"#.to_string(),
"/ops" => {
let v: Value = serde_json::from_str(&body).unwrap();
let results: Vec<Value> = v["ops"]
.as_array()
.unwrap()
.iter()
.map(|op| {
if op["kind"] == "update_node" {
serde_json::json!({"op_id": op["op_id"], "result": "rejected",
"reason": format!("node {} was deleted on the server", op["change_id"].as_str().unwrap())})
} else {
serde_json::json!({"op_id": op["op_id"], "result": "applied"})
}
})
.collect();
serde_json::json!({"workspace": "stub", "results": results}).to_string()
}
_ => {
let _ = req.respond(tiny_http::Response::empty(404));
continue;
}
};
let _ = req.respond(tiny_http::Response::from_string(reply));
}
});
format!("http://127.0.0.1:{port}")
}
fn stdio_mcp(sb: &Sandbox, dir: &Path) -> (std::process::Child, impl FnMut(&str, Value) -> Value) {
use std::io::{BufRead, BufReader, Write};
let mut child = Command::new(env!("CARGO_BIN_EXE_deciduous"))
.arg("mcp")
.current_dir(dir)
.env("HOME", sb.path().join("home"))
.env("XDG_CONFIG_HOME", sb.path().join("home").join(".config"))
.env("DECIDUOUS_MCP_TOKEN", &sb.token)
.env("DECIDUOUS_NO_SERVER", "1")
.env_remove("DECIDUOUS_DB_PATH")
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::null())
.spawn()
.unwrap();
let mut stdin = child.stdin.take().unwrap();
let mut stdout = BufReader::new(child.stdout.take().unwrap());
let mut id = 0;
let call = move |method: &str, params: Value| {
id += 1;
let msg = serde_json::json!({"jsonrpc":"2.0","id":id,"method":method,"params":params});
writeln!(stdin, "{msg}").unwrap();
stdin.flush().unwrap();
let mut line = String::new();
stdout.read_line(&mut line).unwrap();
serde_json::from_str::<Value>(&line).unwrap()
};
(child, call)
}
fn tool_text(reply: &Value) -> String {
reply["result"]["content"]
.as_array()
.map(|c| {
c.iter()
.filter_map(|x| x["text"].as_str())
.collect::<Vec<_>>()
.join("\n")
})
.unwrap_or_default()
}
#[test]
fn rust_n7_a_write_the_server_refused_is_reported_in_a_tool_result() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = sb.repo("refused-mcp");
std::fs::write(
dir.join(".deciduous").join("config.toml"),
format!(
"[remote]\nurl = \"{}\"\nworkspace = \"r\"\n",
refusing_stub()
),
)
.unwrap();
let (mut child, mut call) = stdio_mcp(&sb, &dir);
call(
"initialize",
serde_json::json!({"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"t","version":"0"}}),
);
call(
"tools/call",
serde_json::json!({"name":"add_node","arguments":{"node_type":"goal","title":"g"}}),
);
call(
"tools/call",
serde_json::json!({"name":"update_status","arguments":{"node_id":1,"status":"completed"}}),
);
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
let mut seen = String::new();
while std::time::Instant::now() < deadline && !seen.contains("refused") {
seen = tool_text(&call(
"tools/call",
serde_json::json!({"name":"list_nodes","arguments":{}}),
));
std::thread::sleep(std::time::Duration::from_millis(100));
}
let _ = child.kill();
let _ = child.wait();
assert!(
seen.contains("refused") && seen.contains("was deleted on the server"),
"no tool result said so: {seen}"
);
}
#[test]
fn rust_n7_a_write_that_could_not_be_queued_is_an_error_result() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = offline_repo(&sb, "unqueued-mcp");
std::fs::create_dir_all(log_path(&dir)).unwrap();
let (mut child, mut call) = stdio_mcp(&sb, &dir);
call(
"initialize",
serde_json::json!({"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"t","version":"0"}}),
);
let r = call(
"tools/call",
serde_json::json!({"name":"add_node","arguments":{"node_type":"goal","title":"unqueued"}}),
);
let _ = child.kill();
let _ = child.wait();
assert_eq!(r["result"]["isError"], true, "{r}");
let t = tool_text(&r);
assert!(t.to_lowercase().contains("not queued"), "{t}");
}
#[test]
fn rust_n7_an_offline_stdio_server_says_it_is_offline_once() {
use std::io::{BufRead, BufReader, Read, Write};
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = offline_repo(&sb, "quiet");
let mut child = Command::new(env!("CARGO_BIN_EXE_deciduous"))
.arg("mcp")
.current_dir(&dir)
.env("HOME", sb.path().join("home"))
.env("XDG_CONFIG_HOME", sb.path().join("home").join(".config"))
.env("DECIDUOUS_MCP_TOKEN", &sb.token)
.env("DECIDUOUS_NO_SERVER", "1")
.env_remove("DECIDUOUS_DB_PATH")
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.spawn()
.unwrap();
let mut stdin = child.stdin.take().unwrap();
let mut stdout = BufReader::new(child.stdout.take().unwrap());
let mut send = |id: usize, method: &str, params: Value| {
let msg = serde_json::json!({"jsonrpc":"2.0","id":id,"method":method,"params":params});
writeln!(stdin, "{msg}").unwrap();
stdin.flush().unwrap();
let mut line = String::new();
stdout.read_line(&mut line).unwrap();
};
send(
0,
"initialize",
serde_json::json!({"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"t","version":"0"}}),
);
for i in 1..=20 {
send(
i,
"tools/call",
serde_json::json!({"name":"add_node","arguments":{"node_type":"goal","title":format!("q{i}")}}),
);
std::thread::sleep(std::time::Duration::from_millis(30));
}
drop(stdin);
let mut err = String::new();
child
.stderr
.take()
.unwrap()
.read_to_string(&mut err)
.unwrap();
child.wait().unwrap();
assert_eq!(queued_titles(&dir).len(), 20);
assert_eq!(err.matches("did not get it").count(), 1, "{err}");
}
fn log_ops(dir: &Path) -> Vec<String> {
log_lines(dir)
.iter()
.filter(|l| l["entry"] == "op")
.map(|l| {
let what = l["title"]
.as_str()
.map(str::to_string)
.unwrap_or_else(|| l["change_id"].as_str().unwrap_or("").to_string());
format!("{} {what}", l["kind"].as_str().unwrap())
})
.collect()
}
#[test]
fn new_wal_a_write_whose_append_failed_is_queued_by_the_next_write() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = offline_repo(&sb, "wal-append");
sb.dx_ok(&dir, &["add", "goal", "doomed"]);
let doomed = local_change_id(&sb, &dir, 1);
let log = log_path(&dir);
let aside = dir.join(".deciduous").join("log.bak");
std::fs::rename(&log, &aside).unwrap();
std::fs::create_dir_all(&log).unwrap();
let _ = sb.dx(&dir, &["add", "goal", "lost-create"]);
let _ = sb.dx(&dir, &["delete", "1"]);
std::fs::remove_dir(&log).unwrap();
std::fs::rename(&aside, &log).unwrap();
sb.dx_ok(&dir, &["add", "goal", "after"]);
assert_eq!(
log_ops(&dir),
[
"create_node doomed".to_string(),
"create_node lost-create".to_string(),
format!("delete_node {doomed}"),
"create_node after".to_string(),
]
);
}
#[test]
fn new_wal_a_killed_writer_leaves_no_local_write_without_its_op() {
use std::io::{BufRead, BufReader, Write};
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = offline_repo(&sb, "wal-kill");
for i in 0..40 {
sb.dx_ok(&dir, &["add", "goal", &format!("seed{i}")]);
}
let mut next_delete = 1;
for trial in 0..25 {
let mut child = Command::new(env!("CARGO_BIN_EXE_deciduous"))
.arg("mcp")
.current_dir(&dir)
.env("HOME", sb.path().join("home"))
.env("XDG_CONFIG_HOME", sb.path().join("home").join(".config"))
.env("DECIDUOUS_MCP_TOKEN", &sb.token)
.env("DECIDUOUS_NO_SERVER", "1")
.env_remove("DECIDUOUS_DB_PATH")
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::null())
.spawn()
.unwrap();
let mut stdin = child.stdin.take().unwrap();
writeln!(stdin, "{}", serde_json::json!({"jsonrpc":"2.0","id":0,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"t","version":"0"}}})).unwrap();
let mut stdout = BufReader::new(child.stdout.take().unwrap());
let mut line = String::new();
stdout.read_line(&mut line).unwrap();
let mut msgs = Vec::new();
for i in 0..3 {
msgs.push(serde_json::json!({"jsonrpc":"2.0","id":10 + i,"method":"tools/call","params":{"name":"add_node","arguments":{"node_type":"goal","title":format!("k{trial}-{i}")}}}));
}
for i in 0..2 {
msgs.push(serde_json::json!({"jsonrpc":"2.0","id":20 + i,"method":"tools/call","params":{"name":"delete_node","arguments":{"node_id":next_delete}}}));
next_delete += 1;
}
for m in msgs {
let _ = writeln!(stdin, "{m}");
}
let _ = stdin.flush();
let jitter = (std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.subsec_nanos()
% 80) as u64;
std::thread::sleep(std::time::Duration::from_millis(jitter));
let _ = child.kill();
let _ = child.wait();
}
sb.dx_ok(&dir, &["add", "goal", "final"]);
let g: Value = serde_json::from_str(&sb.dx_ok(&dir, &["graph"])).unwrap();
let local: std::collections::HashSet<String> = g["nodes"]
.as_array()
.unwrap()
.iter()
.map(|n| n["change_id"].as_str().unwrap().to_string())
.collect();
let lines = log_lines(&dir);
let ops = |kind: &str| -> std::collections::HashSet<String> {
lines
.iter()
.filter(|l| l["kind"] == kind)
.map(|l| l["change_id"].as_str().unwrap().to_string())
.collect()
};
let (created, deleted) = (ops("create_node"), ops("delete_node"));
let no_create: Vec<&String> = local.iter().filter(|c| !created.contains(*c)).collect();
let no_delete: Vec<&String> = created
.iter()
.filter(|c| !local.contains(*c) && !deleted.contains(*c))
.collect();
assert!(
no_create.is_empty() && no_delete.is_empty(),
"{} node(s) here with no create op, {} deleted here with no delete op",
no_create.len(),
no_delete.len()
);
}
#[test]
fn rust_n1_a_db_path_with_no_directory_part_still_queues() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = offline_repo(&sb, "bare-db-path");
let inside = dir.join(".deciduous");
let out = sb.dx_db(
&inside,
Path::new("deciduous.db"),
&["add", "goal", "relative-db-path"],
);
assert!(out.status.success(), "{}", all_of(&out));
assert_eq!(
queued_titles(&dir),
["relative-db-path"],
"{}",
all_of(&out)
);
}
fn all_of(out: &Output) -> String {
format!("{}{}", text(&out.stdout), text(&out.stderr))
}
#[allow(clippy::type_complexity)]
fn scripted_stub(
fail: impl Fn(usize, &[Value]) -> Option<(u16, u64)> + Send + 'static,
) -> (
String,
std::sync::Arc<std::sync::Mutex<Vec<String>>>,
std::sync::Arc<std::sync::atomic::AtomicBool>,
) {
let applied = std::sync::Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
let healthy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
let port = server.server_addr().to_ip().unwrap().port();
let (seen, ok) = (applied.clone(), healthy.clone());
std::thread::spawn(move || {
let mut n = 0;
for mut req in server.incoming_requests() {
let mut body = String::new();
let _ = req.as_reader().read_to_string(&mut body);
let path = req.url().split('?').next().unwrap_or("").to_string();
let reply = match path.as_str() {
"/health" => "ok".to_string(),
"/claim" => r#"{"workspace":"stub","claim":"unchecked"}"#.to_string(),
"/locate" => r#"{"workspaces":[]}"#.to_string(),
"/ops" => {
n += 1;
let v: Value = serde_json::from_str(&body).unwrap();
let ops = v["ops"].as_array().unwrap().clone();
if !ok.load(std::sync::atomic::Ordering::SeqCst) {
if let Some((status, delay)) = fail(n, &ops) {
std::thread::sleep(std::time::Duration::from_millis(delay));
let _ = req.respond(tiny_http::Response::empty(status));
continue;
}
}
let mut seen = seen.lock().unwrap();
let results: Vec<Value> = ops
.iter()
.map(|op| {
seen.push(match op["title"].as_str() {
Some(t) => t.to_string(),
None => format!(
"{} {}",
op["kind"].as_str().unwrap_or(""),
op["change_id"].as_str().unwrap_or("")
),
});
serde_json::json!({"op_id": op["op_id"], "result": "applied"})
})
.collect();
serde_json::json!({"workspace": "stub", "results": results}).to_string()
}
_ => {
let _ = req.respond(tiny_http::Response::empty(404));
continue;
}
};
let _ = req.respond(tiny_http::Response::from_string(reply));
}
});
(format!("http://127.0.0.1:{port}"), applied, healthy)
}
#[test]
fn new_a_transient_500_sets_no_op_aside() {
let (url, applied, _) = scripted_stub(|n, _| (n <= 2).then_some((500, 0)));
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = offline_repo(&sb, "transient");
sb.dx_ok(&dir, &["add", "goal", "casc"]);
sb.dx_ok(&dir, &["status", "1", "completed"]);
sb.dx_ok(&dir, &["prompt", "1", "why"]);
set_remote_url(&dir, &url);
let out = sb.dx(&dir, &["remote", "push"]);
let said = all_of(&out);
assert!(said.contains("0 rejected"), "{said}");
assert!(out.status.success(), "{said}");
assert_eq!(applied.lock().unwrap().len(), 3, "{said}");
}
#[test]
fn server_n1_an_outage_during_a_replay_sets_no_op_aside() {
let (url, applied, healthy) =
scripted_stub(|n, ops| (ops.len() > 1 || n > 2).then_some((500, 0)));
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = offline_repo(&sb, "outage");
for t in ["a", "b", "c", "d", "e"] {
sb.dx_ok(&dir, &["add", "goal", t]);
}
sb.dx_ok(&dir, &["link", "1", "2"]);
set_remote_url(&dir, &url);
let out = sb.dx(&dir, &["remote", "push"]);
let said = all_of(&out);
let st = log_lines(&dir);
let rejected = st
.iter()
.filter(|l| l["entry"] == "ack" && l["result"] == "rejected")
.count();
assert_eq!(rejected, 0, "{said}");
assert!(
!out.status.success(),
"a push that left writes waiting: {said}"
);
assert!(!said.contains("--drop-rejected"), "{said}");
healthy.store(true, std::sync::atomic::Ordering::SeqCst);
let again = sb.dx(&dir, &["remote", "push"]);
assert!(again.status.success(), "{}", all_of(&again));
let mut got = applied.lock().unwrap().clone();
got.retain(|t| t.len() == 1);
got.sort();
assert_eq!(got, ["a", "b", "c", "d", "e"], "{}", all_of(&again));
assert!(log_ops(&dir).is_empty(), "{:?}", log_ops(&dir));
}
#[test]
fn new_the_edits_of_a_set_aside_node_wait_with_it() {
let (url, applied, _) = scripted_stub(|_, ops| {
ops.iter()
.any(|op| op["title"] == "poison")
.then_some((500, 0))
});
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = offline_repo(&sb, "held");
sb.dx_ok(&dir, &["add", "goal", "before"]);
sb.dx_ok(&dir, &["add", "goal", "poison"]);
sb.dx_ok(&dir, &["status", "2", "completed"]);
sb.dx_ok(&dir, &["add", "goal", "after"]);
set_remote_url(&dir, &url);
let out = sb.dx(&dir, &["remote", "push"]);
let said = all_of(&out);
assert_eq!(*applied.lock().unwrap(), ["before", "after"], "{said}");
assert!(!out.status.success(), "writes were set aside: {said}");
let ops = log_ops(&dir);
assert_eq!(ops.len(), 2, "the create and its edit both kept: {ops:?}");
}
#[test]
fn rust_n2_a_cli_write_to_a_failing_server_is_bounded() {
let (url, _, _) = scripted_stub(|_, _| Some((500, 1500)));
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = offline_repo(&sb, "slow500");
for i in 0..30 {
sb.dx_ok(&dir, &["add", "goal", &format!("q{i}")]);
}
set_remote_url(&dir, &url);
let t = std::time::Instant::now();
let out = sb.dx(&dir, &["add", "goal", "while-db-down"]);
let took = t.elapsed();
assert!(out.status.success(), "{}", all_of(&out));
assert!(
took < std::time::Duration::from_secs(15),
"took {took:?}: {}",
all_of(&out)
);
assert_eq!(queued_titles(&dir).len(), 31);
}
#[test]
fn rust_n2_a_legacy_config_write_to_a_black_hole_is_bounded() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = sb.repo("legacy-bh");
std::fs::write(
dir.join(".deciduous").join("config.toml"),
format!("[remote]\nurl = \"{}\"\n", black_hole()),
)
.unwrap();
let t = std::time::Instant::now();
let out = sb.dx(&dir, &["add", "goal", "first"]);
let took = t.elapsed();
assert!(out.status.success(), "{}", all_of(&out));
assert!(
took < std::time::Duration::from_secs(15),
"took {took:?}: {}",
all_of(&out)
);
assert_eq!(queued_titles(&dir), ["first"]);
}
#[test]
fn new_a_stdio_server_exits_promptly_when_stdin_closes() {
use std::io::{BufRead, BufReader, Write};
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = sb.repo("bh-exit");
std::fs::write(
dir.join(".deciduous").join("config.toml"),
format!("[remote]\nurl = \"{}\"\nworkspace = \"bh\"\n", black_hole()),
)
.unwrap();
let mut child = Command::new(env!("CARGO_BIN_EXE_deciduous"))
.arg("mcp")
.current_dir(&dir)
.env("HOME", sb.path().join("home"))
.env("XDG_CONFIG_HOME", sb.path().join("home").join(".config"))
.env("DECIDUOUS_MCP_TOKEN", &sb.token)
.env("DECIDUOUS_NO_SERVER", "1")
.env_remove("DECIDUOUS_DB_PATH")
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::null())
.spawn()
.unwrap();
let mut stdin = child.stdin.take().unwrap();
let mut stdout = BufReader::new(child.stdout.take().unwrap());
for (id, (m, p)) in [
("initialize", serde_json::json!({"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"t","version":"0"}})),
("tools/call", serde_json::json!({"name":"add_node","arguments":{"node_type":"goal","title":"x1"}})),
("tools/call", serde_json::json!({"name":"add_node","arguments":{"node_type":"goal","title":"x2"}})),
]
.into_iter()
.enumerate()
{
writeln!(stdin, "{}", serde_json::json!({"jsonrpc":"2.0","id":id,"method":m,"params":p})).unwrap();
stdin.flush().unwrap();
let mut line = String::new();
stdout.read_line(&mut line).unwrap();
}
drop(stdin);
let t = std::time::Instant::now();
let mut exited = None;
while t.elapsed() < std::time::Duration::from_secs(15) {
if let Some(s) = child.try_wait().unwrap() {
exited = Some((s, t.elapsed()));
break;
}
std::thread::sleep(std::time::Duration::from_millis(50));
}
let _ = child.kill();
let _ = child.wait();
let (_, took) = exited.expect("still running 15 s after stdin closed");
assert!(
took < std::time::Duration::from_secs(5),
"exited after {took:?}"
);
assert_eq!(queued_titles(&dir), ["x1", "x2"]);
}
fn add_with_prompt(sb: &Sandbox, dir: &Path, title: &str, prompt: &[u8]) -> Output {
use std::io::Write;
let mut child = Command::new(env!("CARGO_BIN_EXE_deciduous"))
.args(["add", "goal", title, "--prompt-stdin"])
.current_dir(dir)
.env("HOME", sb.path().join("home"))
.env("XDG_CONFIG_HOME", sb.path().join("home").join(".config"))
.env("DECIDUOUS_MCP_TOKEN", &sb.token)
.env("DECIDUOUS_NO_SERVER", "1")
.env("NO_COLOR", "1")
.env_remove("DECIDUOUS_DB_PATH")
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.spawn()
.unwrap();
child.stdin.take().unwrap().write_all(prompt).unwrap();
child.wait_with_output().unwrap()
}
#[test]
fn new_a_write_holding_a_nul_is_refused_before_it_is_made() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = offline_repo(&sb, "nul-write");
let out = add_with_prompt(&sb, &dir, "poison", b"has\0nul");
let said = all_of(&out);
assert!(!out.status.success(), "{said}");
assert!(said.contains("NUL") && said.contains("prompt"), "{said}");
let g: Value = serde_json::from_str(&sb.dx_ok(&dir, &["graph"])).unwrap();
assert_eq!(g["nodes"].as_array().unwrap().len(), 0, "nothing written");
assert!(queued_titles(&dir).is_empty());
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn new_a_nul_row_does_not_stop_seed_sending_the_others() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-nulseed");
let dir = sb.repo("nulseed");
let out = add_with_prompt(&sb, &dir, "poison", b"has\0nul");
assert!(out.status.success(), "{}", all_of(&out));
sb.dx_ok(&dir, &["add", "goal", "pre-remote-row"]);
sb.dx_ok(&dir, &["remote", "init", &url, "--workspace", &ws]);
let st = all_of(&sb.dx(&dir, &["remote", "status"]));
let poison_line = st
.lines()
.find(|l| l.contains("\"poison\""))
.unwrap_or_else(|| panic!("{st}"));
assert!(
poison_line.contains("NUL") && !poison_line.contains("--seed"),
"{st}"
);
let seed = sb.dx(&dir, &["remote", "push", "--seed"]);
let said = all_of(&seed);
assert!(!seed.status.success(), "a row was not sent: {said}");
assert!(said.contains("poison") && said.contains("NUL"), "{said}");
assert_eq!(
live_titles(&export(&url, &token, &ws)),
["pre-remote-row"],
"{said}"
);
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn new_status_says_what_to_do_about_unreadable_lines() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("wal-unreadable");
let dir = sb.remote_repo("unreadable", &url, &ws);
sb.dx_ok(&dir, &["add", "goal", "fine"]);
std::fs::write(
dir.join(".deciduous").join("remote-log.unreadable"),
"{\"entry\":\"op\",\"op_\n",
)
.unwrap();
let out = sb.dx(&dir, &["remote", "status"]);
let st = all_of(&out);
assert!(!out.status.success(), "{st}");
assert!(!st.contains("were moved to"), "{st}");
assert!(!st.contains("sends what is waiting"), "{st}");
assert!(st.contains("remote-log.unreadable"), "{st}");
}
#[test]
fn new8_remote_init_keeps_the_updates_comments_with_updates() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = sb.repo("cfg");
let path = dir.join(".deciduous").join("config.toml");
let before = std::fs::read_to_string(&path).unwrap();
assert!(
before.contains("[updates]\n# Version checking"),
"init's config changed shape; this test needs updating:\n{before}"
);
let url = stub_server(serde_json::json!({"nodes": [], "edges": []}));
sb.dx_ok(&dir, &["remote", "init", &url, "--workspace", "cfg"]);
let after = std::fs::read_to_string(&path).unwrap();
assert!(
after.starts_with(&before),
"remote init changed what was already in config.toml:\n--- before\n{before}\n--- after\n{after}"
);
let added = &after[before.len()..];
assert!(
added.contains("[remote]") && added.contains(&url) && added.contains("workspace = \"cfg\""),
"remote init did not append [remote]:\n{after}"
);
}
#[test]
fn new5_pull_of_unchanged_content_leaves_graph_json_alone() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = sb.repo("stamp");
std::fs::write(
dir.join(".deciduous").join("config.toml"),
format!(
"[remote]\nurl = \"{}\"\nworkspace = \"stamp\"\n",
dead_url()
),
)
.unwrap();
sb.dx_ok(&dir, &["add", "goal", "written online", "-c", "80"]);
sb.dx_ok(&dir, &["status", "1", "active"]);
let cid = local_change_id(&sb, &dir, 1);
let graph_file = dir.join(".deciduous").join("graph.json");
let before = std::fs::read_to_string(&graph_file).unwrap();
let doc: Value = serde_json::from_str(&before).unwrap();
let rec = &doc["nodes"][&cid];
assert!(rec.is_object(), "no record for {cid} in {doc}");
let local_stamp = chrono::DateTime::parse_from_rfc3339(rec["updated_at"].as_str().unwrap())
.unwrap()
.with_timezone(&chrono::Utc);
let applied = local_stamp + chrono::Duration::milliseconds(50);
let server_node = serde_json::json!({
"id": format!("srv-{cid}"), "change_id": cid, "node_type": rec["node_type"],
"title": rec["title"], "description": rec.get("description").cloned().unwrap_or(Value::Null),
"status": rec["status"], "metadata": rec["metadata"], "created_at": rec["created_at"],
"updated_at": applied.to_rfc3339_opts(chrono::SecondsFormat::Micros, true),
"deleted_at": null
});
let url = stub_server(serde_json::json!({
"nodes": [server_node], "edges": [], "documents": []
}));
set_remote_url(&dir, &url);
let out = sb.dx_ok(&dir, &["remote", "pull"]);
let after = std::fs::read_to_string(&graph_file).unwrap();
assert_eq!(
before, after,
"a pull that changed no content rewrote graph.json (pull said: {out})"
);
assert!(!out.contains("updated 1"), "{out}");
let changed = server_node_with(&cid, rec, "completed", applied);
let url = stub_server(serde_json::json!({
"nodes": [changed], "edges": [], "documents": []
}));
set_remote_url(&dir, &url);
sb.dx_ok(&dir, &["remote", "pull"]);
let doc: Value = serde_json::from_str(&std::fs::read_to_string(&graph_file).unwrap()).unwrap();
assert_eq!(doc["nodes"][&cid]["status"], "completed");
}
#[test]
fn new5_pull_of_unchanged_content_leaves_graph_json_alone_without_a_metadata_key() {
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let dir = sb.path().join("unborn");
std::fs::create_dir_all(&dir).unwrap();
sb.git(&dir, &["init", "-q", "-b", "main"]);
sb.dx_ok(&dir, &["init"]);
std::fs::write(
dir.join(".deciduous").join("config.toml"),
format!(
"[remote]\nurl = \"{}\"\nworkspace = \"stamp\"\n",
dead_url()
),
)
.unwrap();
sb.dx_ok(&dir, &["add", "action", "A1"]);
sb.dx_ok(&dir, &["status", "1", "completed"]);
let cid = local_change_id(&sb, &dir, 1);
let graph_file = dir.join(".deciduous").join("graph.json");
let before = std::fs::read_to_string(&graph_file).unwrap();
let doc: Value = serde_json::from_str(&before).unwrap();
let rec = &doc["nodes"][&cid];
assert!(rec.is_object(), "no record for {cid} in {doc}");
assert!(
rec.get("metadata").is_none(),
"the record has a metadata key, so this test proves nothing: {rec}"
);
let local_stamp = chrono::DateTime::parse_from_rfc3339(rec["updated_at"].as_str().unwrap())
.unwrap()
.with_timezone(&chrono::Utc);
let applied = local_stamp + chrono::Duration::milliseconds(54);
let mut server_node = server_node_with(&cid, rec, "completed", applied);
server_node["metadata"] = serde_json::json!({});
let url = stub_server(serde_json::json!({
"nodes": [server_node], "edges": [], "documents": []
}));
set_remote_url(&dir, &url);
let out = sb.dx_ok(&dir, &["remote", "pull"]);
let after = std::fs::read_to_string(&graph_file).unwrap();
assert_eq!(
before, after,
"a pull that changed no content rewrote graph.json (pull said: {out})"
);
let check = sb.dx(&dir, &["sync", "--check"]);
assert!(
check.status.success(),
"sync --check after the pull wants to do something:\n{}{}",
text(&check.stdout),
text(&check.stderr)
);
let synced = sb.dx_ok(&dir, &["sync"]);
assert_eq!(
before,
std::fs::read_to_string(&graph_file).unwrap(),
"sync after the pull rewrote graph.json: {synced}"
);
}
fn server_node_with(
cid: &str,
rec: &Value,
status: &str,
at: chrono::DateTime<chrono::Utc>,
) -> Value {
serde_json::json!({
"id": format!("srv-{cid}"), "change_id": cid, "node_type": rec["node_type"],
"title": rec["title"], "description": rec.get("description").cloned().unwrap_or(Value::Null),
"status": status, "metadata": rec["metadata"], "created_at": rec["created_at"],
"updated_at": at.to_rfc3339_opts(chrono::SecondsFormat::Micros, true),
"deleted_at": null
})
}
type Deleted = std::sync::Arc<std::sync::Mutex<std::collections::HashMap<String, String>>>;
type Seen = std::sync::Arc<std::sync::Mutex<Vec<String>>>;
fn dating_stub() -> (String, Deleted, Seen) {
use chrono::DateTime;
let deleted: Deleted = Default::default();
let seen: Seen = Default::default();
let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
let port = server.server_addr().to_ip().unwrap().port();
let (dead, log) = (deleted.clone(), seen.clone());
std::thread::spawn(move || {
for mut req in server.incoming_requests() {
let mut body = String::new();
let _ = req.as_reader().read_to_string(&mut body);
let path = req.url().split('?').next().unwrap_or("").to_string();
let reply = match path.as_str() {
"/health" => "ok".to_string(),
"/claim" => r#"{"workspace":"stub","claim":"unchecked"}"#.to_string(),
"/locate" => r#"{"workspaces":[]}"#.to_string(),
"/ops" => {
let v: Value = serde_json::from_str(&body).unwrap();
let mut dead = dead.lock().unwrap();
let mut log = log.lock().unwrap();
let results: Vec<Value> = v["ops"]
.as_array()
.unwrap()
.iter()
.map(|op| {
let kind = op["kind"].as_str().unwrap_or("");
let cid = op["change_id"].as_str().unwrap_or("").to_string();
let at = op["at"].as_str().unwrap_or("").to_string();
let (result, reason) = match (kind, dead.get(&cid)) {
("create_node", Some(_)) => ("rejected", "deleted".to_string()),
("update_node", Some(d)) => {
let made = DateTime::parse_from_rfc3339(&at).unwrap();
let d = DateTime::parse_from_rfc3339(d).unwrap();
if made > d {
dead.remove(&cid);
("applied", String::new())
} else {
(
"rejected",
format!(
"node {cid} was deleted on the server at {d}, \
after this edit was made at {made}"
),
)
}
}
("delete_node", _) => {
dead.insert(cid.clone(), at.clone());
("applied", String::new())
}
_ => ("applied", String::new()),
};
log.push(format!("{kind} {cid} {result} {at}"));
serde_json::json!({"op_id": op["op_id"], "result": result, "reason": reason})
})
.collect();
serde_json::json!({"workspace": "stub", "results": results}).to_string()
}
_ => {
let _ = req.respond(tiny_http::Response::empty(404));
continue;
}
};
let _ = req.respond(tiny_http::Response::from_string(reply));
}
});
(format!("http://127.0.0.1:{port}"), deleted, seen)
}
fn alice_and_bob(sb: &Sandbox, url: &str, workspace: &str) -> (PathBuf, PathBuf) {
let alice = sb.repo("alice");
sb.git(
&alice,
&["add", ".deciduous/config.toml", ".deciduous/graph.json"],
);
sb.git(
&alice,
&["commit", "-q", "--allow-empty", "-m", "deciduous"],
);
let bob = sb.path().join("bob");
sb.git(
sb.path(),
&[
"clone",
"-q",
alice.to_str().unwrap(),
bob.to_str().unwrap(),
],
);
std::fs::write(
bob.join(".deciduous").join("config.toml"),
format!("[remote]\nurl = \"{url}\"\nworkspace = \"{workspace}\"\n"),
)
.unwrap();
(alice, bob)
}
fn commit_graph(sb: &Sandbox, dir: &Path, msg: &str) {
sb.git(dir, &["add", ".deciduous/graph.json"]);
sb.git(dir, &["commit", "-q", "-m", msg]);
}
fn pull_and_sync(sb: &Sandbox, dir: &Path) -> String {
sb.git(dir, &["pull", "-q", "--no-rebase"]);
sb.dx_ok(dir, &["sync"])
}
fn graph_record(dir: &Path, cid: &str) -> Value {
let g: Value = serde_json::from_str(
&std::fs::read_to_string(dir.join(".deciduous").join("graph.json")).unwrap(),
)
.unwrap();
g["nodes"]
.get(cid)
.unwrap_or_else(|| panic!("no record {cid} in graph.json: {g}"))
.clone()
}
fn same_instant(a: &str, b: &str) -> bool {
chrono::DateTime::parse_from_rfc3339(a).unwrap()
== chrono::DateTime::parse_from_rfc3339(b).unwrap()
}
#[test]
fn a_git_edit_made_before_a_server_delete_does_not_revive_the_node() {
let (url, deleted, seen) = dating_stub();
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let (alice, bob) = alice_and_bob(&sb, &url, "dating");
sb.dx_ok(&alice, &["add", "goal", "deleted by an agent"]);
commit_graph(&sb, &alice, "add");
pull_and_sync(&sb, &bob);
let cid = local_change_id(&sb, &alice, 1);
sb.dx_ok(&alice, &["status", "1", "completed"]);
commit_graph(&sb, &alice, "before the delete");
let edited = graph_record(&alice, &cid)["updated_at"]
.as_str()
.unwrap()
.to_string();
std::thread::sleep(std::time::Duration::from_millis(20));
let deleted_at = chrono::Utc::now().to_rfc3339();
deleted
.lock()
.unwrap()
.insert(cid.clone(), deleted_at.clone());
std::thread::sleep(std::time::Duration::from_millis(20));
pull_and_sync(&sb, &bob);
let last = seen.lock().unwrap().last().cloned().unwrap_or_default();
assert!(
last.starts_with(&format!("update_node {cid} rejected ")),
"an edit made at {edited}, before the delete at {deleted_at}, reached the server as: {last}"
);
let at = last.rsplit(' ').next().unwrap();
assert!(
same_instant(at, &edited),
"the op is dated {at}; alice made the edit at {edited}"
);
sb.dx_ok(&alice, &["status", "1", "rejected"]);
commit_graph(&sb, &alice, "after the delete");
pull_and_sync(&sb, &bob);
let last = seen.lock().unwrap().last().cloned().unwrap_or_default();
assert!(
last.starts_with(&format!("update_node {cid} applied ")),
"an edit made after the delete at {deleted_at} was refused: {last}"
);
assert!(!deleted.lock().unwrap().contains_key(&cid));
}
#[test]
fn a_git_delete_is_dated_when_it_was_made() {
let (url, _deleted, seen) = dating_stub();
let sb = Sandbox::new("0123456789abcdef0123456789abcdef");
let (alice, bob) = alice_and_bob(&sb, &url, "dating-delete");
sb.dx_ok(&alice, &["add", "goal", "deleted through git"]);
commit_graph(&sb, &alice, "add");
pull_and_sync(&sb, &bob);
let cid = local_change_id(&sb, &alice, 1);
sb.dx_ok(&alice, &["delete", "1"]);
commit_graph(&sb, &alice, "delete");
let made = graph_record(&alice, &cid)["deleted_at"]
.as_str()
.unwrap()
.to_string();
std::thread::sleep(std::time::Duration::from_millis(20));
pull_and_sync(&sb, &bob);
let last = seen.lock().unwrap().last().cloned().unwrap_or_default();
assert!(
last.starts_with(&format!("delete_node {cid} applied ")),
"{last}"
);
let at = last.rsplit(' ').next().unwrap();
assert!(
same_instant(at, &made),
"the tombstone is dated {at}; alice deleted the node at {made}"
);
}
#[test]
#[ignore = "needs a real server: set DECIDUOUS_TEST_SERVER and DECIDUOUS_TEST_TOKEN"]
fn a_git_edit_made_before_an_agents_delete_does_not_revive_the_node_on_the_server() {
let (url, token) = server();
let sb = Sandbox::new(&token);
let ws = unique("git-dating");
let (alice, bob) = alice_and_bob(&sb, &url, &ws);
sb.dx_ok(&alice, &["add", "goal", "deleted by an agent"]);
commit_graph(&sb, &alice, "add");
pull_and_sync(&sb, &bob);
let id = server_id(&export(&url, &token, &ws), "deleted by an agent");
sb.dx_ok(&alice, &["status", "1", "completed"]);
commit_graph(&sb, &alice, "before the delete");
std::thread::sleep(std::time::Duration::from_millis(20));
mcp(
&url,
&token,
&ws,
&[("delete_node", serde_json::json!({"node_id": id}))],
);
let out = pull_and_sync(&sb, &bob);
assert!(
live_titles(&export(&url, &token, &ws)).is_empty(),
"an edit made before the delete brought the node back (sync said: {out})"
);
sb.dx_ok(&alice, &["prompt", "1", "made after the delete"]);
commit_graph(&sb, &alice, "after the delete");
let out = pull_and_sync(&sb, &bob);
let g = export(&url, &token, &ws);
assert_eq!(
live_titles(&g),
["deleted by an agent"],
"an edit made after the delete did not bring the node back (sync said: {out})"
);
assert_eq!(
server_node(&g, "deleted by an agent")["metadata"]["prompt"],
"made after the delete"
);
}