From a3605ec0d483dd88f3888df645249e94892e258a Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 23 Feb 2026 17:19:09 -0500 Subject: [PATCH] Speed up slow tests: drop SSE broadcast on completion, switch to rustls-tls, replace hardcoded sleeps with poll loops, and prevent real API calls in ullm tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Drop event_tx from ManagedPipeline when pipeline completes/cancels/fails so SSE streams end promptly instead of blocking until timeout (3.5s → 0.02s) - Switch reqwest from native-tls to rustls-tls to avoid 500ms macOS cert store load per process (0.67s → 0.005s per OpenAI adapter test) - Replace hardcoded sleep(500ms)/sleep(200ms)/sleep(100ms) in server and integration tests with 10ms poll loops (0.2-0.5s → 0.02-0.03s each) - Add env_clear() to ullm prompt tests to prevent .env from triggering real Anthropic API calls (0.45s → 0.15s) Co-Authored-By: Claude Opus 4.6 --- Cargo.lock | 326 ++-- Cargo.toml | 2 +- crates/attractor/src/server.rs | 48 +- crates/attractor/tests/integration.rs | 2412 ++++++++++++++++++++++++- crates/llm/src/bin/ullm.rs | 4 + 5 files changed, 2589 insertions(+), 203 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1cb470894..5659f1eff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -169,7 +169,7 @@ dependencies = [ "llm", "nom", "predicates", - "rand", + "rand 0.8.5", "serde", "serde_json", "tempfile", @@ -391,6 +391,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" + [[package]] name = "chrono" version = "0.4.43" @@ -470,16 +476,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "core-foundation" -version = "0.9.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91e195e091a93c46f7102ec7818a2aa394e1e1771c3ab4825963fa03e45afb8f" -dependencies = [ - "core-foundation-sys", - "libc", -] - [[package]] name = "core-foundation" version = "0.10.1" @@ -556,15 +552,6 @@ dependencies = [ "serde", ] -[[package]] -name = "encoding_rs" -version = "0.8.35" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "75030f3c4f45dafd7586dd6780965a8c7e8e285a5ecb86713e63a79c5b2766f3" -dependencies = [ - "cfg-if", -] - [[package]] name = "equivalent" version = "1.0.2" @@ -653,21 +640,6 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" -[[package]] -name = "foreign-types" -version = "0.3.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f6f339eb8adc052cd2ca78910fda869aefa38d22d5cb648e6485e4d3fc06f3b1" -dependencies = [ - "foreign-types-shared", -] - -[[package]] -name = "foreign-types-shared" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "00b0228411908ca8685dba7fc2cdd70ec9990a6e753e89b6ac91a84c40fbaf4b" - [[package]] name = "form_urlencoded" version = "1.2.2" @@ -788,8 +760,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" dependencies = [ "cfg-if", + "js-sys", "libc", "wasi", + "wasm-bindgen", ] [[package]] @@ -979,22 +953,7 @@ dependencies = [ "tokio", "tokio-rustls", "tower-service", -] - -[[package]] -name = "hyper-tls" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "70206fc6890eaca9fde8a0bf71caa2ddfc9fe045ac9e5c70df101a7dbde866e0" -dependencies = [ - "bytes", - "http-body-util", - "hyper", - "hyper-util", - "native-tls", - "tokio", - "tokio-native-tls", - "tower-service", + "webpki-roots", ] [[package]] @@ -1015,11 +974,9 @@ dependencies = [ "percent-encoding", "pin-project-lite", "socket2", - "system-configuration", "tokio", "tower-service", "tracing", - "windows-registry", ] [[package]] @@ -1346,7 +1303,7 @@ dependencies = [ "futures", "http", "predicates", - "rand", + "rand 0.8.5", "reqwest 0.12.28", "serde", "serde_json", @@ -1372,6 +1329,12 @@ version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +[[package]] +name = "lru-slab" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" + [[package]] name = "matchit" version = "0.8.4" @@ -1407,23 +1370,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "native-tls" -version = "0.2.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "465500e14ea162429d264d44189adc38b199b62b1c21eea9f69e4b73cb03bbf2" -dependencies = [ - "libc", - "log", - "openssl", - "openssl-probe", - "openssl-sys", - "schannel", - "security-framework", - "security-framework-sys", - "tempfile", -] - [[package]] name = "nom" version = "7.1.3" @@ -1537,50 +1483,12 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" -[[package]] -name = "openssl" -version = "0.10.75" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08838db121398ad17ab8531ce9de97b244589089e290a384c900cb9ff7434328" -dependencies = [ - "bitflags", - "cfg-if", - "foreign-types", - "libc", - "once_cell", - "openssl-macros", - "openssl-sys", -] - -[[package]] -name = "openssl-macros" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - [[package]] name = "openssl-probe" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" -[[package]] -name = "openssl-sys" -version = "0.9.111" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "82cab2d520aa75e3c58898289429321eb788c3106963d0dc886ec7a5f4adc321" -dependencies = [ - "cc", - "libc", - "pkg-config", - "vcpkg", -] - [[package]] name = "outref" version = "0.5.2" @@ -1628,12 +1536,6 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" -[[package]] -name = "pkg-config" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" - [[package]] name = "potential_utf" version = "0.1.4" @@ -1707,6 +1609,61 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "quinn" +version = "0.11.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9e20a958963c291dc322d98411f541009df2ced7b5a4f2bd52337638cfccf20" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls", + "socket2", + "thiserror 2.0.18", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1906b49b0c3bc04b5fe5d86a77925ae6524a19b816ae38ce1e426255f1d8a31" +dependencies = [ + "bytes", + "getrandom 0.3.4", + "lru-slab", + "rand 0.9.2", + "ring", + "rustc-hash", + "rustls", + "rustls-pki-types", + "slab", + "thiserror 2.0.18", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "addec6a0dcad8a8d96a771f815f0eaf55f9d1805756410b39f5fa81332574cbd" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2", + "tracing", + "windows-sys 0.52.0", +] + [[package]] name = "quote" version = "1.0.44" @@ -1729,8 +1686,18 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404" dependencies = [ "libc", - "rand_chacha", - "rand_core", + "rand_chacha 0.3.1", + "rand_core 0.6.4", +] + +[[package]] +name = "rand" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6db2770f06117d490610c7488547d543617b21bfa07796d7a12f6f1bd53850d1" +dependencies = [ + "rand_chacha 0.9.0", + "rand_core 0.9.5", ] [[package]] @@ -1740,7 +1707,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" dependencies = [ "ppv-lite86", - "rand_core", + "rand_core 0.6.4", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", ] [[package]] @@ -1752,6 +1729,15 @@ dependencies = [ "getrandom 0.2.17", ] +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -1842,30 +1828,27 @@ checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" dependencies = [ "base64", "bytes", - "encoding_rs", "futures-core", "futures-util", - "h2", "http", "http-body", "http-body-util", "hyper", "hyper-rustls", - "hyper-tls", "hyper-util", "js-sys", "log", - "mime", - "native-tls", "percent-encoding", "pin-project-lite", + "quinn", + "rustls", "rustls-pki-types", "serde", "serde_json", "serde_urlencoded", "sync_wrapper", "tokio", - "tokio-native-tls", + "tokio-rustls", "tokio-util", "tower", "tower-http", @@ -1875,6 +1858,7 @@ dependencies = [ "wasm-bindgen-futures", "wasm-streams", "web-sys", + "webpki-roots", ] [[package]] @@ -1930,6 +1914,12 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "rustc-hash" +version = "2.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "357703d41365b4b27c590e3ed91eabb1b663f07c4c084095e60cbed4362dff0d" + [[package]] name = "rustix" version = "1.1.3" @@ -1951,6 +1941,7 @@ checksum = "c665f33d38cea657d9614f766881e4d510e0eda4239891eea56b4cadcf01801b" dependencies = [ "aws-lc-rs", "once_cell", + "ring", "rustls-pki-types", "rustls-webpki", "subtle", @@ -1975,6 +1966,7 @@ version = "1.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be040f8b0a225e40375822a563fa9524378b9d63112f53e19ffff34df5d33fdd" dependencies = [ + "web-time", "zeroize", ] @@ -1984,7 +1976,7 @@ version = "0.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d99feebc72bae7ab76ba994bb5e121b8d83d910ca40b36e0921f53becc41784" dependencies = [ - "core-foundation 0.10.1", + "core-foundation", "core-foundation-sys", "jni", "log", @@ -2084,7 +2076,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d17b898a6d6948c3a8ee4372c17cb384f90d2e6e912ef00895b14fd7ab54ec38" dependencies = [ "bitflags", - "core-foundation 0.10.1", + "core-foundation", "core-foundation-sys", "libc", "security-framework-sys", @@ -2288,27 +2280,6 @@ dependencies = [ "syn", ] -[[package]] -name = "system-configuration" -version = "0.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a13f3d0daba03132c0aa9767f98351b3488edc2c100cda2d2ec2b04f3d8d3c8b" -dependencies = [ - "bitflags", - "core-foundation 0.9.4", - "system-configuration-sys", -] - -[[package]] -name = "system-configuration-sys" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e1d1b10ced5ca923a1fcb8d03e96b8d3268065d724548c0211415ff6ac6bac4" -dependencies = [ - "core-foundation-sys", - "libc", -] - [[package]] name = "tar" version = "0.4.44" @@ -2424,6 +2395,21 @@ dependencies = [ "zerovec", ] +[[package]] +name = "tinyvec" +version = "1.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfa5fdc3bce6191a1dbc8c02d5c8bffcf557bafa17c124c5264a458f1b0613fa" +dependencies = [ + "tinyvec_macros", +] + +[[package]] +name = "tinyvec_macros" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" + [[package]] name = "tokio" version = "1.49.0" @@ -2452,16 +2438,6 @@ dependencies = [ "syn", ] -[[package]] -name = "tokio-native-tls" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bbae76ab933c85776efabc971569dd6119c580d8f5d448769dec1764bf796ef2" -dependencies = [ - "native-tls", - "tokio", -] - [[package]] name = "tokio-rustls" version = "0.26.4" @@ -2638,12 +2614,6 @@ dependencies = [ "vsimd", ] -[[package]] -name = "vcpkg" -version = "0.2.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" - [[package]] name = "version_check" version = "0.9.5" @@ -2824,6 +2794,16 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "webpki-root-certs" version = "1.0.6" @@ -2833,6 +2813,15 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "webpki-roots" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22cfaf3c063993ff62e73cb4311efde4db1efb31ab78a3e5c457939ad5cc0bed" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "winapi" version = "0.3.9" @@ -2905,17 +2894,6 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" -[[package]] -name = "windows-registry" -version = "0.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "02752bf7fbdcce7f2a27a742f798510f3e5ad88dbe84871e5168e2120c3d5720" -dependencies = [ - "windows-link", - "windows-result", - "windows-strings", -] - [[package]] name = "windows-result" version = "0.4.1" diff --git a/Cargo.toml b/Cargo.toml index e3c7d5296..11f1a9b24 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,7 +20,7 @@ thiserror = "2" serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { version = "1", features = ["full"] } -reqwest = { version = "0.12", features = ["json", "stream"] } +reqwest = { version = "0.12", default-features = false, features = ["json", "stream", "rustls-tls"] } uuid = { version = "1", features = ["v4"] } rand = "0.8" dotenvy = "0.15" diff --git a/crates/attractor/src/server.rs b/crates/attractor/src/server.rs index d46b0d80f..ef01abb6e 100644 --- a/crates/attractor/src/server.rs +++ b/crates/attractor/src/server.rs @@ -54,7 +54,7 @@ struct ManagedPipeline { status: PipelineStatus, error: Option, interviewer: Arc, - event_tx: broadcast::Sender, + event_tx: Option>, context: Option, checkpoint: Option, cancel_tx: Option>, @@ -189,7 +189,7 @@ async fn start_pipeline( status: PipelineStatus::Running, error: None, interviewer: Arc::clone(&interviewer), - event_tx: event_tx.clone(), + event_tx: Some(event_tx.clone()), context: Some(context.clone()), checkpoint: None, cancel_tx: Some(cancel_tx), @@ -212,6 +212,7 @@ async fn start_pipeline( let mut pipelines = state_clone.pipelines.lock().expect("pipelines lock poisoned"); if let Some(pipeline) = pipelines.get_mut(&id_clone) { pipeline.status = PipelineStatus::Cancelled; + pipeline.event_tx = None; } return; } @@ -235,6 +236,7 @@ async fn start_pipeline( } } pipeline.checkpoint = checkpoint; + pipeline.event_tx = None; } }); @@ -346,7 +348,10 @@ async fn get_events( let rx = { let pipelines = state.pipelines.lock().expect("pipelines lock poisoned"); match pipelines.get(&id) { - Some(pipeline) => pipeline.event_tx.subscribe(), + Some(pipeline) => match &pipeline.event_tx { + Some(tx) => tx.subscribe(), + None => return StatusCode::GONE.into_response(), + }, None => return StatusCode::NOT_FOUND.into_response(), } }; @@ -566,8 +571,8 @@ mod tests { let body = body_json(response.into_body()).await; let pipeline_id = body["id"].as_str().unwrap().to_string(); - // Give pipeline a moment to run - tokio::time::sleep(std::time::Duration::from_millis(200)).await; + // Give pipeline a moment to start + tokio::time::sleep(std::time::Duration::from_millis(10)).await; // Check status let req = Request::builder() @@ -840,21 +845,24 @@ mod tests { let body = body_json(response.into_body()).await; let pipeline_id = body["id"].as_str().unwrap().to_string(); - // Wait for pipeline to complete - tokio::time::sleep(std::time::Duration::from_millis(500)).await; - - // Check status - let req = Request::builder() - .method("GET") - .uri(format!("/pipelines/{pipeline_id}")) - .body(Body::empty()) - .unwrap(); - - let response = app.oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::OK); - - let body = body_json(response.into_body()).await; - assert_eq!(body["status"].as_str().unwrap(), "completed"); + // Poll until pipeline completes + let mut status = String::new(); + for _ in 0..100 { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + let req = Request::builder() + .method("GET") + .uri(format!("/pipelines/{pipeline_id}")) + .body(Body::empty()) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = body_json(response.into_body()).await; + status = body["status"].as_str().unwrap().to_string(); + if status == "completed" || status == "failed" { + break; + } + } + assert_eq!(status, "completed"); } #[tokio::test] diff --git a/crates/attractor/tests/integration.rs b/crates/attractor/tests/integration.rs index 6c288ee8f..e7b5ceb23 100644 --- a/crates/attractor/tests/integration.rs +++ b/crates/attractor/tests/integration.rs @@ -2943,8 +2943,8 @@ mod server_lifecycle { // 2. Poll for question to appear (pipeline runs start -> work -> gate, then blocks) let mut question_id = String::new(); - for _ in 0..50 { - tokio::time::sleep(Duration::from_millis(100)).await; + for _ in 0..500 { + tokio::time::sleep(Duration::from_millis(10)).await; let req = Request::builder() .method("GET") .uri(format!("/pipelines/{pipeline_id}/questions")) @@ -2978,8 +2978,8 @@ mod server_lifecycle { // 4. Poll until completed let mut final_status = String::new(); - for _ in 0..50 { - tokio::time::sleep(Duration::from_millis(100)).await; + for _ in 0..500 { + tokio::time::sleep(Duration::from_millis(10)).await; let req = Request::builder() .method("GET") .uri(format!("/pipelines/{pipeline_id}")) @@ -3039,7 +3039,7 @@ mod server_lifecycle { let pipeline_id = body["id"].as_str().unwrap().to_string(); // Wait briefly for pipeline to start running - tokio::time::sleep(Duration::from_millis(200)).await; + tokio::time::sleep(Duration::from_millis(10)).await; // Cancel it let req = Request::builder() @@ -3143,7 +3143,7 @@ mod sse_events { let mut body = response.into_body(); let mut sse_data = String::new(); loop { - match tokio::time::timeout(Duration::from_secs(3), body.frame()).await { + match tokio::time::timeout(Duration::from_millis(500), body.frame()).await { Ok(Some(Ok(frame))) => { if let Some(data) = frame.data_ref() { sse_data.push_str(&String::from_utf8_lossy(data)); @@ -3185,8 +3185,9 @@ mod sse_events { ); } - // Wait for completion, then verify checkpoint - tokio::time::sleep(Duration::from_millis(500)).await; + // Pipeline is complete (SSE stream ended), verify checkpoint + // Small yield to let the spawned task update state + tokio::time::sleep(Duration::from_millis(10)).await; let req = Request::builder() .method("GET") @@ -3648,6 +3649,1425 @@ async fn graph_merge_e2e_through_engine() { ); } +// =========================================================================== +// Context fidelity integration tests (spec Section 5.4) +// =========================================================================== + +/// Shared capture storage for fidelity tests. +#[derive(Clone)] +struct FidelityCaptures { + fidelities: Arc>>, + thread_ids: Arc)>>>, + preambles: Arc>>, +} + +impl FidelityCaptures { + fn new() -> Self { + Self { + fidelities: Arc::new(std::sync::Mutex::new(Vec::new())), + thread_ids: Arc::new(std::sync::Mutex::new(Vec::new())), + preambles: Arc::new(std::sync::Mutex::new(Vec::new())), + } + } +} + +/// A handler that captures the resolved fidelity and thread_id from the context. +struct FidelityCapturingHandler { + captures: FidelityCaptures, +} + +#[async_trait::async_trait] +impl Handler for FidelityCapturingHandler { + async fn execute( + &self, + node: &Node, + context: &Context, + _graph: &Graph, + _logs_root: &Path, + _services: &attractor::handler::EngineServices, + ) -> Result { + let fidelity = context.get_string("internal.fidelity", "none"); + self.captures + .fidelities + .lock() + .unwrap() + .push((node.id.clone(), fidelity)); + + let thread_id = context + .get("internal.thread_id") + .and_then(|v| v.as_str().map(String::from)); + self.captures + .thread_ids + .lock() + .unwrap() + .push((node.id.clone(), thread_id)); + + let preamble = context.get_string("current.preamble", ""); + self.captures + .preambles + .lock() + .unwrap() + .push((node.id.clone(), preamble)); + + Ok(Outcome::success()) + } +} + +#[tokio::test] +async fn fidelity_default_is_compact() { + let mut graph = make_graph_with_start_exit("FidelityDefaultTest"); + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + graph.nodes.insert("work".to_string(), work); + graph.edges.push(Edge::new("start", "work")); + graph.edges.push(Edge::new("work", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let fidelities = captures.fidelities.lock().unwrap(); + assert_eq!(fidelities.len(), 1); + assert_eq!(fidelities[0].0, "work"); + assert_eq!(fidelities[0].1, "compact"); + + let preambles = captures.preambles.lock().unwrap(); + assert!(!preambles[0].1.is_empty(), "compact fidelity should produce a preamble"); +} + +#[tokio::test] +async fn fidelity_graph_default_applied() { + let mut graph = make_graph_with_start_exit("FidelityGraphDefaultTest"); + graph.attrs.insert( + "default_fidelity".to_string(), + AttrValue::String("truncate".to_string()), + ); + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + graph.nodes.insert("work".to_string(), work); + graph.edges.push(Edge::new("start", "work")); + graph.edges.push(Edge::new("work", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let fidelities = captures.fidelities.lock().unwrap(); + assert_eq!(fidelities[0].1, "truncate"); +} + +#[tokio::test] +async fn fidelity_node_overrides_graph_default() { + let mut graph = make_graph_with_start_exit("FidelityNodeOverrideTest"); + graph.attrs.insert( + "default_fidelity".to_string(), + AttrValue::String("truncate".to_string()), + ); + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + work.attrs.insert( + "fidelity".to_string(), + AttrValue::String("summary:medium".to_string()), + ); + graph.nodes.insert("work".to_string(), work); + graph.edges.push(Edge::new("start", "work")); + graph.edges.push(Edge::new("work", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let fidelities = captures.fidelities.lock().unwrap(); + assert_eq!(fidelities[0].1, "summary:medium"); +} + +#[tokio::test] +async fn fidelity_edge_overrides_node_and_graph() { + let mut graph = make_graph_with_start_exit("FidelityEdgeOverrideTest"); + graph.attrs.insert( + "default_fidelity".to_string(), + AttrValue::String("truncate".to_string()), + ); + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + work.attrs.insert( + "fidelity".to_string(), + AttrValue::String("compact".to_string()), + ); + graph.nodes.insert("work".to_string(), work); + + let mut edge_with_fidelity = Edge::new("start", "work"); + edge_with_fidelity.attrs.insert( + "fidelity".to_string(), + AttrValue::String("summary:high".to_string()), + ); + graph.edges.push(edge_with_fidelity); + graph.edges.push(Edge::new("work", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let fidelities = captures.fidelities.lock().unwrap(); + assert_eq!(fidelities[0].1, "summary:high"); +} + +#[tokio::test] +async fn fidelity_full_produces_empty_preamble() { + let mut graph = make_graph_with_start_exit("FidelityFullPreambleTest"); + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + work.attrs.insert( + "fidelity".to_string(), + AttrValue::String("full".to_string()), + ); + graph.nodes.insert("work".to_string(), work); + graph.edges.push(Edge::new("start", "work")); + graph.edges.push(Edge::new("work", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let fidelities = captures.fidelities.lock().unwrap(); + assert_eq!(fidelities[0].1, "full"); + + let preambles = captures.preambles.lock().unwrap(); + assert_eq!(preambles[0].1, "", "full fidelity should produce empty preamble"); +} + +#[tokio::test] +async fn fidelity_truncate_preamble_minimal() { + let mut graph = make_graph_with_start_exit("FidelityTruncateTest"); + graph.attrs.insert( + "goal".to_string(), + AttrValue::String("Test truncate mode".to_string()), + ); + graph.attrs.insert( + "default_fidelity".to_string(), + AttrValue::String("truncate".to_string()), + ); + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + graph.nodes.insert("work".to_string(), work); + graph.edges.push(Edge::new("start", "work")); + graph.edges.push(Edge::new("work", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let preambles = captures.preambles.lock().unwrap(); + let preamble = &preambles[0].1; + assert!( + preamble.contains("Goal: Test truncate mode"), + "truncate preamble should contain the goal" + ); + assert!( + preamble.contains("Run ID:"), + "truncate preamble should contain run ID" + ); + assert!( + !preamble.contains("Completed stages:"), + "truncate should not include stage details" + ); +} + +#[tokio::test] +async fn fidelity_summary_low_mode() { + let mut graph = make_graph_with_start_exit("SummaryLow"); + graph.attrs.insert( + "goal".to_string(), + AttrValue::String("Test summary".to_string()), + ); + graph.attrs.insert( + "default_fidelity".to_string(), + AttrValue::String("summary:low".to_string()), + ); + let mut step_a = Node::new("step_a"); + step_a.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + graph.nodes.insert("step_a".to_string(), step_a); + let mut step_b = Node::new("step_b"); + step_b.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + graph.nodes.insert("step_b".to_string(), step_b); + graph.edges.push(Edge::new("start", "step_a")); + graph.edges.push(Edge::new("step_a", "step_b")); + graph.edges.push(Edge::new("step_b", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let fidelities = captures.fidelities.lock().unwrap(); + assert_eq!(fidelities[0].1, "summary:low"); + assert_eq!(fidelities[1].1, "summary:low"); + + let preambles = captures.preambles.lock().unwrap(); + assert!( + preambles[1].1.contains("Test summary"), + "summary:low preamble should contain goal" + ); +} + +#[tokio::test] +async fn fidelity_summary_medium_mode() { + let mut graph = make_graph_with_start_exit("SummaryMedium"); + graph.attrs.insert( + "goal".to_string(), + AttrValue::String("Test summary".to_string()), + ); + graph.attrs.insert( + "default_fidelity".to_string(), + AttrValue::String("summary:medium".to_string()), + ); + let mut step_a = Node::new("step_a"); + step_a.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + graph.nodes.insert("step_a".to_string(), step_a); + let mut step_b = Node::new("step_b"); + step_b.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + graph.nodes.insert("step_b".to_string(), step_b); + graph.edges.push(Edge::new("start", "step_a")); + graph.edges.push(Edge::new("step_a", "step_b")); + graph.edges.push(Edge::new("step_b", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let fidelities = captures.fidelities.lock().unwrap(); + assert_eq!(fidelities[0].1, "summary:medium"); + assert_eq!(fidelities[1].1, "summary:medium"); + + let preambles = captures.preambles.lock().unwrap(); + assert!( + preambles[1].1.contains("Test summary"), + "summary:medium preamble should contain goal" + ); +} + +#[tokio::test] +async fn fidelity_summary_high_mode() { + let mut graph = make_graph_with_start_exit("SummaryHigh"); + graph.attrs.insert( + "goal".to_string(), + AttrValue::String("Test summary".to_string()), + ); + graph.attrs.insert( + "default_fidelity".to_string(), + AttrValue::String("summary:high".to_string()), + ); + let mut step_a = Node::new("step_a"); + step_a.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + graph.nodes.insert("step_a".to_string(), step_a); + let mut step_b = Node::new("step_b"); + step_b.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + graph.nodes.insert("step_b".to_string(), step_b); + graph.edges.push(Edge::new("start", "step_a")); + graph.edges.push(Edge::new("step_a", "step_b")); + graph.edges.push(Edge::new("step_b", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let fidelities = captures.fidelities.lock().unwrap(); + assert_eq!(fidelities[0].1, "summary:high"); + assert_eq!(fidelities[1].1, "summary:high"); + + let preambles = captures.preambles.lock().unwrap(); + assert!( + preambles[1].1.contains("Test summary"), + "summary:high preamble should contain goal" + ); +} + +#[tokio::test] +async fn fidelity_full_sets_thread_id_in_context() { + let mut graph = make_graph_with_start_exit("FidelityThreadTest"); + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + work.attrs.insert( + "fidelity".to_string(), + AttrValue::String("full".to_string()), + ); + work.attrs.insert( + "thread_id".to_string(), + AttrValue::String("my-session".to_string()), + ); + graph.nodes.insert("work".to_string(), work); + graph.edges.push(Edge::new("start", "work")); + graph.edges.push(Edge::new("work", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let thread_ids = captures.thread_ids.lock().unwrap(); + assert_eq!(thread_ids[0].0, "work"); + assert_eq!(thread_ids[0].1, Some("my-session".to_string())); +} + +#[tokio::test] +async fn fidelity_full_nodes_share_thread_id() { + let mut graph = make_graph_with_start_exit("FidelitySharedThreadTest"); + let mut step_a = Node::new("step_a"); + step_a.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + step_a.attrs.insert( + "fidelity".to_string(), + AttrValue::String("full".to_string()), + ); + step_a.attrs.insert( + "thread_id".to_string(), + AttrValue::String("shared-session".to_string()), + ); + graph.nodes.insert("step_a".to_string(), step_a); + + let mut step_b = Node::new("step_b"); + step_b.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + step_b.attrs.insert( + "fidelity".to_string(), + AttrValue::String("full".to_string()), + ); + step_b.attrs.insert( + "thread_id".to_string(), + AttrValue::String("shared-session".to_string()), + ); + graph.nodes.insert("step_b".to_string(), step_b); + + graph.edges.push(Edge::new("start", "step_a")); + graph.edges.push(Edge::new("step_a", "step_b")); + graph.edges.push(Edge::new("step_b", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let thread_ids = captures.thread_ids.lock().unwrap(); + assert_eq!(thread_ids[0].0, "step_a"); + assert_eq!(thread_ids[0].1, Some("shared-session".to_string())); + assert_eq!(thread_ids[1].0, "step_b"); + assert_eq!(thread_ids[1].1, Some("shared-session".to_string())); +} + +#[tokio::test] +async fn fidelity_resume_degrades_full_to_summary_high() { + let mut graph = make_graph_with_start_exit("FidelityResumeTest"); + let mut step_a = Node::new("step_a"); + step_a.attrs.insert( + "fidelity".to_string(), + AttrValue::String("full".to_string()), + ); + graph.nodes.insert("step_a".to_string(), step_a); + + let mut step_b = Node::new("step_b"); + step_b.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + step_b.attrs.insert( + "fidelity".to_string(), + AttrValue::String("full".to_string()), + ); + graph.nodes.insert("step_b".to_string(), step_b); + + graph.edges.push(Edge::new("start", "step_a")); + graph.edges.push(Edge::new("step_a", "step_b")); + graph.edges.push(Edge::new("step_b", "exit")); + + let ctx = Context::new(); + ctx.set("outcome", serde_json::json!("success")); + ctx.set("internal.fidelity", serde_json::json!("full")); + + let mut outcomes = std::collections::HashMap::new(); + outcomes.insert("start".to_string(), Outcome::success()); + outcomes.insert("step_a".to_string(), Outcome::success()); + + let checkpoint = Checkpoint::from_context( + &ctx, + "step_a", + vec!["start".to_string(), "step_a".to_string()], + std::collections::HashMap::new(), + outcomes, + Some("step_b".to_string()), + ); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine + .run_from_checkpoint(&graph, &config, &checkpoint) + .await + .expect("resume should succeed"); + + let fidelities = captures.fidelities.lock().unwrap(); + assert_eq!(fidelities[0].0, "step_b"); + assert_eq!( + fidelities[0].1, "summary:high", + "first node after resume from full fidelity should be degraded to summary:high" + ); +} + +#[tokio::test] +async fn fidelity_resume_degrade_only_affects_first_hop() { + let mut graph = make_graph_with_start_exit("FidelityResumeSingleHopTest"); + let mut step_a = Node::new("step_a"); + step_a.attrs.insert( + "fidelity".to_string(), + AttrValue::String("full".to_string()), + ); + graph.nodes.insert("step_a".to_string(), step_a); + + let mut step_b = Node::new("step_b"); + step_b.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + step_b.attrs.insert( + "fidelity".to_string(), + AttrValue::String("full".to_string()), + ); + graph.nodes.insert("step_b".to_string(), step_b); + + let mut step_c = Node::new("step_c"); + step_c.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + step_c.attrs.insert( + "fidelity".to_string(), + AttrValue::String("full".to_string()), + ); + graph.nodes.insert("step_c".to_string(), step_c); + + graph.edges.push(Edge::new("start", "step_a")); + graph.edges.push(Edge::new("step_a", "step_b")); + graph.edges.push(Edge::new("step_b", "step_c")); + graph.edges.push(Edge::new("step_c", "exit")); + + let ctx = Context::new(); + ctx.set("outcome", serde_json::json!("success")); + ctx.set("internal.fidelity", serde_json::json!("full")); + + let mut outcomes = std::collections::HashMap::new(); + outcomes.insert("start".to_string(), Outcome::success()); + outcomes.insert("step_a".to_string(), Outcome::success()); + + let checkpoint = Checkpoint::from_context( + &ctx, + "step_a", + vec!["start".to_string(), "step_a".to_string()], + std::collections::HashMap::new(), + outcomes, + Some("step_b".to_string()), + ); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine + .run_from_checkpoint(&graph, &config, &checkpoint) + .await + .expect("resume should succeed"); + + let fidelities = captures.fidelities.lock().unwrap(); + assert_eq!(fidelities[0].0, "step_b"); + assert_eq!(fidelities[0].1, "summary:high"); + assert_eq!(fidelities[1].0, "step_c"); + assert_eq!(fidelities[1].1, "full"); +} + +#[tokio::test] +async fn fidelity_resume_no_degrade_when_not_full() { + let mut graph = make_graph_with_start_exit("FidelityResumeNoDegrade"); + let mut step_a = Node::new("step_a"); + step_a.attrs.insert( + "fidelity".to_string(), + AttrValue::String("compact".to_string()), + ); + graph.nodes.insert("step_a".to_string(), step_a); + + let mut step_b = Node::new("step_b"); + step_b.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + step_b.attrs.insert( + "fidelity".to_string(), + AttrValue::String("full".to_string()), + ); + graph.nodes.insert("step_b".to_string(), step_b); + + graph.edges.push(Edge::new("start", "step_a")); + graph.edges.push(Edge::new("step_a", "step_b")); + graph.edges.push(Edge::new("step_b", "exit")); + + let ctx = Context::new(); + ctx.set("outcome", serde_json::json!("success")); + ctx.set("internal.fidelity", serde_json::json!("compact")); + + let mut outcomes = std::collections::HashMap::new(); + outcomes.insert("start".to_string(), Outcome::success()); + outcomes.insert("step_a".to_string(), Outcome::success()); + + let checkpoint = Checkpoint::from_context( + &ctx, + "step_a", + vec!["start".to_string(), "step_a".to_string()], + std::collections::HashMap::new(), + outcomes, + Some("step_b".to_string()), + ); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine + .run_from_checkpoint(&graph, &config, &checkpoint) + .await + .expect("resume should succeed"); + + let fidelities = captures.fidelities.lock().unwrap(); + assert_eq!(fidelities[0].0, "step_b"); + assert_eq!(fidelities[0].1, "full"); +} + +#[tokio::test] +async fn fidelity_stored_in_checkpoint_context() { + let mut graph = make_graph_with_start_exit("FidelityCheckpointTest"); + graph.attrs.insert( + "default_fidelity".to_string(), + AttrValue::String("summary:low".to_string()), + ); + let work = Node::new("work"); + graph.nodes.insert("work".to_string(), work); + graph.edges.push(Edge::new("start", "work")); + graph.edges.push(Edge::new("work", "exit")); + + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let cp = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap(); + assert_eq!( + cp.context_values.get("internal.fidelity"), + Some(&serde_json::json!("summary:low")), + "checkpoint should record the resolved fidelity" + ); +} + +#[tokio::test] +async fn fidelity_precedence_multi_node_pipeline() { + let mut graph = make_graph_with_start_exit("FidelityPrecedenceTest"); + graph.attrs.insert( + "default_fidelity".to_string(), + AttrValue::String("truncate".to_string()), + ); + + let mut step_a = Node::new("step_a"); + step_a.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + graph.nodes.insert("step_a".to_string(), step_a); + + let mut step_b = Node::new("step_b"); + step_b.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + step_b.attrs.insert( + "fidelity".to_string(), + AttrValue::String("summary:medium".to_string()), + ); + graph.nodes.insert("step_b".to_string(), step_b); + + let mut step_c = Node::new("step_c"); + step_c.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + step_c.attrs.insert( + "fidelity".to_string(), + AttrValue::String("compact".to_string()), + ); + graph.nodes.insert("step_c".to_string(), step_c); + + graph.edges.push(Edge::new("start", "step_a")); + graph.edges.push(Edge::new("step_a", "step_b")); + + let mut edge_b_c = Edge::new("step_b", "step_c"); + edge_b_c.attrs.insert( + "fidelity".to_string(), + AttrValue::String("summary:high".to_string()), + ); + graph.edges.push(edge_b_c); + + graph.edges.push(Edge::new("step_c", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let fidelities = captures.fidelities.lock().unwrap(); + assert_eq!(fidelities[0].0, "step_a"); + assert_eq!(fidelities[0].1, "truncate"); + assert_eq!(fidelities[1].0, "step_b"); + assert_eq!(fidelities[1].1, "summary:medium"); + assert_eq!(fidelities[2].0, "step_c"); + assert_eq!(fidelities[2].1, "summary:high"); +} + +#[tokio::test] +async fn fidelity_compact_preamble_includes_completed_stages_and_context() { + let mut graph = make_graph_with_start_exit("FidelityCompactContentTest"); + graph.attrs.insert( + "goal".to_string(), + AttrValue::String("Build the widget".to_string()), + ); + graph.attrs.insert( + "default_fidelity".to_string(), + AttrValue::String("compact".to_string()), + ); + + let mut step_a = Node::new("step_a"); + step_a.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + graph.nodes.insert("step_a".to_string(), step_a); + + let mut step_b = Node::new("step_b"); + step_b.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + graph.nodes.insert("step_b".to_string(), step_b); + + graph.edges.push(Edge::new("start", "step_a")); + graph.edges.push(Edge::new("step_a", "step_b")); + graph.edges.push(Edge::new("step_b", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let preambles = captures.preambles.lock().unwrap(); + // step_b's preamble should contain structured summary of completed work + let step_b_preamble = &preambles[1].1; + assert!( + step_b_preamble.contains("Build the widget"), + "compact preamble should contain the goal" + ); + assert!( + step_b_preamble.contains("Completed stages:"), + "compact preamble should include completed stages section" + ); + assert!( + step_b_preamble.contains("step_a"), + "compact preamble should mention completed node step_a" + ); +} + +#[tokio::test] +async fn fidelity_summary_detail_increases_with_level() { + // Run three separate pipelines with summary:low, summary:medium, and summary:high. + // Verify that higher detail levels produce longer preambles. + let mut preamble_lengths: Vec<(String, usize)> = Vec::new(); + + // -- summary:low -- + let mut graph_low = make_graph_with_start_exit("SummaryLenLow"); + graph_low.attrs.insert("goal".to_string(), AttrValue::String("Detail test".to_string())); + graph_low.attrs.insert("default_fidelity".to_string(), AttrValue::String("summary:low".to_string())); + let mut step_a_low = Node::new("step_a"); + step_a_low.attrs.insert("type".to_string(), AttrValue::String("fidelity_capture".to_string())); + graph_low.nodes.insert("step_a".to_string(), step_a_low); + let mut step_b_low = Node::new("step_b"); + step_b_low.attrs.insert("type".to_string(), AttrValue::String("fidelity_capture".to_string())); + graph_low.nodes.insert("step_b".to_string(), step_b_low); + graph_low.edges.push(Edge::new("start", "step_a")); + graph_low.edges.push(Edge::new("step_a", "step_b")); + graph_low.edges.push(Edge::new("step_b", "exit")); + + let captures_low = FidelityCaptures::new(); + let dir_low = tempfile::tempdir().unwrap(); + let mut registry_low = HandlerRegistry::new(Box::new(StartHandler)); + registry_low.register("start", Box::new(StartHandler)); + registry_low.register("exit", Box::new(ExitHandler)); + registry_low.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures_low.clone() })); + let engine_low = PipelineEngine::new(registry_low, EventEmitter::new()); + let config_low = RunConfig { logs_root: dir_low.path().to_path_buf(), cancel_token: None }; + engine_low.run(&graph_low, &config_low).await.expect("run low"); + let preambles_low = captures_low.preambles.lock().unwrap(); + preamble_lengths.push(("summary:low".to_string(), preambles_low[1].1.len())); + + // -- summary:medium -- + let mut graph_med = make_graph_with_start_exit("SummaryLenMed"); + graph_med.attrs.insert("goal".to_string(), AttrValue::String("Detail test".to_string())); + graph_med.attrs.insert("default_fidelity".to_string(), AttrValue::String("summary:medium".to_string())); + let mut step_a_med = Node::new("step_a"); + step_a_med.attrs.insert("type".to_string(), AttrValue::String("fidelity_capture".to_string())); + graph_med.nodes.insert("step_a".to_string(), step_a_med); + let mut step_b_med = Node::new("step_b"); + step_b_med.attrs.insert("type".to_string(), AttrValue::String("fidelity_capture".to_string())); + graph_med.nodes.insert("step_b".to_string(), step_b_med); + graph_med.edges.push(Edge::new("start", "step_a")); + graph_med.edges.push(Edge::new("step_a", "step_b")); + graph_med.edges.push(Edge::new("step_b", "exit")); + + let captures_med = FidelityCaptures::new(); + let dir_med = tempfile::tempdir().unwrap(); + let mut registry_med = HandlerRegistry::new(Box::new(StartHandler)); + registry_med.register("start", Box::new(StartHandler)); + registry_med.register("exit", Box::new(ExitHandler)); + registry_med.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures_med.clone() })); + let engine_med = PipelineEngine::new(registry_med, EventEmitter::new()); + let config_med = RunConfig { logs_root: dir_med.path().to_path_buf(), cancel_token: None }; + engine_med.run(&graph_med, &config_med).await.expect("run med"); + let preambles_med = captures_med.preambles.lock().unwrap(); + preamble_lengths.push(("summary:medium".to_string(), preambles_med[1].1.len())); + + // -- summary:high -- + let mut graph_high = make_graph_with_start_exit("SummaryLenHigh"); + graph_high.attrs.insert("goal".to_string(), AttrValue::String("Detail test".to_string())); + graph_high.attrs.insert("default_fidelity".to_string(), AttrValue::String("summary:high".to_string())); + let mut step_a_high = Node::new("step_a"); + step_a_high.attrs.insert("type".to_string(), AttrValue::String("fidelity_capture".to_string())); + graph_high.nodes.insert("step_a".to_string(), step_a_high); + let mut step_b_high = Node::new("step_b"); + step_b_high.attrs.insert("type".to_string(), AttrValue::String("fidelity_capture".to_string())); + graph_high.nodes.insert("step_b".to_string(), step_b_high); + graph_high.edges.push(Edge::new("start", "step_a")); + graph_high.edges.push(Edge::new("step_a", "step_b")); + graph_high.edges.push(Edge::new("step_b", "exit")); + + let captures_high = FidelityCaptures::new(); + let dir_high = tempfile::tempdir().unwrap(); + let mut registry_high = HandlerRegistry::new(Box::new(StartHandler)); + registry_high.register("start", Box::new(StartHandler)); + registry_high.register("exit", Box::new(ExitHandler)); + registry_high.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures_high.clone() })); + let engine_high = PipelineEngine::new(registry_high, EventEmitter::new()); + let config_high = RunConfig { logs_root: dir_high.path().to_path_buf(), cancel_token: None }; + engine_high.run(&graph_high, &config_high).await.expect("run high"); + let preambles_high = captures_high.preambles.lock().unwrap(); + preamble_lengths.push(("summary:high".to_string(), preambles_high[1].1.len())); + + // Higher summary levels should produce more detailed (longer) preambles + assert!( + preamble_lengths[0].1 <= preamble_lengths[1].1, + "summary:low ({}) should be no longer than summary:medium ({})", + preamble_lengths[0].1, + preamble_lengths[1].1, + ); + assert!( + preamble_lengths[1].1 <= preamble_lengths[2].1, + "summary:medium ({}) should be no longer than summary:high ({})", + preamble_lengths[1].1, + preamble_lengths[2].1, + ); +} + +#[tokio::test] +async fn fidelity_thread_id_fallback_to_previous_node_in_pipeline() { + // When no thread_id is set on the node, edge, graph, or class, + // the thread ID should fall back to the previous node's ID. + let mut graph = make_graph_with_start_exit("ThreadFallbackTest"); + let mut step_a = Node::new("step_a"); + step_a.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + graph.nodes.insert("step_a".to_string(), step_a); + + let mut step_b = Node::new("step_b"); + step_b.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + graph.nodes.insert("step_b".to_string(), step_b); + + graph.edges.push(Edge::new("start", "step_a")); + graph.edges.push(Edge::new("step_a", "step_b")); + graph.edges.push(Edge::new("step_b", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let thread_ids = captures.thread_ids.lock().unwrap(); + // step_a should have previous node = start + assert_eq!(thread_ids[0].0, "step_a"); + assert_eq!(thread_ids[0].1, Some("start".to_string())); + // step_b should have previous node = step_a + assert_eq!(thread_ids[1].0, "step_b"); + assert_eq!(thread_ids[1].1, Some("step_a".to_string())); +} + +#[tokio::test] +async fn fidelity_thread_id_from_node_class_in_pipeline() { + // When a node has classes (from subgraph derivation), thread_id resolves + // from the first class name per spec step 4. + let mut graph = make_graph_with_start_exit("ThreadClassTest"); + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + work.classes = vec!["planning".to_string(), "review".to_string()]; + graph.nodes.insert("work".to_string(), work); + + graph.edges.push(Edge::new("start", "work")); + graph.edges.push(Edge::new("work", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let thread_ids = captures.thread_ids.lock().unwrap(); + assert_eq!(thread_ids[0].0, "work"); + assert_eq!( + thread_ids[0].1, + Some("planning".to_string()), + "thread_id should resolve from first class name" + ); +} + +#[tokio::test] +async fn fidelity_edge_thread_id_override_in_pipeline() { + // Edge thread_id should override the previous-node fallback. + let mut graph = make_graph_with_start_exit("EdgeThreadOverrideTest"); + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + graph.nodes.insert("work".to_string(), work); + + let mut edge_to_work = Edge::new("start", "work"); + edge_to_work.attrs.insert( + "thread_id".to_string(), + AttrValue::String("edge-session".to_string()), + ); + graph.edges.push(edge_to_work); + graph.edges.push(Edge::new("work", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let thread_ids = captures.thread_ids.lock().unwrap(); + assert_eq!(thread_ids[0].0, "work"); + assert_eq!( + thread_ids[0].1, + Some("edge-session".to_string()), + "edge thread_id should override the previous-node fallback" + ); +} + +#[tokio::test] +async fn fidelity_full_without_explicit_thread_id_uses_previous_node() { + // When fidelity=full but no explicit thread_id is set, thread resolution + // should still fall back to the previous node ID. + let mut graph = make_graph_with_start_exit("FullNoExplicitThreadTest"); + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + work.attrs.insert( + "fidelity".to_string(), + AttrValue::String("full".to_string()), + ); + // No thread_id set explicitly + graph.nodes.insert("work".to_string(), work); + + graph.edges.push(Edge::new("start", "work")); + graph.edges.push(Edge::new("work", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let fidelities = captures.fidelities.lock().unwrap(); + assert_eq!(fidelities[0].1, "full"); + + let thread_ids = captures.thread_ids.lock().unwrap(); + assert_eq!(thread_ids[0].0, "work"); + assert_eq!( + thread_ids[0].1, + Some("start".to_string()), + "full fidelity without explicit thread_id should fall back to previous node" + ); + + let preambles = captures.preambles.lock().unwrap(); + assert_eq!(preambles[0].1, "", "full fidelity should produce empty preamble"); +} + +#[tokio::test] +async fn fidelity_from_parsed_dot_pipeline() { + // Parse a DOT file with fidelity attributes and run the pipeline. + let input = r#"digraph FidelityDotTest { + graph [goal="Test DOT fidelity", default_fidelity="truncate"] + + start [shape=Mdiamond] + exit [shape=Msquare] + + step_a [type="fidelity_capture"] + step_b [type="fidelity_capture", fidelity="summary:medium"] + step_c [type="fidelity_capture"] + + start -> step_a -> step_b + step_b -> step_c [fidelity="summary:high"] + step_c -> exit + }"#; + + let graph = parse(input).expect("parsing should succeed"); + validate_or_raise(&graph, &[]).expect("validation should pass"); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let fidelities = captures.fidelities.lock().unwrap(); + // step_a: no node fidelity, no edge fidelity -> graph default "truncate" + assert_eq!(fidelities[0].0, "step_a"); + assert_eq!(fidelities[0].1, "truncate"); + // step_b: node fidelity "summary:medium" overrides graph default + assert_eq!(fidelities[1].0, "step_b"); + assert_eq!(fidelities[1].1, "summary:medium"); + // step_c: node has no fidelity but incoming edge has "summary:high" -> edge wins + assert_eq!(fidelities[2].0, "step_c"); + assert_eq!(fidelities[2].1, "summary:high"); +} + +#[tokio::test] +async fn fidelity_checkpoint_roundtrip_preserves_fidelity() { + // Run a pipeline that sets a specific fidelity, save checkpoint, + // load it, and verify the fidelity value survives the roundtrip. + let mut graph = make_graph_with_start_exit("FidelityCheckpointRoundtripTest"); + graph.attrs.insert( + "default_fidelity".to_string(), + AttrValue::String("summary:high".to_string()), + ); + let work = Node::new("work"); + graph.nodes.insert("work".to_string(), work); + graph.edges.push(Edge::new("start", "work")); + graph.edges.push(Edge::new("work", "exit")); + + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + // Load, save, load again to verify roundtrip + let checkpoint_path = dir.path().join("checkpoint.json"); + let cp1 = Checkpoint::load(&checkpoint_path).expect("first load"); + assert_eq!( + cp1.context_values.get("internal.fidelity"), + Some(&serde_json::json!("summary:high")), + ); + + let roundtrip_path = dir.path().join("checkpoint_roundtrip.json"); + cp1.save(&roundtrip_path).expect("save"); + let cp2 = Checkpoint::load(&roundtrip_path).expect("second load"); + assert_eq!( + cp2.context_values.get("internal.fidelity"), + Some(&serde_json::json!("summary:high")), + "fidelity should survive checkpoint save/load roundtrip" + ); +} + +#[tokio::test] +async fn fidelity_node_thread_id_overrides_edge_thread_id_in_pipeline() { + // When both node and edge have thread_id, the node's takes precedence (spec step 1 > step 2). + let mut graph = make_graph_with_start_exit("NodeOverridesEdgeThreadTest"); + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + work.attrs.insert( + "thread_id".to_string(), + AttrValue::String("node-thread".to_string()), + ); + graph.nodes.insert("work".to_string(), work); + + let mut edge_to_work = Edge::new("start", "work"); + edge_to_work.attrs.insert( + "thread_id".to_string(), + AttrValue::String("edge-thread".to_string()), + ); + graph.edges.push(edge_to_work); + graph.edges.push(Edge::new("work", "exit")); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine.run(&graph, &config).await.expect("run"); + + let thread_ids = captures.thread_ids.lock().unwrap(); + assert_eq!(thread_ids[0].0, "work"); + assert_eq!( + thread_ids[0].1, + Some("node-thread".to_string()), + "node thread_id should take precedence over edge thread_id" + ); +} + +#[tokio::test] +async fn fidelity_resume_preserves_context_values_across_checkpoint() { + // After resuming from a checkpoint, context values from the checkpoint + // should be available to the resumed nodes. This tests that fidelity-related + // context survives the resume path. + let mut graph = make_graph_with_start_exit("FidelityResumeContextTest"); + let mut step_a = Node::new("step_a"); + step_a.attrs.insert( + "fidelity".to_string(), + AttrValue::String("compact".to_string()), + ); + graph.nodes.insert("step_a".to_string(), step_a); + + let mut step_b = Node::new("step_b"); + step_b.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + step_b.attrs.insert( + "fidelity".to_string(), + AttrValue::String("summary:low".to_string()), + ); + graph.nodes.insert("step_b".to_string(), step_b); + + graph.edges.push(Edge::new("start", "step_a")); + graph.edges.push(Edge::new("step_a", "step_b")); + graph.edges.push(Edge::new("step_b", "exit")); + + let ctx = Context::new(); + ctx.set("outcome", serde_json::json!("success")); + ctx.set("internal.fidelity", serde_json::json!("compact")); + ctx.set("context.custom_key", serde_json::json!("custom_value")); + + let mut outcomes = std::collections::HashMap::new(); + outcomes.insert("start".to_string(), Outcome::success()); + outcomes.insert("step_a".to_string(), Outcome::success()); + + let checkpoint = Checkpoint::from_context( + &ctx, + "step_a", + vec!["start".to_string(), "step_a".to_string()], + std::collections::HashMap::new(), + outcomes, + Some("step_b".to_string()), + ); + + let captures = FidelityCaptures::new(); + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("fidelity_capture", Box::new(FidelityCapturingHandler { captures: captures.clone() })); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + engine + .run_from_checkpoint(&graph, &config, &checkpoint) + .await + .expect("resume should succeed"); + + let fidelities = captures.fidelities.lock().unwrap(); + assert_eq!(fidelities[0].0, "step_b"); + assert_eq!( + fidelities[0].1, "summary:low", + "resumed node should use its own fidelity (no degrade since checkpoint was compact, not full)" + ); + + // Verify the final checkpoint still has the fidelity + let final_cp = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap(); + assert_eq!( + final_cp.context_values.get("internal.fidelity"), + Some(&serde_json::json!("summary:low")), + ); +} + // =========================================================================== // 20. Real LLM pipeline tests (requires ANTHROPIC_API_KEY) // =========================================================================== @@ -4086,3 +5506,979 @@ mod real_llm { ); } } + +// --------------------------------------------------------------------------- +// Wait.human freeform edge integration tests (Section 4.6) +// --------------------------------------------------------------------------- + +/// Freeform-only human gate: free-text input routes through the freeform edge +/// and stores the text in human.gate.text context variable. +#[tokio::test] +async fn human_gate_freeform_only_routes_text() { + // Graph: start -> gate -> freeform_target -> exit + // gate has only a freeform edge (no fixed choices) + let mut graph = Graph::new("FreeformOnlyTest"); + + let mut start = Node::new("start"); + start + .attrs + .insert("shape".to_string(), AttrValue::String("Mdiamond".to_string())); + graph.nodes.insert("start".to_string(), start); + + let mut exit = Node::new("exit"); + exit.attrs + .insert("shape".to_string(), AttrValue::String("Msquare".to_string())); + graph.nodes.insert("exit".to_string(), exit); + + let mut gate = Node::new("gate"); + gate.attrs + .insert("shape".to_string(), AttrValue::String("hexagon".to_string())); + gate.attrs.insert( + "type".to_string(), + AttrValue::String("wait.human".to_string()), + ); + gate.attrs.insert( + "label".to_string(), + AttrValue::String("Enter feedback".to_string()), + ); + graph.nodes.insert("gate".to_string(), gate); + graph + .nodes + .insert("freeform_target".to_string(), Node::new("freeform_target")); + + graph.edges.push(Edge::new("start", "gate")); + + let mut freeform_edge = Edge::new("gate", "freeform_target"); + freeform_edge + .attrs + .insert("freeform".to_string(), AttrValue::Boolean(true)); + graph.edges.push(freeform_edge); + + graph.edges.push(Edge::new("freeform_target", "exit")); + + let answers = VecDeque::from([Answer::text("my free text input")]); + let interviewer = Arc::new(QueueInterviewer::new(answers)); + + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("wait.human", Box::new(WaitHumanHandler::new(interviewer))); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + + let outcome = engine.run(&graph, &config).await.expect("run should succeed"); + assert_eq!(outcome.status, StageStatus::Success); + + let checkpoint = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap(); + assert!( + checkpoint + .completed_nodes + .contains(&"freeform_target".to_string()), + "should have routed through freeform_target" + ); + assert_eq!( + checkpoint.context_values.get("human.gate.text"), + Some(&serde_json::json!("my free text input")), + "human.gate.text should contain the freeform input" + ); + assert_eq!( + checkpoint.context_values.get("human.gate.selected"), + Some(&serde_json::json!("freeform")), + "human.gate.selected should be 'freeform'" + ); + assert_eq!( + checkpoint.context_values.get("human.gate.label"), + Some(&serde_json::json!("my free text input")), + "human.gate.label should contain the freeform text" + ); +} + +/// Human gate with both fixed choices and a freeform edge: +/// when the answer matches a fixed choice, it routes to the fixed choice target. +#[tokio::test] +async fn human_gate_freeform_with_fixed_choice_match() { + // Graph: start -> gate -> {approve, reject, freeform_target} -> exit + // gate has fixed choices plus a freeform edge + // Answer selects "A" which matches "Approve" -> routes to approve + let mut graph = Graph::new("FreeformFixedMatchTest"); + + let mut start = Node::new("start"); + start + .attrs + .insert("shape".to_string(), AttrValue::String("Mdiamond".to_string())); + graph.nodes.insert("start".to_string(), start); + + let mut exit = Node::new("exit"); + exit.attrs + .insert("shape".to_string(), AttrValue::String("Msquare".to_string())); + graph.nodes.insert("exit".to_string(), exit); + + let mut gate = Node::new("gate"); + gate.attrs + .insert("shape".to_string(), AttrValue::String("hexagon".to_string())); + gate.attrs.insert( + "type".to_string(), + AttrValue::String("wait.human".to_string()), + ); + gate.attrs.insert( + "label".to_string(), + AttrValue::String("Review Changes".to_string()), + ); + graph.nodes.insert("gate".to_string(), gate); + graph + .nodes + .insert("approve".to_string(), Node::new("approve")); + graph + .nodes + .insert("reject".to_string(), Node::new("reject")); + graph + .nodes + .insert("freeform_target".to_string(), Node::new("freeform_target")); + + graph.edges.push(Edge::new("start", "gate")); + + let mut e_approve = Edge::new("gate", "approve"); + e_approve.attrs.insert( + "label".to_string(), + AttrValue::String("[A] Approve".to_string()), + ); + graph.edges.push(e_approve); + + let mut e_reject = Edge::new("gate", "reject"); + e_reject.attrs.insert( + "label".to_string(), + AttrValue::String("[R] Reject".to_string()), + ); + graph.edges.push(e_reject); + + let mut freeform_edge = Edge::new("gate", "freeform_target"); + freeform_edge + .attrs + .insert("freeform".to_string(), AttrValue::Boolean(true)); + graph.edges.push(freeform_edge); + + graph.edges.push(Edge::new("approve", "exit")); + graph.edges.push(Edge::new("reject", "exit")); + graph.edges.push(Edge::new("freeform_target", "exit")); + + // Answer selects "A" which matches the Approve choice + let answers = VecDeque::from([Answer { + value: AnswerValue::Selected("A".to_string()), + selected_option: None, + text: None, + }]); + let interviewer = Arc::new(QueueInterviewer::new(answers)); + + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("wait.human", Box::new(WaitHumanHandler::new(interviewer))); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + + let outcome = engine.run(&graph, &config).await.expect("run should succeed"); + assert_eq!(outcome.status, StageStatus::Success); + + let checkpoint = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap(); + assert!( + checkpoint.completed_nodes.contains(&"approve".to_string()), + "fixed choice match should route to approve" + ); + assert!( + !checkpoint + .completed_nodes + .contains(&"freeform_target".to_string()), + "should NOT route through freeform when fixed choice matches" + ); +} + +/// Human gate with both fixed choices and a freeform edge: +/// when the answer does NOT match any fixed choice, it falls through to the freeform edge. +#[tokio::test] +async fn human_gate_freeform_fallback_on_unmatched_text() { + // Graph: start -> gate -> {approve, reject, freeform_target} -> exit + // gate has fixed choices plus a freeform edge + // Answer is free text that doesn't match any choice -> routes to freeform_target + let mut graph = Graph::new("FreeformFallbackTest"); + + let mut start = Node::new("start"); + start + .attrs + .insert("shape".to_string(), AttrValue::String("Mdiamond".to_string())); + graph.nodes.insert("start".to_string(), start); + + let mut exit = Node::new("exit"); + exit.attrs + .insert("shape".to_string(), AttrValue::String("Msquare".to_string())); + graph.nodes.insert("exit".to_string(), exit); + + let mut gate = Node::new("gate"); + gate.attrs + .insert("shape".to_string(), AttrValue::String("hexagon".to_string())); + gate.attrs.insert( + "type".to_string(), + AttrValue::String("wait.human".to_string()), + ); + gate.attrs.insert( + "label".to_string(), + AttrValue::String("Review Changes".to_string()), + ); + graph.nodes.insert("gate".to_string(), gate); + graph + .nodes + .insert("approve".to_string(), Node::new("approve")); + graph + .nodes + .insert("reject".to_string(), Node::new("reject")); + graph + .nodes + .insert("freeform_target".to_string(), Node::new("freeform_target")); + + graph.edges.push(Edge::new("start", "gate")); + + let mut e_approve = Edge::new("gate", "approve"); + e_approve.attrs.insert( + "label".to_string(), + AttrValue::String("[A] Approve".to_string()), + ); + graph.edges.push(e_approve); + + let mut e_reject = Edge::new("gate", "reject"); + e_reject.attrs.insert( + "label".to_string(), + AttrValue::String("[R] Reject".to_string()), + ); + graph.edges.push(e_reject); + + let mut freeform_edge = Edge::new("gate", "freeform_target"); + freeform_edge + .attrs + .insert("freeform".to_string(), AttrValue::Boolean(true)); + graph.edges.push(freeform_edge); + + graph.edges.push(Edge::new("approve", "exit")); + graph.edges.push(Edge::new("reject", "exit")); + graph.edges.push(Edge::new("freeform_target", "exit")); + + // Free-text answer that doesn't match any fixed choice + let answers = VecDeque::from([Answer::text("I need more context before deciding")]); + let interviewer = Arc::new(QueueInterviewer::new(answers)); + + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("wait.human", Box::new(WaitHumanHandler::new(interviewer))); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + + let outcome = engine.run(&graph, &config).await.expect("run should succeed"); + assert_eq!(outcome.status, StageStatus::Success); + + let checkpoint = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap(); + assert!( + checkpoint + .completed_nodes + .contains(&"freeform_target".to_string()), + "unmatched text should fall through to freeform_target" + ); + assert!( + !checkpoint.completed_nodes.contains(&"approve".to_string()), + "should NOT route to approve" + ); + assert!( + !checkpoint.completed_nodes.contains(&"reject".to_string()), + "should NOT route to reject" + ); + assert_eq!( + checkpoint.context_values.get("human.gate.text"), + Some(&serde_json::json!("I need more context before deciding")), + "human.gate.text should contain the freeform input" + ); + assert_eq!( + checkpoint.context_values.get("human.gate.selected"), + Some(&serde_json::json!("freeform")), + "human.gate.selected should be 'freeform' for freeform fallback" + ); + assert_eq!( + checkpoint.context_values.get("human.gate.label"), + Some(&serde_json::json!("I need more context before deciding")), + "human.gate.label should contain the freeform text" + ); +} + +/// Verifies that the Question presented to the interviewer has allow_freeform=true +/// when a freeform edge is present on the human gate. +#[tokio::test] +async fn human_gate_freeform_sets_allow_freeform_on_question() { + // Graph: start -> gate -> {approve, freeform_target} -> exit + // gate has a fixed choice plus a freeform edge + // We use RecordingInterviewer to capture the question and verify allow_freeform + let mut graph = Graph::new("AllowFreeformTest"); + + let mut start = Node::new("start"); + start + .attrs + .insert("shape".to_string(), AttrValue::String("Mdiamond".to_string())); + graph.nodes.insert("start".to_string(), start); + + let mut exit = Node::new("exit"); + exit.attrs + .insert("shape".to_string(), AttrValue::String("Msquare".to_string())); + graph.nodes.insert("exit".to_string(), exit); + + let mut gate = Node::new("gate"); + gate.attrs + .insert("shape".to_string(), AttrValue::String("hexagon".to_string())); + gate.attrs.insert( + "type".to_string(), + AttrValue::String("wait.human".to_string()), + ); + gate.attrs.insert( + "label".to_string(), + AttrValue::String("Pick or type".to_string()), + ); + graph.nodes.insert("gate".to_string(), gate); + graph + .nodes + .insert("approve".to_string(), Node::new("approve")); + graph + .nodes + .insert("freeform_target".to_string(), Node::new("freeform_target")); + + graph.edges.push(Edge::new("start", "gate")); + + let mut e_approve = Edge::new("gate", "approve"); + e_approve.attrs.insert( + "label".to_string(), + AttrValue::String("[A] Approve".to_string()), + ); + graph.edges.push(e_approve); + + let mut freeform_edge = Edge::new("gate", "freeform_target"); + freeform_edge + .attrs + .insert("freeform".to_string(), AttrValue::Boolean(true)); + graph.edges.push(freeform_edge); + + graph.edges.push(Edge::new("approve", "exit")); + graph.edges.push(Edge::new("freeform_target", "exit")); + + let answers = VecDeque::from([Answer { + value: AnswerValue::Selected("A".to_string()), + selected_option: None, + text: None, + }]); + let inner = QueueInterviewer::new(answers); + let recorder = Arc::new(RecordingInterviewer::new(Box::new(inner))); + let interviewer: Arc = recorder.clone(); + + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("wait.human", Box::new(WaitHumanHandler::new(interviewer))); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + + let outcome = engine.run(&graph, &config).await.expect("run should succeed"); + assert_eq!(outcome.status, StageStatus::Success); + + let recordings = recorder.recordings(); + assert_eq!(recordings.len(), 1, "should have recorded exactly one question"); + assert!( + recordings[0].0.allow_freeform, + "Question should have allow_freeform=true when a freeform edge is present" + ); +} + +/// Verifies that the Question presented to the interviewer has allow_freeform=false +/// when no freeform edge is present on the human gate (fixed choices only). +#[tokio::test] +async fn human_gate_without_freeform_sets_allow_freeform_false() { + // Graph: start -> gate -> {approve, reject} -> exit + // gate has only fixed choices, no freeform edge + let mut graph = Graph::new("NoFreeformTest"); + + let mut start = Node::new("start"); + start + .attrs + .insert("shape".to_string(), AttrValue::String("Mdiamond".to_string())); + graph.nodes.insert("start".to_string(), start); + + let mut exit = Node::new("exit"); + exit.attrs + .insert("shape".to_string(), AttrValue::String("Msquare".to_string())); + graph.nodes.insert("exit".to_string(), exit); + + let mut gate = Node::new("gate"); + gate.attrs + .insert("shape".to_string(), AttrValue::String("hexagon".to_string())); + gate.attrs.insert( + "type".to_string(), + AttrValue::String("wait.human".to_string()), + ); + gate.attrs.insert( + "label".to_string(), + AttrValue::String("Pick one".to_string()), + ); + graph.nodes.insert("gate".to_string(), gate); + graph + .nodes + .insert("approve".to_string(), Node::new("approve")); + graph + .nodes + .insert("reject".to_string(), Node::new("reject")); + + graph.edges.push(Edge::new("start", "gate")); + + let mut e_approve = Edge::new("gate", "approve"); + e_approve.attrs.insert( + "label".to_string(), + AttrValue::String("[A] Approve".to_string()), + ); + graph.edges.push(e_approve); + + let mut e_reject = Edge::new("gate", "reject"); + e_reject.attrs.insert( + "label".to_string(), + AttrValue::String("[R] Reject".to_string()), + ); + graph.edges.push(e_reject); + + graph.edges.push(Edge::new("approve", "exit")); + graph.edges.push(Edge::new("reject", "exit")); + + let answers = VecDeque::from([Answer { + value: AnswerValue::Selected("A".to_string()), + selected_option: None, + text: None, + }]); + let inner = QueueInterviewer::new(answers); + let recorder = Arc::new(RecordingInterviewer::new(Box::new(inner))); + let interviewer: Arc = recorder.clone(); + + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("wait.human", Box::new(WaitHumanHandler::new(interviewer))); + + let engine = PipelineEngine::new(registry, EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + + let outcome = engine.run(&graph, &config).await.expect("run should succeed"); + assert_eq!(outcome.status, StageStatus::Success); + + let recordings = recorder.recordings(); + assert_eq!(recordings.len(), 1, "should have recorded exactly one question"); + assert!( + !recordings[0].0.allow_freeform, + "Question should have allow_freeform=false when no freeform edge is present" + ); +} + +// --------------------------------------------------------------------------- +// Subgraph features (Section 2.10) +// --------------------------------------------------------------------------- + +#[test] +fn subgraph_node_defaults_scoped_to_subgraph() { + let input = r#"digraph SubgraphDefaults { + graph [goal="Test subgraph defaults"] + start [shape=Mdiamond] + exit [shape=Msquare] + + subgraph cluster_loop { + label = "Loop A" + node [thread_id="loop-a", timeout="900s"] + + plan [label="Plan next step"] + implement [label="Implement", timeout="1800s"] + } + + outside [label="Outside node"] + + start -> plan -> implement -> outside -> exit + }"#; + + let graph = parse(input).expect("parsing should succeed"); + + // Plan inherits both thread_id and timeout from subgraph defaults + let plan = &graph.nodes["plan"]; + assert_eq!(plan.thread_id(), Some("loop-a")); + assert_eq!( + plan.timeout(), + Some(std::time::Duration::from_secs(900)) + ); + + // Implement inherits thread_id but overrides timeout + let implement = &graph.nodes["implement"]; + assert_eq!(implement.thread_id(), Some("loop-a")); + assert_eq!( + implement.timeout(), + Some(std::time::Duration::from_secs(1800)) + ); + + // Outside node should NOT have subgraph defaults + let outside = &graph.nodes["outside"]; + assert_eq!(outside.thread_id(), None); + assert_eq!(outside.timeout(), None); +} + +#[test] +fn subgraph_class_derived_from_label() { + let input = r#"digraph SubgraphClass { + graph [goal="Test class derivation"] + start [shape=Mdiamond] + exit [shape=Msquare] + + subgraph cluster_loop { + label = "Loop A" + plan [label="Plan"] + implement [label="Implement"] + } + + start -> plan -> implement -> exit + }"#; + + let graph = parse(input).expect("parsing should succeed"); + + // Nodes inside subgraph receive derived class "loop-a" + assert!(graph.nodes["plan"].classes.contains(&"loop-a".to_string())); + assert!(graph.nodes["implement"].classes.contains(&"loop-a".to_string())); + + // Nodes outside subgraph do not get the class + assert!(!graph.nodes["start"].classes.contains(&"loop-a".to_string())); + assert!(!graph.nodes["exit"].classes.contains(&"loop-a".to_string())); +} + +#[test] +fn subgraph_class_derivation_strips_special_chars() { + let input = r#"digraph SubgraphClassStrip { + graph [goal="Test class derivation with special chars"] + + subgraph cluster_review { + label = "Code Review!!!" + reviewer [label="Reviewer"] + } + }"#; + + let graph = parse(input).expect("parsing should succeed"); + // "Code Review!!!" -> lowercase "code review!!!" -> spaces to hyphens "code-review!!!" + // -> strip non-alphanumeric except hyphens -> "code-review" + assert!(graph.nodes["reviewer"].classes.contains(&"code-review".to_string())); +} + +#[test] +fn subgraph_scoping_does_not_leak_to_outer_scope() { + let input = r#"digraph SubgraphScoping { + graph [goal="Test scoping"] + node [timeout="300s"] + + subgraph cluster_inner { + label = "Inner" + node [timeout="900s"] + inner_node [label="Inner"] + } + + outer_node [label="Outer"] + }"#; + + let graph = parse(input).expect("parsing should succeed"); + + // Inner node gets the subgraph-scoped timeout of 900s + let inner = &graph.nodes["inner_node"]; + assert_eq!( + inner.timeout(), + Some(std::time::Duration::from_secs(900)) + ); + + // Outer node gets the graph-level default of 300s, not the subgraph's 900s + let outer = &graph.nodes["outer_node"]; + assert_eq!( + outer.timeout(), + Some(std::time::Duration::from_secs(300)) + ); +} + +#[test] +fn subgraph_global_defaults_plus_subgraph_defaults() { + let input = r#"digraph SubgraphMerge { + graph [goal="Test merged defaults"] + node [shape=box, timeout="300s"] + + subgraph cluster_loop { + label = "Loop" + node [thread_id="loop-thread"] + step [label="Step"] + } + + plain [label="Plain"] + }"#; + + let graph = parse(input).expect("parsing should succeed"); + + // Step should have both the global shape=box + timeout=300s and subgraph thread_id + let step = &graph.nodes["step"]; + assert_eq!(step.shape(), "box"); + assert_eq!(step.thread_id(), Some("loop-thread")); + assert_eq!( + step.timeout(), + Some(std::time::Duration::from_secs(300)) + ); + + // Plain should have the global defaults but no thread_id + let plain = &graph.nodes["plain"]; + assert_eq!(plain.shape(), "box"); + assert_eq!(plain.thread_id(), None); + assert_eq!( + plain.timeout(), + Some(std::time::Duration::from_secs(300)) + ); +} + +#[test] +fn subgraph_edges_inherit_class() { + let input = r#"digraph SubgraphEdgeClass { + graph [goal="Test edge nodes get class"] + + subgraph cluster_loop { + label = "My Loop" + a [label="A"] + b [label="B"] + a -> b + } + }"#; + + let graph = parse(input).expect("parsing should succeed"); + + // Both nodes referenced in edges within the subgraph get the derived class + assert!(graph.nodes["a"].classes.contains(&"my-loop".to_string())); + assert!(graph.nodes["b"].classes.contains(&"my-loop".to_string())); +} + +#[test] +fn subgraph_without_label_no_class_derived() { + let input = r#"digraph SubgraphNoLabel { + graph [goal="Test subgraph without label"] + + subgraph cluster_unnamed { + node [timeout="600s"] + worker [label="Worker"] + } + }"#; + + let graph = parse(input).expect("parsing should succeed"); + + // No label means no class should be derived + let worker = &graph.nodes["worker"]; + assert!(worker.classes.is_empty()); + // But the default should still apply + assert_eq!( + worker.timeout(), + Some(std::time::Duration::from_secs(600)) + ); +} + +// --------------------------------------------------------------------------- +// Tool Call Hooks (Section 9.7) +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn tool_hooks_pre_success_allows_pipeline_to_proceed() { + let input = r#"digraph HookTest { + graph [goal="Test pre-hook success"] + start [shape=Mdiamond] + exit [shape=Msquare] + work [shape=box, label="Work", prompt="Do work", tool_hooks.pre="exit 0"] + start -> work -> exit + }"#; + + let graph = parse(input).expect("parse should succeed"); + validate_or_raise(&graph, &[]).expect("validation should pass"); + + let dir = tempfile::tempdir().unwrap(); + let engine = PipelineEngine::new(make_linear_registry(), EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + + let outcome = engine.run(&graph, &config).await.expect("run should succeed"); + assert_eq!(outcome.status, StageStatus::Success); + + // The work node should have executed normally + let stage_dir = dir.path().join("work"); + assert!( + stage_dir.join("prompt.md").exists(), + "prompt.md should exist when pre-hook succeeds" + ); + assert!( + stage_dir.join("response.md").exists(), + "response.md should exist when pre-hook succeeds" + ); +} + +#[tokio::test] +async fn tool_hooks_pre_failure_skips_tool_call() { + let input = r#"digraph HookTest { + graph [goal="Test pre-hook failure"] + start [shape=Mdiamond] + exit [shape=Msquare] + work [shape=box, label="Work", prompt="Do work", tool_hooks.pre="exit 1"] + start -> work -> exit + }"#; + + let graph = parse(input).expect("parse should succeed"); + validate_or_raise(&graph, &[]).expect("validation should pass"); + + let dir = tempfile::tempdir().unwrap(); + let engine = PipelineEngine::new(make_linear_registry(), EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + + engine.run(&graph, &config).await.expect("run should complete"); + + // The pipeline should still complete (skipped is not a fatal status), + // but the work node's handler returns Skipped when pre-hook fails. + let checkpoint = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap(); + assert!( + checkpoint.completed_nodes.contains(&"work".to_string()), + "work should appear in completed_nodes even when skipped" + ); + + // response.md should NOT exist because the LLM call was skipped + let stage_dir = dir.path().join("work"); + assert!( + !stage_dir.join("response.md").exists(), + "response.md should not exist when pre-hook skips tool call" + ); +} + +#[tokio::test] +async fn tool_hooks_post_success_does_not_affect_outcome() { + let input = r#"digraph HookTest { + graph [goal="Test post-hook success"] + start [shape=Mdiamond] + exit [shape=Msquare] + work [shape=box, label="Work", prompt="Do work", tool_hooks.post="exit 0"] + start -> work -> exit + }"#; + + let graph = parse(input).expect("parse should succeed"); + validate_or_raise(&graph, &[]).expect("validation should pass"); + + let dir = tempfile::tempdir().unwrap(); + let engine = PipelineEngine::new(make_linear_registry(), EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + + let outcome = engine.run(&graph, &config).await.expect("run should succeed"); + assert_eq!(outcome.status, StageStatus::Success); + + let stage_dir = dir.path().join("work"); + assert!( + stage_dir.join("response.md").exists(), + "response.md should exist when post-hook succeeds" + ); +} + +#[tokio::test] +async fn tool_hooks_post_failure_does_not_block_pipeline() { + let input = r#"digraph HookTest { + graph [goal="Test post-hook failure"] + start [shape=Mdiamond] + exit [shape=Msquare] + work [shape=box, label="Work", prompt="Do work", tool_hooks.post="exit 1"] + start -> work -> exit + }"#; + + let graph = parse(input).expect("parse should succeed"); + validate_or_raise(&graph, &[]).expect("validation should pass"); + + let dir = tempfile::tempdir().unwrap(); + let engine = PipelineEngine::new(make_linear_registry(), EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + + let outcome = engine.run(&graph, &config).await.expect("run should succeed"); + // Post-hook failure should not block the pipeline (spec 9.7) + assert_eq!(outcome.status, StageStatus::Success); + + let stage_dir = dir.path().join("work"); + assert!( + stage_dir.join("response.md").exists(), + "response.md should exist even when post-hook fails" + ); +} + +#[tokio::test] +async fn tool_hooks_graph_level_applies_to_all_nodes() { + let input = r#"digraph HookTest { + graph [goal="Test graph-level hooks", tool_hooks.pre="exit 0"] + start [shape=Mdiamond] + exit [shape=Msquare] + step1 [shape=box, label="Step1", prompt="First step"] + step2 [shape=box, label="Step2", prompt="Second step"] + start -> step1 -> step2 -> exit + }"#; + + let graph = parse(input).expect("parse should succeed"); + validate_or_raise(&graph, &[]).expect("validation should pass"); + + let dir = tempfile::tempdir().unwrap(); + let engine = PipelineEngine::new(make_linear_registry(), EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + + let outcome = engine.run(&graph, &config).await.expect("run should succeed"); + assert_eq!(outcome.status, StageStatus::Success); + + // Both steps should have executed since graph-level pre-hook exits 0 + assert!( + dir.path().join("step1").join("response.md").exists(), + "step1 should execute with graph-level pre-hook success" + ); + assert!( + dir.path().join("step2").join("response.md").exists(), + "step2 should execute with graph-level pre-hook success" + ); +} + +#[tokio::test] +async fn tool_hooks_node_level_overrides_graph_level() { + let input = r#"digraph HookTest { + graph [goal="Test node override", tool_hooks.pre="exit 0"] + start [shape=Mdiamond] + exit [shape=Msquare] + step1 [shape=box, label="Step1", prompt="First step", tool_hooks.pre="exit 1"] + step2 [shape=box, label="Step2", prompt="Second step"] + start -> step1 -> step2 -> exit + }"#; + + let graph = parse(input).expect("parse should succeed"); + validate_or_raise(&graph, &[]).expect("validation should pass"); + + let dir = tempfile::tempdir().unwrap(); + let engine = PipelineEngine::new(make_linear_registry(), EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + + let _outcome = engine.run(&graph, &config).await.expect("run should complete"); + + // step1 has node-level pre-hook "exit 1" which overrides graph-level "exit 0" + // So step1's tool call should be skipped (no response.md) + assert!( + !dir.path().join("step1").join("response.md").exists(), + "step1 should be skipped because node-level pre-hook overrides graph-level" + ); + + // step2 inherits graph-level "exit 0", so it should execute normally + assert!( + dir.path().join("step2").join("response.md").exists(), + "step2 should execute with inherited graph-level pre-hook" + ); +} + +#[tokio::test] +async fn tool_hooks_pre_receives_node_id_env_var() { + // Use a pre-hook that writes the ATTRACTOR_NODE_ID env var to a file + let dir = tempfile::tempdir().unwrap(); + let marker_path = dir.path().join("node_id.txt"); + let hook_cmd = format!( + "echo $ATTRACTOR_NODE_ID > {}", + marker_path.display() + ); + + let input = format!( + r#"digraph HookTest {{ + graph [goal="Test env vars"] + start [shape=Mdiamond] + exit [shape=Msquare] + my_step [shape=box, label="MyStep", prompt="Do work", tool_hooks.pre="{}"] + start -> my_step -> exit + }}"#, + hook_cmd + ); + + let graph = parse(&input).expect("parse should succeed"); + validate_or_raise(&graph, &[]).expect("validation should pass"); + + let engine = PipelineEngine::new(make_linear_registry(), EventEmitter::new()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + }; + + engine.run(&graph, &config).await.expect("run should succeed"); + + let written = std::fs::read_to_string(&marker_path) + .expect("marker file should exist"); + assert_eq!( + written.trim(), + "my_step", + "ATTRACTOR_NODE_ID should contain the node id" + ); +} + +#[test] +fn parse_tool_hooks_from_dot_syntax() { + let input = r#"digraph HookTest { + graph [goal="Test parsing", tool_hooks.pre="echo pre", tool_hooks.post="echo post"] + start [shape=Mdiamond] + exit [shape=Msquare] + work [shape=box, label="Work", prompt="Do it", tool_hooks.pre="node pre"] + start -> work -> exit + }"#; + + let graph = parse(input).expect("parse should succeed"); + + // Graph-level hooks + assert_eq!( + graph.attrs.get("tool_hooks.pre").and_then(|v| v.as_str()), + Some("echo pre") + ); + assert_eq!( + graph.attrs.get("tool_hooks.post").and_then(|v| v.as_str()), + Some("echo post") + ); + + // Node-level hook overrides + let work = &graph.nodes["work"]; + assert_eq!( + work.attrs.get("tool_hooks.pre").and_then(|v| v.as_str()), + Some("node pre") + ); +} diff --git a/crates/llm/src/bin/ullm.rs b/crates/llm/src/bin/ullm.rs index 152d5a62a..3b68ba2a4 100644 --- a/crates/llm/src/bin/ullm.rs +++ b/crates/llm/src/bin/ullm.rs @@ -367,7 +367,9 @@ mod tests { fn prompt_reads_from_stdin() { // This test verifies stdin is read, but will fail at the API call stage // since no API key is set. The error should NOT be "no prompt provided". + // env_clear() prevents .env loading from triggering real API calls. let result = ullm() + .env_clear() .args(["prompt", "--no-stream", "-m", "test-model"]) .write_stdin("hello from stdin") .assert() @@ -381,7 +383,9 @@ mod tests { #[test] fn prompt_concatenates_stdin_and_arg() { // Same as above — verifies it doesn't error on "no prompt" + // env_clear() prevents .env loading from triggering real API calls. let result = ullm() + .env_clear() .args(["prompt", "--no-stream", "-m", "test-model", "summarize this"]) .write_stdin("some input text") .assert()