diff --git a/Cargo.lock b/Cargo.lock index a415a6d..d81ebc4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2873,6 +2873,7 @@ dependencies = [ "pin-project-lite", "smallvec", "tokio", + "want", ] [[package]] @@ -2881,13 +2882,22 @@ version = "0.1.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ddc03d96684f9226b8a787cdb71488417b53ab5ea8fdb1dac946cb9431cc8bff" dependencies = [ + "base64 0.23.1", "bytes", + "futures-channel", + "futures-util", "http", "http-body", + "httparse", "hyper", + "ipnet", + "libc", + "percent-encoding", "pin-project-lite", + "socket2", "tokio", "tower-service", + "tracing", ] [[package]] @@ -3226,6 +3236,12 @@ dependencies = [ "hybrid-array", ] +[[package]] +name = "ipnet" +version = "2.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "791930b43c0d5973160d90a8f3894509f2b273430f5c5c73b668636d0287c5c0" + [[package]] name = "is-docker" version = "0.2.0" @@ -5358,6 +5374,40 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19b30a45b0cd0bcca8037f3d0dc3421eaf95327a17cad11964fb8179b4fc4832" +[[package]] +name = "reqwest" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "16a1cfa75cc186dd73d5818e510e042e40927bccc9c236b061cea97e1eb08029" +dependencies = [ + "base64 0.23.1", + "bytes", + "futures-core", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "js-sys", + "log", + "percent-encoding", + "pin-project-lite", + "serde", + "serde_json", + "sync_wrapper", + "tokio", + "tokio-util", + "tower", + "tower-http", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "wasm-streams", + "web-sys", +] + [[package]] name = "resvg" version = "0.47.0" @@ -5402,6 +5452,7 @@ dependencies = [ "pastey", "pin-project-lite", "rand 0.10.3", + "reqwest", "rmcp-macros", "schemars", "serde", @@ -6152,6 +6203,9 @@ name = "sync_wrapper" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +dependencies = [ + "futures-core", +] [[package]] name = "synstructure" @@ -6510,6 +6564,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "tower-http" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" +dependencies = [ + "bitflags 2.13.2", + "bytes", + "futures-util", + "http", + "http-body", + "pin-project-lite", + "tower", + "tower-layer", + "tower-service", + "url", +] + [[package]] name = "tower-layer" version = "0.3.3" @@ -6554,6 +6626,12 @@ dependencies = [ "once_cell", ] +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + [[package]] name = "ttf-parser" version = "0.25.1" @@ -7301,6 +7379,15 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "want" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec4cdd0dd910afe868b7ef477227d8d538b46b3075031afee8a9f2acb0a2ed0b" +dependencies = [ + "try-lock", +] + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -7373,6 +7460,19 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "wasm-streams" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1ec4f6517c9e11ae630e200b2b65d193279042e28edd4a2cda233e46670bbb" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "wasmi" version = "1.1.0" diff --git a/crates/ledger-mcp/Cargo.toml b/crates/ledger-mcp/Cargo.toml index c017f62..bbe0a2c 100644 --- a/crates/ledger-mcp/Cargo.toml +++ b/crates/ledger-mcp/Cargo.toml @@ -34,5 +34,7 @@ toml = "1" workspace = true [dev-dependencies] +tokio = { version = "1", features = ["time", "macros", "rt-multi-thread"] } +rmcp = { version = "3", features = ["client", "transport-streamable-http-client-reqwest"] } jsonschema = { version = "0.58", default-features = false } proptest = "1" diff --git a/crates/ledger-mcp/tests/http_concurrency.rs b/crates/ledger-mcp/tests/http_concurrency.rs new file mode 100644 index 0000000..0e9d29c --- /dev/null +++ b/crates/ledger-mcp/tests/http_concurrency.rs @@ -0,0 +1,275 @@ +//! Concurrency over HTTP: several agents at once against one store. +//! +//! In `--http` mode every connection gets its own server over one shared, +//! locked store (main.rs `serve_http`). Oracle (server.rs INVARIANTS 2 and +//! 5, under contention): clients posting at the same time leave books +//! that verify, a trial balance that balances, every posted entry with +//! its receipt (threshold 0.00), and each draft posted exactly once even +//! when two clients race to post it; a restart replays the same books. + +use rmcp::ServiceExt; +use rmcp::model::{CallToolRequestParams, ClientConfig}; +use rmcp::service::{RoleClient, RunningService}; +use rmcp::transport::StreamableHttpClientTransport; +use serde_json::{Value, json}; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +type Client = RunningService; + +const CLIENTS: usize = 6; +const ENTRIES: usize = 12; + +struct Server(Child); +impl Drop for Server { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn free_port() -> u16 { + std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port() +} + +fn b64(bytes: &[u8]) -> String { + const T: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + let mut out = String::new(); + for c in bytes.chunks(3) { + let n = (u32::from(c[0]) << 16) + | (u32::from(*c.get(1).unwrap_or(&0)) << 8) + | u32::from(*c.get(2).unwrap_or(&0)); + for (i, shift) in [18, 12, 6, 0].iter().enumerate() { + if i <= c.len() { + out.push(T[((n >> shift) & 63) as usize] as char); + } else { + out.push('='); + } + } + } + out +} + +async fn connect(url: &str) -> Client { + let deadline = Instant::now() + Duration::from_secs(30); + loop { + let transport = StreamableHttpClientTransport::from_uri(url.to_string()); + match ClientConfig::default().serve(transport).await { + Ok(c) => return c, + Err(e) if Instant::now() > deadline => panic!("connect: {e}"), + Err(_) => tokio::time::sleep(Duration::from_millis(100)).await, + } + } +} + +/// Call a tool: Ok(structured result) or Err(the tool's error text). +async fn call(c: &Client, tool: &str, args: Value) -> Result { + let args = args.as_object().cloned().unwrap_or_default(); + let r = c + .call_tool(CallToolRequestParams::new(tool.to_string()).with_arguments(args)) + .await + .map_err(|e| format!("protocol: {e}"))?; + let text = r + .content + .iter() + .find_map(|c| c.as_text().map(|t| t.text.clone())) + .unwrap_or_default(); + if r.is_error == Some(true) { + Err(text) + } else { + Ok(r.structured_content + .unwrap_or_else(|| serde_json::from_str(&text).unwrap())) + } +} + +async fn draft(c: &Client, tag: &str, cents: u64, with_receipt: bool) -> u64 { + let e = call( + c, + "books.draft_entry", + json!({"date": "2026-03-01", "description": tag}), + ) + .await + .unwrap(); + let id = e["entry_id"].as_u64().unwrap(); + let amount = format!("{}.{:02}", cents / 100, cents % 100); + for (account, side) in [("6000", "debit"), ("1000", "credit")] { + call( + c, + "books.add_line", + json!({"entry_id": id, "account": account, "side": side, "amount": amount}), + ) + .await + .unwrap(); + } + if with_receipt { + call( + c, + "books.attach_receipt", + json!({"entry_id": id, "file_name": format!("{tag}.pdf"), + "mime_type": "application/pdf", "content_base64": b64(tag.as_bytes())}), + ) + .await + .unwrap(); + } + id +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn agents_posting_at_once_leave_whole_books() { + let dir = std::env::temp_dir().join(format!("ledger-mcp-http-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let toml = dir.join("wanderbooks.toml"); + std::fs::write(&toml, "[books]\nreceipt_threshold = \"0.00\"\n").unwrap(); + let data = dir.join("books-data"); + let port = free_port(); + let addr = format!("127.0.0.1:{port}"); + let bin = env!("CARGO_BIN_EXE_ledger-mcp"); + let server = Server( + Command::new(bin) + .arg("--data-dir") + .arg(&data) + .arg("--books-toml") + .arg(&toml) + .arg("--http") + .arg(&addr) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(), + ); + let url = format!("http://{addr}/mcp"); + let setup = connect(&url).await; + for (code, name, ty) in [ + ("1000", "Cash", "asset"), + ("3000", "Capital", "equity"), + ("6000", "Supplies", "expense"), + ] { + call( + &setup, + "books.add_account", + json!({"code": code, "name": name, "account_type": ty}), + ) + .await + .unwrap(); + } + // Drafts every client will race to post. + let mut contested = Vec::new(); + for k in 0..ENTRIES { + contested.push(draft(&setup, &format!("contested {k}"), 100 + k as u64, true).await); + } + + let mut tasks = Vec::new(); + for n in 0..CLIENTS { + let url = url.clone(); + let contested = contested.clone(); + tasks.push(tokio::spawn(async move { + let c = connect(&url).await; + let (mut posted, mut refused, mut won) = (0usize, 0usize, 0usize); + for (k, &race) in contested.iter().enumerate() { + // A receipted entry posts; a bare one is refused. + let receipted = k % 3 != 0; + let id = draft( + &c, + &format!("c{n} e{k}"), + (n * 1000 + k + 1) as u64, + receipted, + ) + .await; + match call( + &c, + "books.post_entry", + json!({"entry_id": id, "confirmed": true}), + ) + .await + { + Ok(_) => posted += 1, + Err(_) => refused += 1, + } + assert_eq!(posted + refused, k + 1); + // Race everyone for the contested draft. + if call( + &c, + "books.post_entry", + json!({"entry_id": race, "confirmed": true}), + ) + .await + .is_ok() + { + won += 1; + } + } + let _ = c.cancel().await; + (posted, refused, won) + })); + } + let (mut posted, mut refused, mut won) = (0, 0, 0); + for t in tasks { + let (p, r, w) = t.await.unwrap(); + posted += p; + refused += r; + won += w; + } + let per_client_bare = (0..ENTRIES).filter(|k| k % 3 == 0).count(); + assert_eq!( + refused, + CLIENTS * per_client_bare, + "every bare entry refused" + ); + assert_eq!(posted, CLIENTS * (ENTRIES - per_client_bare)); + assert_eq!(won, ENTRIES, "each contested draft posted exactly once"); + + let v = call(&setup, "books.verify", json!({})).await.unwrap(); + assert_eq!(v["status"], "ok", "{v}"); + let tb = call(&setup, "books.trial_balance", json!({})) + .await + .unwrap(); + assert_eq!(tb["balanced"], true, "{tb}"); + // Every posted entry has its receipt (threshold 0.00). + for id in 1..=(ENTRIES + CLIENTS * ENTRIES) as u64 { + let e = call(&setup, "books.get_entry", json!({"entry_id": id})) + .await + .unwrap(); + if e["status"] == "posted" { + assert!( + e["receipts"].as_array().is_some_and(|r| !r.is_empty()), + "entry {id} posted without a receipt: {e}" + ); + } + } + let head = v["head"].clone(); + let _ = setup.cancel().await; + drop(server); + + // A restart replays the same books. + let port = free_port(); + let addr = format!("127.0.0.1:{port}"); + let _server = Server( + Command::new(bin) + .arg("--data-dir") + .arg(&data) + .arg("--books-toml") + .arg(&toml) + .arg("--http") + .arg(&addr) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(), + ); + let c = connect(&format!("http://{addr}/mcp")).await; + let v2 = call(&c, "books.verify", json!({})).await.unwrap(); + assert_eq!(v2["status"], "ok"); + assert_eq!( + v2["head"], head, + "the restarted server sees the same journal" + ); + let _ = c.cancel().await; + let _ = std::fs::remove_dir_all(&dir); +} diff --git a/crates/ledger-mcp/tests/spec_mcp.rs b/crates/ledger-mcp/tests/spec_mcp.rs index a9f9a48..c59d508 100644 --- a/crates/ledger-mcp/tests/spec_mcp.rs +++ b/crates/ledger-mcp/tests/spec_mcp.rs @@ -27,7 +27,18 @@ use std::sync::mpsc::{Receiver, RecvTimeoutError, channel}; use std::time::Duration; /// How long any single response may take before the server counts as hung. -const TIMEOUT: Duration = Duration::from_secs(30); +/// Long enough for a server's first answer while a mutation run's jobs +/// contend for the disk (every journal append is synced). +const TIMEOUT: Duration = Duration::from_secs(180); + +/// Random cases for the state machine: 64 on every check; `MCP_CASES` +/// sets fewer for a mutation run, more for a deeper one. +fn cases() -> u32 { + std::env::var("MCP_CASES") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(64) +} /// Keys whose values are money. Derived from the `outputSchema`s: every /// string property described as an amount or "decimal string", plus the @@ -1279,7 +1290,7 @@ fn run_op(m: &mut Mcp, k: &mut Known, op: &Op) { } proptest! { - #![proptest_config(ProptestConfig { cases: 64, failure_persistence: None, ..ProptestConfig::default() })] + #![proptest_config(ProptestConfig { cases: cases(), failure_persistence: None, ..ProptestConfig::default() })] /// Invariants 2 and 7 under random call sequences: every call gets /// exactly one response with its id, result xor error; any refusal diff --git a/justfile b/justfile index bc7136c..fd3548f 100644 --- a/justfile +++ b/justfile @@ -80,9 +80,10 @@ mutants-import: mutants-config: cargo mutants -p ledger-config --test-package ledger-config -j 4 --timeout 120 -o mutants-config -# Mutation testing of the MCP server (every mutant runs the stdio suites). +# Mutation testing of the MCP server (every mutant runs the stdio suites; +# fewer random cases each, as for the store). mutants-mcp: - cargo mutants -p ledger-mcp --test-package ledger-mcp -j 4 --timeout 600 -o mutants-mcp + MCP_CASES=16 cargo mutants -p ledger-mcp --test-package ledger-mcp -j 4 --timeout 600 -o mutants-mcp # Mutation testing of the viewer's logic (formatting, file names, history, # search, ticks, the view model, the attention list; not the painting).